Python Action Guide

MaaNTE Python CustomAction 编写指南

架构定位

Python CustomAction 处理 Pipeline JSON 无法覆盖的复杂逻辑（图像算法、状态机、音频检测、MIDI 播放等）。流程控制由 Pipeline 负责，Python 只处理难点。 一句话：Pipeline 管流程，Python 管难点。

没有必要的 Python 逻辑会大大增加代码复杂度，造成下一位开发者开发调试极其困难。

所有坐标与图像以 720p (1280×720) 为基准。

目录结构

agent/
├── main.py                         # 入口：venv、依赖安装、AgentServer 启动
├── custom/
│   ├── __init__.py                 # from .action import *
│   └── action/
│       ├── __init__.py             # 所有 CustomAction 的 import 与 __all__ 注册
│       ├── Common/                 # 通用工具（click、alt_click、logger、utils）
│       │   ├── __init__.py
│       │   ├── click.py
│       │   ├── alt_click.py
│       │   ├── logger.py
│       │   └── utils.py            # get_image、click_rect、match_template_in_region
│       ├── AutoFish/               # 钓鱼业务（按功能聚合）
│       ├── Tetris/                 # 俄罗斯方块业务
│       │   ├── feats/              # 功能实现
│       │   └── utils/              # 工具类（棋盘、场景检测）
│       ├── rhythm/                 # 节奏任务
│       │   ├── feats/
│       │   └── utils/
│       ├── SoundTrigger/           # 音频触发业务
│       ├── Movement/               # 移动控制
│       ├── auto_piano/             # 自动钢琴
│       ├── pinkpaw/                # 粉爪大劫案
│       ├── DatasetCollection/      # 数据集采集
│       ├── auto_make_coffee.py     # 单文件简单动作
│       ├── auto_tetris.py          # 俄罗斯方块入口
│       ├── auto_f_scroll.py        # 自动滚屏
│       ├── furniture_claim.py      # 家具收取
│       ├── realtime_task.py        # 实时任务调度
│       ├── withdraw_money_choose_item.py  # 补货取钱物品选择
│       └── __init__.py             # 所有 CustomAction 的导入与注册
└── utils/
    ├── __init__.py                 # 子模块聚合导出
    ├── logger.py                   # 日志系统（loguru/logging）
    ├── pienv.py                    # PI 环境变量
    ├── maafocus.py                 # 用户可见消息（Pipeline focus 协议）
    ├── i18n.py                     # 多语言/翻译
    ├── screen.py                   # 屏幕缩放/分辨率
    ├── time.py                     # 时间工具
    └── win32_process.py            # Win32 窗口查找

注册机制

装饰器注册

每个 CustomAction 类使用 @AgentServer.custom_action("name") 注册，名称必须与 Pipeline JSON 中 custom_action 的值一致：

from maa.agent.agent_server import AgentServer
from maa.custom_action import CustomAction
from maa.context import Context

@AgentServer.custom_action("auto_make_coffee")
class AutoMakeCoffee(CustomAction):
    def run(
        self, context: Context, argv: CustomAction.RunArg
    ) -> CustomAction.RunResult:
        ...

Pipeline JSON 中对应：

{
    "action": {
        "type": "Custom",
        "param": {
            "custom_action": "auto_make_coffee",
            "custom_action_param": { "count": 10 }
        }
    }
}

__init__.py 注册

所有 CustomAction 必须在 agent/custom/action/__init__.py 中导入并加入 __all__：

from .auto_make_coffee import *
from .Tetris import *

__all__ = [
    "AutoMakeCoffee",
    "AutoTetris",
    "TetrisResetContext",
    "TetrisCheckVitalityAction",
]

遗漏导入 = 动作不生效。

命名

• 类名：PascalCase，语义化。如 AutoMakeCoffee、TetrisGamePlayer、SoundDodgeAction。
• 注册名：snake_case，与 Pipeline JSON 一致。如 "auto_make_coffee"、"auto_tetris"。
• 文件名：snake_case。单文件简单动作直接用 .py 文件；复杂模块用目录 + __init__.py。
• 内部函数/变量：snake_case，模块内部使用的加前导下划线 _。

Common 内置动作

以下 Custom Action 位于 agent/custom/action/Common/，可在 Pipeline 中直接使用，无需额外编写 Python 代码。

