跳转到内容
快速开始

基于 Mirobody 进行开发

添加自定义工具

把一个 .py 丢进工具目录、重启,引擎就会把你的函数变成一个 MCP 工具

工具就是一个普通的 Python 函数。你不用注册它,不用写它的 JSON schema,也没有什么要编译: 引擎在启动时读你的类型标注和 docstring,据此生成模型看到的 schema。

工具与 Agent 概览 讲过这遍发现流程。本页是实操手册: 文件放哪、它的每一部分会变成什么、以及怎么判断它到底加载上了没有。

发现过程会遍历 MCP_TOOL_DIRS 配置键里列出的目录。默认值只有一条,也就是引擎自带的那个包内 目录,所以你自己的工具要放进一个自己加的目录。把你的列在最前面:目录是按顺序扫的,这正是 让一个部署在不改包内代码的前提下覆盖随包工具的机制。

config.localdb.yaml
MCP_TOOL_DIRS:
- tools # 你的,先扫
- mirobody/agent/tools # 包内默认

在列出的目录里,规则很窄,命名文件之前建议先确认:

  • 只读以 .py 结尾的文件,而且不会递归进子目录:嵌套的包对发现过程是不可见的。
  • 文件名前的下划线会把这个文件排除掉__init__.py 和私有的辅助模块因此自动被跳过; 想在工具旁边放一个共享模块,也正是用这个办法。
  • 模块级函数会被注册成工具,除非名字以 _ 开头,或者它是从别处导入进来的。
  • 只在类名以 Service 结尾时才会被检查。这样的类里,公开方法成为工具;_ 开头的方法、继承自基类的方法、以及从别的模块导入进来的方法都会被忽略。
  • 导入失败只记一条日志并跳过这一个模块。服务照常启动,其它工具照常加载。

下面这个例子是一个完整可用的工具:它获得调用者身份、一个必填参数和一个可选参数, 读引擎自己的时序数据表,返回一个结构化结果。

建文件

Terminal window
touch tools/goal_service.py

文件名无所谓,类名有所谓:必须以 Service 结尾。

写这个 service

tools/goal_service.py
from datetime import datetime, timedelta
from typing import Any
from mirobody.utils import execute_query
class GoalService:
"""把用户记录下来的读数与一个目标值作比较。"""
def __init__(self):
self.name = "Goal Service"
self.version = "1.0.0"
async def count_readings_above_goal(
self,
user_info: dict[str, Any],
indicator: str,
goal: float,
days: int = 30,
) -> dict[str, Any]:
"""
统计用户最近有多少条读数达到了某个目标值。
请先调用 query_health_indicators 获得准确的指标名。
Args:
indicator: 准确的指标名,来自 query_health_indicators 的返回。
goal: 一条读数需要达到的值,单位与该指标本身一致。
days: 往前看多少天。
"""
user_id = user_info.get("user_id")
if not user_id:
return {"success": False, "error": "Authorization required."}
rows = await execute_query(
"""
SELECT tsd.start_time, tsd.value
FROM th_series_data tsd
WHERE tsd.user_id = :user_id
AND tsd.indicator = :indicator
AND tsd.deleted = 0
AND tsd.start_time >= :since
ORDER BY tsd.start_time DESC
""",
{
"user_id": user_id,
"indicator": indicator,
"since": datetime.now() - timedelta(days=days),
},
)
hits = []
for row in rows or []:
try:
if float(row["value"]) >= goal:
hits.append(str(row["start_time"]))
except (TypeError, ValueError):
continue
return {
"success": True,
"data": {
"indicator": indicator,
"goal": goal,
"readings_examined": len(rows or []),
"readings_above_goal": len(hits),
"times": hits[:20],
},
}

把目录加进列表

照上面那样,在你的 config.{env}.yaml 里把 tools 加进 MCP_TOOL_DIRS。 如果你是把文件丢进了已经列出的目录,这步可以跳过。

重启进程

Terminal window
docker compose restart mirobody

发现只在启动时运行一遍,所以新增或改动工具都需要重启。若是从检出目录直接运行, 重启 mirobody serve

确认它在了

Terminal window
curl -sX POST http://localhost:18080/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | grep count_readings_above_goal

每个参数的类型标注对应一个 JSON Schema 类型。这张映射很小,背后也没有校验层:

str
string
int
integer
float
number
bool
boolean
list[str]
array, items: string
dict[str, Any]
object
其它任何写法
string 兜底值,不会报错

没有默认值的参数是必填;有默认值的是可选,并且默认值会写进 schema。默认值为 None 时会被刻意省略,因为有些模型 API 不接受 schema 里出现 null。

