Custom 编写指南

596 字约 2 分钟

模块类型

MaaKEDR 支持三类自定义模块，通过 @AgentServer 装饰器注册。

自定义识别

当 TemplateMatch 和 OCR 无法满足需求时（如动态 ROI、条件判断），使用 Custom Recognition。

from maa.agent.agent_server import AgentServer
from maa.custom_recognition import CustomRecognition
from maa.context import Context
from maa.pipeline import JOCR, JRecognitionType

@AgentServer.custom_recognition("MyRecognizer")
class MyRecognizer(CustomRecognition):
    def analyze(
        self, context: Context, argv: CustomRecognition.AnalyzeArg
    ) -> CustomRecognition.AnalyzeResult | None:
        params = json.loads(argv.custom_recognition_param)
        # OCR 识别
        detail = context.run_recognition_direct(
            JRecognitionType.OCR,
            JOCR(expected=["text"], roi=(x, y, w, h)),
            argv.image,
        )
        if not detail or not detail.box:
            return None  # 未命中
        return CustomRecognition.AnalyzeResult(
            box=detail.box,
            detail={"status": "found"}
        )

Pipeline 中调用：

"NodeName": {
    "recognition": "Custom",
    "custom_recognition": "MyRecognizer",
    "custom_recognition_param": "{\"key\": \"value\"}",
    "action": { "type": "Click" }
}

custom_recognition_param 是 JSON 字符串（需序列化），不是 JSON 对象。

自定义动作

复杂操作（状态管理、条件逻辑）用 Custom Action。

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

@AgentServer.custom_action("MyAction")
class MyAction(CustomAction):
    def run(
        self, context: Context, argv: CustomAction.RunArg
    ) -> CustomAction.RunResult:
        params = json.loads(argv.custom_action_param)
        target = params.get("target_count", 1)
        # ... 业务逻辑
        return CustomAction.RunResult(success=True)

Pipeline 中调用：

"NodeName": {
    "recognition": "DirectHit",
    "action": {
        "type": "Custom",
        "param": {
            "custom_action": "MyAction",
            "custom_action_param": { "key": "value" }
        }
    }
}

custom_action_param 是 JSON 对象（直接传，不序列化）。

自定义 Sink（事件监听器）

Sink 用于监听任务事件（开始、完成、错误等），适合做前置检查、日志记录或性能监控。

from maa.agent.agent_server import AgentServer
from maa.event import TaskerEventSink, TaskerEvent

@AgentServer.tasker_sink()
class MySink(TaskerEventSink):
    def on_event(self, event: TaskerEvent) -> None:
        if event.event_type == TaskerEvent.Type.RECOGNITION_FAILED:
            node_name = event.detail.get("node_name", "")
            logger.warning(f"节点识别失败: {node_name}")

识别结果处理

Custom Recognition 的 analyze 返回 AnalyzeResult 或 None：

返回 AnalyzeResult(box=..., detail=...)：命中，使用指定 box
返回 None：未命中，框架走 on_error

返回值中的 detail 会记录在日志中，可用于调试。

Context API 参考

# OCR 识别
ocr = context.run_recognition_direct(
    JRecognitionType.OCR,
    JOCR(expected=["text"], roi=(x, y, w, h)),
    image,
)
# 获取结果文本
if ocr and ocr.all_results:
    text = ocr.all_results[0].text

# 模板匹配
match = context.run_recognition_direct(
    JRecognitionType.TemplateMatch,
    JTemplateMatch(template="path.png", roi=(x, y, w, h), threshold=0.8),
    image,
)

# 点击
context.run_action_direct(JActionType.Click, JClick(), box, "")

# 获取缓存截图
image = context.tasker.controller.cached_image

# 发送点击（跳过 pipeline）
context.tasker.controller.post_click(x, y).wait()

# 覆盖 next 跳转
context.override_next(argv.node_name, ["NextNodeA", "NextNodeB"])

# 动态覆盖 pipeline 配置
context.override_pipeline({
    "SomeNode": {
        "next": ["CustomNext"]
    }
})

注册模块

在 agent/custom/recognition/ 或 agent/custom/action/ 下创建 Python 文件
添加 @AgentServer.custom_recognition("Name") / @AgentServer.custom_action("Name") / @AgentServer.tasker_sink() 装饰器
在 agent/custom/recognition/__init__.py 的 RECOGNITION_MODULES 中注册模块名
Pipeline 中通过 custom_recognition / custom_action 字段引用

开发建议

先阅读项目已有的 Custom 实现（farm_resources.py、pvp.py）了解模式
复杂逻辑先在单独的 Python 文件中测试，再集成到 Pipeline 中
使用 from utils.logger import logger 输出日志，方便调试