click_override

自定义点击。通过 custom_action_param 指定目标 rect，或使用当前识别结果的 box。

• 注册名：click_override
• 参数 custom_action_param：{ "target": [x, y, w, h] }
• 若未提供 custom_action_param，则使用 argv.box（识别结果 box）

{
    "action": {
        "type": "Custom",
        "param": {
            "custom_action": "click_override",
            "custom_action_param": { "target": [100, 200, 50, 50] }
        }
    }
}

alt_click

Alt + 点击。先按下 Alt 键，再点击识别结果 box 位置，最后松开 Alt。

• 注册名：alt_click
• 无需额外参数，点击位置由识别结果的 box 决定

{
    "recognition": { "type": "TemplateMatch", "param": { "template": "xxx.png" } },
    "action": {
        "type": "Custom",
        "param": { "custom_action": "alt_click" }
    }
}

Common 工具函数

agent/custom/action/Common/utils.py 提供常用辅助函数：

from Common.utils import get_image, click_rect, match_template_in_region

img = get_image(controller)
hit, score, x, y = match_template_in_region(img, [0, 0, 1280, 720], template, 0.8)
if hit:
    click_rect(controller, [x, y, 50, 50])

CustomAction 模板

简单动作

import json
from maa.agent.agent_server import AgentServer
from maa.custom_action import CustomAction
from maa.context import Context
from utils.logger import logger
from utils.maafocus import PrintT

@AgentServer.custom_action("my_action")
class MyAction(CustomAction):
    def run(
        self, context: Context, argv: CustomAction.RunArg
    ) -> CustomAction.RunResult:
        PrintT(context, "my_action.started")

        params = {}
        if argv.custom_action_param:
            try:
                params = json.loads(argv.custom_action_param)
            except (json.JSONDecodeError, TypeError):
                pass

        controller = context.tasker.controller

        # 截图
        controller.post_screencap().wait()
        img = controller.cached_image

        # 检查停止信号
        if context.tasker.stopping:
            return CustomAction.RunResult(success=False)

        # 业务逻辑 ...
        logger.debug("识别结果: score=%.2f", score)

        return CustomAction.RunResult(success=True)

带状态管理的动作

对于需要跨多次调用保持状态的场景（如连打计数），使用模块级全局变量，并通过独立的 reset action 重置：

_round_count = 0
_task_config = {}

@AgentServer.custom_action("my_reset")
class MyResetContext(CustomAction):
    def run(self, context, argv) -> CustomAction.RunResult:
        global _round_count, _task_config
        _round_count = 0
        _task_config = {}
        return CustomAction.RunResult(success=True)

@AgentServer.custom_action("my_action")
class MyAction(CustomAction):
    def run(self, context, argv) -> CustomAction.RunResult:
        global _round_count
        _round_count += 1
        ...

Controller API

通过 context.tasker.controller 访问，常用操作：

controller = context.tasker.controller

# 截图
controller.post_screencap().wait()
img = controller.cached_image  # numpy array (BGR or BGRA)

# 点击（绝对坐标，720p 基准）
controller.post_touch_down(x, y).wait()
time.sleep(0.005)  # 短暂按下
controller.post_touch_up().wait()

# 按键（虚拟键码）
controller.post_key_down(0x46).wait()  # F 键
time.sleep(0.1)
controller.post_key_up(0x46).wait()

# 组合键：先 Alt 再点
controller.post_key_down(0x12).wait()  # VK_MENU (Alt)
# ... 执行点击 ...
controller.post_key_up(0x12).wait()

常用虚拟键码：

Pipeline 集成

context.run_recognition

调用已定义的 Pipeline 识别节点：

# 基本调用
result = context.run_recognition("FurnitureOcrRec", image)

# 带参数覆盖（动态 expected 文本、自定义 roi 等）
result = context.run_recognition(
    "FurnitureOcrRec",
    image,
    pipeline_override={
        "FurnitureOcrRec": {
            "recognition": {"param": {"expected": "仓鼠球"}}
        }
    },
)
if result and result.box:
    x, y, w, h = result.box.x, result.box.y, result.box.w, result.box.h

RecognitionDetail 结构

context.run_recognition() 返回 RecognitionDetail（maa.define.RecognitionDetail），结构如下：

