Custom Module Writing Guide
About 716 wordsAbout 2 min
Development Approach
M9A is based on MaaFramework v5.1+ and adopts the JSON + Custom Logic Extension approach, registering custom modules through AgentServer.
Custom Module Types
M9A's custom modules are divided into three categories:
1. Custom Action
Location: agent/custom/action/
Purpose: Implement complex game operation logic, such as recognition result processing, multi-step operations, data analysis, etc.
Basic Structure:
from maa.agent.agent_server import AgentServer
from maa.custom_action import CustomAction
from maa.context import Context
@AgentServer.custom_action("YourActionName")
class YourAction(CustomAction):
def run(
self,
context: Context,
argv: CustomAction.RunArg,
) -> CustomAction.RunResult:
# Read passed parameters (optional)
# param = json.loads(argv.custom_action_param)
# Implement your logic
return CustomAction.RunResult(success=True)The method to call the above action in the pipeline is as follows. The passed parameters can be any JSON object, which will be loaded as a dict type in the code above. If you need to change the input parameters based on user input, you can use the pipeline_override operation of the interface (see Interface V2 Protocol for details).
{
"YourNodeName":{
"action":"Custom",
"custom_action":"YourActionName",
"custom_action_param": {
// Pass parameters (optional)
// object
}
// ...
},
// or
"YourNodeName":{
"action": {
"type": "Custom",
"param": {
"custom_action": "YourActionName",
"custom_action_param": {
// Pass parameters (optional)
// object
}
}
},
// ...
},
// ...
}Project Examples:
Screenshot- Save screenshot after task timeoutDisableNode- Set specific node to disabled stateNodeOverride- Dynamically override Pipeline configuration in nodeResetCount- Reset counter state
2. Custom Recognition
Location: agent/custom/reco/
Purpose: Implement complex recognition logic that cannot be accomplished with Pipeline JSON alone, such as dynamic OCR, color matching, multi-step recognition, etc.
Basic Structure:
from maa.agent.agent_server import AgentServer
from maa.custom_recognition import CustomRecognition
from maa.context import Context
from typing import Union, Optional
from maa.define import RectType
@AgentServer.custom_recognition("YourRecognitionName")
class YourRecognition(CustomRecognition):
def analyze(
self,
context: Context,
argv: CustomRecognition.AnalyzeArg,
) -> Union[CustomRecognition.AnalyzeResult, Optional[RectType]]:
# Read passed parameters (optional)
# param = json.loads(argv.custom_recognition_param)
# Implement recognition logic
return CustomRecognition.AnalyzeResult(box=[x, y, w, h], detail={})The method to call the above recognition process in the pipeline is as follows. The passed parameters can be any JSON object, which will be loaded as a dict type in the code above. If you need to change the input parameters based on user input, you can use the pipeline_override operation of the interface (see Interface V2 Protocol for details).
{
"YourNodeName":{
"recognition": "Custom",
"custom_recognition": "YourRecognitionName",
"custom_recognition_param": {
// Pass parameters (optional)
// object
}
// ...
},
// or
"YourNodeName":{
"recognition": {
"type": "Custom",
"param": {
"custom_recognition": "YourRecognitionName",
"custom_recognition_param": {
// Pass parameters (optional)
// object
}
}
},
// ...
},
// ...
}Project Examples:
MultiRecognition- Multi-algorithm combined recognition, supports AND/OR/custom logicCount- Recognition counter, stops after specified number of executionsCheckStopping- Check if task is about to stop
3. Sink (Event Listener)
Location: agent/custom/sink/
Purpose: Listen to MaaFramework runtime events for logging, debug output, performance monitoring, etc.
Basic Structure:
from maa.agent.agent_server import AgentServer
from maa.context import Context, ContextEventSink
@AgentServer.context_sink()
class MyContextSink(ContextEventSink):
def on_raw_notification(self, context: Context, msg: str, details: dict):
# Handle event notifications
passProject Examples:
- Sink Module - Implements Resource, Controller, Tasker, Context event listeners
- Logger Module - Structured logging system(Deprecated)
Common APIs
Context Common Methods
# Execute recognition
reco_detail = context.run_recognition("NodeName", image)
# Execute task
context.run_task("NodeName")
# Override pipeline configuration
context.override_pipeline({"NodeName": {"next": ["NewNode"]}})
# Override image
context.override_image("image_name", image_array)
# Get screenshot
img = context.tasker.controller.post_screencap().wait().get()Info
MaaFramework uses OpenCV to process images, so the captured image data is of numpy.ndarray type and in BGR format. No additional processing is required for interaction with MaaFramework, but if you need to save the image, you must convert it to RGB format first. (Refer to the example: Screenshot)
Recognition Result Processing
from maa.define import OCRResult, TemplateMatchResult
if reco_detail and reco_detail.hit:
# OCR result
ocr_result = reco_detail.best_result
text = ocr_result.text
box = reco_detail.box # [x, y, w, h]
# All recognition results
all_results = reco_detail.all_resultsDevelopment Resources
- Integration Interface: Integration Interface
- Python Binding Source: maa/source/binding/Python
- Project Examples: Browse existing implementations in
agent/custom/
Learning Recommendations
- Read existing custom implementations in the project to understand common patterns
- Refer to MaaFramework official documentation to understand core concepts
- Study Python Binding source code for deeper API understanding