模型判断该不该调用、怎么调用你的工具,唯一的依据就是 docstring。它被分三部分读取:

  1. 第一个段落标记之前的全部内容 成为工具描述;空行丢弃,其余行拼接起来
  2. Args: 每行 name: text 成为该参数的描述;缩进的续行会追加到上一行后面
  3. Returns: 会被解析,然后丢弃;它是写给人看的文档,永远到不了模型那里

段落标记不区分大小写,而且认好几种写法:Args:Arguments:Params:Parameters: (以及它们的单数形式)都能开启参数段;Returns:Results:(及其单数形式)结束它。

Args: 段里对不上任何参数名的键会被忽略,这也是为什么按惯例不在 docstring 里写 user_info:它不在 schema 里,没有什么可描述的。

声明一个 user_info 参数,引擎就会在你的函数执行前,按已认证的调用者把它填好。 它会从 schema 里被剔除,所以模型既提供不了、也伪造不了:

{"success": True, "user_id": "...", "session_id": "..."}

从里面取出调用者,取不到就拒绝这次调用:

user_id = user_info.get("user_id")
if not user_id:
return {"success": False, "error": "Authorization required."}

把它声明为第一个参数,自带的工具都是这么写的,并且每一次查询都限定在这个 user_id 上。

Service 类可以主动选择不存在。定义一个静态方法 _enabled(),返回 False,整个类就被跳过, 它的工具在任何地方都不会出现:不在 agent 的工具集里,也不在 tools/list 里。可选集成正是 这样干净地消失,而不是等到被调用时才失败。你自己的可选集成照这个写法即可:

tools/my_integration_service.py
@staticmethod
def _enabled() -> bool:
"""Only register when MY_INTEGRATION_API_KEY is configured."""
from mirobody.utils import global_config
return bool(global_config().get_str("MY_INTEGRATION_API_KEY"))

_enabled() 内部抛出的异常等同于返回 False:类被跳过并记一条警告, 所以一个写坏了的判断不会顺手把服务带崩。

返回一个 dict。引擎把结果交给 MCP 客户端时会读其中三个键:

作用
success一个 boolFalse 会把这次 MCP 结果标记为错误。
data成功时它会被拆出来,同时作为 structuredContent 一并返回。
error失败时,客户端看到的就是这个字符串。

普通 defasync def 都行,只有当函数是协程时调用方才会 await 它。任何逃出你函数的异常 都会被接住、记录,并以 {"success": False, "error": "..."} 的形式返回,因此一个坏工具不会让 一轮对话崩掉。不过自己接住更好:那样你还能说一句模型能据此行动的话。

结果要小。宁可返回一个 URL 或一个 key,也不要返回一大块数据;列表长度也要设上限: 你返回的每一个字都花在模型的上下文窗口里。

类型标注只能描述扁平的标量、数组和无类型的对象。参数是嵌套结构的工具,可以给函数挂一份 手写的 JSON Schema,属性名叫 inputSchema,引擎会原样采用:

tools/report_service.py
class ReportService:
async def build_report(self, sections: list[dict], user_info: dict) -> dict:
"""Build a report from a nested section tree."""
...
ReportService.build_report.inputSchema = {
"type": "object",
"properties": {
"sections": {
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"indicators": {"type": "array", "items": {"type": "string"}},
},
"required": ["title"],
},
}
},
"required": ["sections"],
}

随包发布的工具里就有用这个机制的:在类定义之后,把 chart_schema/*.json 里的 schema 挂到每个图表方法上。这个属性一旦存在, 生成 schema 时就不再看类型标注了:标注留着自己看,契约以挂上去的那份 schema 为准。

新发现的工具默认对每个 agent 都开放,因为发行版 config.yaml 两个名单都没配。想按 agent 收窄,用 ALLOWED_TOOLS_{NAME}(白名单)和 DISALLOWED_TOOLS_{NAME}(黑名单,最后生效, 因此它总是说了算),其中 {NAME} 是 agent 名字的大写形式, 见 工具与 Agent 概览

tools/list 是唯一的事实。你的工具不在里面,启动日志会告诉你为什么:每个注册成功的工具都会记 Loaded tool: <name>,导入失败的模块记 Error importing tool module <module>, 主动退出的类记 Skipping disabled tool class: <Class>

Terminal window
docker compose logs mirobody | grep -E "Loaded tool|Error importing tool module|Skipping disabled"

如果这几行里都没提到你的文件,就顺着往下查:目录在 MCP_TOOL_DIRS 里吗、文件名是不是以下划线 开头、类名有没有以 Service 结尾、方法是不是公开的,最后,你重启了吗?