@dataclass
class RecognitionDetail:
    reco_id: int
    name: str
    algorithm: Union[AlgorithmEnum, str]
    hit: bool                              # 是否有命中
    box: Optional[Rect]                    # 最佳命中的 box
    all_results: List[RecognitionResult]   # 全部原始结果
    filtered_results: List[RecognitionResult]  # 过滤后的结果（常用）
    best_result: Optional[RecognitionResult]   # 最佳结果
    raw_detail: Dict
    raw_image: numpy.ndarray
    draw_images: List[numpy.ndarray]

关键方法/属性：

result = context.run_recognition("MyNode", image)

# 是否存在命中
if result and result.hit:
    # 遍历所有命中（正确的多结果方式，不是 .all）
    for r in result.filtered_results:
        # r.box  — Rect(x, y, w, h)
        # 根据算法类型，r 还有不同属性：
        #   OCRResult:        r.text (str), r.score (float)
        #   ColorMatchResult: r.count (int)
        #   TemplateMatchResult: r.score (float)

# result.box — 最佳命中的 Rect（快捷方式）
# result.box.x, result.box.y, result.box.w, result.box.h 都是 int

多结果模式

点击所有命中 rect（如 ColorMatch connected 组件、OCR 多匹配）：

def _filtered_boxes(result):
    """返回 filtered_results 中所有命中的 Rect 列表"""
    if result is None or not result.hit:
        return []
    return [r.box for r in result.filtered_results if r.box is not None]

result = context.run_recognition("SomeColorMatch", image)
for box in _filtered_boxes(result):
    cx, cy = box.x + box.w // 2, box.y + box.h // 2
    controller.post_touch_down(cx, cy).wait()
    time.sleep(0.001)
    controller.post_touch_up().wait()

遍历 OCR 多结果并提取文本：

result = context.run_recognition("SomeOCRNode", image)
for r in result.filtered_results if result else []:
    text = getattr(r, "text", "")           # OCRResult 有 .text，其他类型无
    box = r.box                              # Rect
    # 处理 text 和 box ...

常见错误：

• result.all → 不存在，应该用 result.filtered_results
• box.text → Rect 没有 .text，应该用 getattr(r, "text", "") 在结果对象上取
• ColorMatchResult.box.text → ColorMatchResult 只有 .box（Rect）和 .count（int），无 .text

box 类型注意：MaaFramework Python 绑定返回的 .box 实际是 list [x,y,w,h] 而非 Rect 对象（虽然类型标注为 Rect），转换时需兼容：

def _box_to_rect(box):
    if isinstance(box, (list, tuple)):
        return list(box)
    return [box.x, box.y, box.w, box.h]

真实 raw_detail 结构参考（ColorMatch）：

{
  "reco_id": 400000008,
  "algorithm": "ColorMatch",
  "box": [256, 384, 19, 18],
  "detail": {
    "all": [
      {"box": [256, 384, 19, 18], "count": 185},
      {"box": [256, 531, 19, 17], "count": 182}
    ],
    "best": {"box": [256, 384, 19, 18], "count": 185},
    "filtered": [
      {"box": [256, 384, 19, 18], "count": 185},
      {"box": [256, 531, 19, 17], "count": 182}
    ]
  }
}

• result.filtered_results[i].box → list [x,y,w,h]
• result.filtered_results[i].count → int（ColorMatch）/ result.filtered_results[i].text → str（OCR）
• result.raw_detail 包含完整 JSON-serializable dict，可作为兜底访问

context.run_recognition_direct

在 Python 中直接调用 MaaFramework 识别算法（无需 Pipeline 节点定义）：

from maa.pipeline import JRecognitionType, JOCR

detail = context.run_recognition_direct(
    JRecognitionType.OCR, JOCR(roi=[x, y, w, h]), frame
)
if detail and detail.hit and detail.best_result:
    text = detail.best_result.text

用户可见消息（maafocus）

所有面向用户的进度、状态、错误提示必须使用 maafocus.Print() / PrintT()，禁止 print() 和 logger.info()。

两个通道职责分离：

PrintT — i18n 用户消息（推荐）

from utils.maafocus import PrintT

PrintT(context, "tetris.task_started", " | ".join(parts))
PrintT(context, "coffee.making", count + 1, make_count)
PrintT(context, "rhythm.playing_started", fps, "ON", scene_lock)

首个参数是 i18n key（定义在 assets/resource/locales/agent/zh_cn.json），后续参数为 %-格式化值。

Print — 原样文本

from utils.maafocus import Print

Print(context, "纯文本消息，不做翻译")

模块依赖

maafocus.py 内部导入 utils.i18n，因此需确保 i18n_init() 已先执行（main.py 中已有）。

辅助类中无 context 的情况

辅助类（如 _KeyScheduler、MaaKeyboardBridge）不持有 context，内部只用 logger.debug() 记录调试信息。用户消息由调用方（CustomAction.run()）发送。

管道 focus 消息（JSON 侧）

Pipeline JSON 中的 "focus" 字段是纯 pipeline 层消息，不走 Python i18n：

"focus": { "Node.Action.Succeeded": "钓到鱼了！" }

其 i18n 依赖 MXU 侧支持，与 agent 的 T() 系统不互通。

context.run_action

调用 Pipeline 中定义的动作节点：

context.run_action("SomeClickNode")

带参数覆盖：

context.run_action("SomeClickNode", pipeline_override={
    "SomeClickNode": {"target": [100, 200]},
})

context.run_task

启动子任务链：

context.run_task("FurnitureClaim", pipeline_override={...})

context.override_next

动态修改节点的 next 列表，用于运行时流程控制：

context.override_next("RhythmRepeatCheck", ["RhythmExit"])

错误处理

• run() 返回 CustomAction.RunResult(success=True/False) 表示动作成功/失败。
• 失败时 MaaFramework 会走 Pipeline 中该节点的 on_error 分支。
• 异常应捕获并记录，避免静默吞掉：

try:
    params = json.loads(argv.custom_action_param)
except (json.JSONDecodeError, ValueError, TypeError) as e:
    logger.warning("参数解析失败: %r, error: %s", argv.custom_action_param, e)
    return CustomAction.RunResult(success=False)

• 停止信号检查：在长循环中定期检查 context.tasker.stopping，及时退出：

while True:
    if context.tasker.stopping:
        return CustomAction.RunResult(success=False)
    # ... 循环体 ...

日志

logger 用于开发者调试，用户消息走 maafocus。 详见 maa-logging。

from utils.logger import logger

logger.debug("识别结果: %s", result)        # 内部细节
logger.warning("模板缺失，功能降级")         # 开发者关注的异常
logger.error("不可恢复错误: %s", e)          # 严重错误

禁止 print()。MXU 模式下 logger 的 console_level 已降为 "WARNING"，因此 logger.info() 不再出现在用户面前。

资源路径

模板图片等资源文件放在 assets/resource/base/image/ 下。在代码中引用：

from pathlib import Path

base = Path(__file__).parents[4] / "assets" / "resource" / "base"
if not base.exists():
    base = Path(__file__).parents[4] / "resource" / "base"

image_dir = base / "sounds" / "dodge.wav"

开发模式下（interface.json 版本为 DEBUG），cwd 切换到 assets/，使用 resource/base/ 路径。

坐标系统

• 所有坐标以 1280×720 为基准。
• controller.post_screencap() 返回的图片可能不是 720p（取决于实际窗口分辨率），但坐标逻辑仍按 720p 编写。
• Pipeline 中 roi、target 等同样基于 720p。

审查清单

• 注册名与 Pipeline custom_action 值一致
• 已在 agent/custom/action/__init__.py 中导入并加入 __all__
• 使用 utils.logger，无 print() 调用
• 日志使用 % 风格占位符，不拼接字符串
• 无大规模流程代码——流程由 Pipeline 驱动
• 坐标/图像基于 720p
• 长循环中有 context.tasker.stopping 检查
• 异常有合理的错误处理和返回值
• 模块级状态变量有对应的 reset action 或清理逻辑
• 重复逻辑已考虑抽取为 Common/utils.py 中的共用函数
• 配套文件齐全：Python 文件在 agent/custom/action/ 下、已在 __init__.py 中导入并加入 __all__

参考

• Pipeline 协议规范：MaaFramework PipelineProtocol
• 日志规范：maa-logging skill
• Pipeline 编写：pipeline-guide skill
• 项目注册示例：agent/custom/action/__init__.py
• Python binding：maa.custom_action、maa.context 的 MaaFramework Python 绑定