MCPServer 与工具 schema 自动生成

本章讲最贴近使用者的一层:@mcp.tool() 装饰器背后发生了什么,以及「不写 JSON Schema」到底是怎么做到的。

1. MCPServer:一层薄糖衣

MCPServer 不是核心,它是好用接口。它内部创建了一个 lowlevel Server,把自己的 _handle_* 方法当 handler 挂上去(server/mcpserver/server.py:210 起 __init__):

# server/mcpserver/server.py 内部(节选示意,非源码)
self._lowlevel_server = Server(
    name=name or "mcp-server",
    on_list_tools=self._handle_list_tools,   # tools/list  → 列工具
    on_call_tool=self._handle_call_tool,     # tools/call  → 调工具
    on_read_resource=self._handle_read_resource,
    ...
)

真实位置:MCPServer.__init__ 把七八个 on_* 回调指向自己的 _handle_*(server/mcpserver/server.py:210-229)。所以「MCPServer 层」= 三个 Manager(ToolManager / ResourceManager / PromptManager)+ 把它们接到 lowlevel Server 的胶水。

三个 Manager 各管一类注册项,都是普通字典:

Manager	存什么	文件
ToolManager	name → Tool	server/mcpserver/tools/tool_manager.py
ResourceManager	静态资源 + URI 模板	server/mcpserver/resources/resource_manager.py
PromptManager	name → Prompt	server/mcpserver/prompts/

2. @mcp.tool() 装饰器:注册,不改函数

装饰器不包装你的函数,它只是把函数登记进 ToolManager,然后原样返回(server/mcpserver/server.py:609 tool):

# server/mcpserver/server.py:664 decorator(节选,示意
def decorator(fn):
    self.add_tool(fn, name=..., title=..., ...)   # 登记
    return fn                                       # 原样还回去

所以被 @mcp.tool() 装饰后,add 还是那个普通可直接调用的 add。真正的分析发生在 add_tool → ToolManager.add_tool → Tool.from_function。

3. 核心魔法:Tool.from_function 把函数拆成元数据

Tool.from_function(server/mcpserver/tools/base.py)在注册时一次性把函数解剖成一份 Tool 记录:

字段	怎么来的
name	name 参数或 fn.__name__
description	description 参数或 fn.__doc__(docstring)
parameters	由 func_metadata 生成的 JSON Schema
fn_metadata	一个 FuncMetadata,内含参数校验模型和输出模型
is_async	is_async_callable(fn) 判断
context_kwarg	哪个参数要注入 Context(见 §6)
resolved_params	哪些参数用 Resolve(...) 依赖注入(见第 5 章)

关键一行:parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)(tools/base.py,from_function 内)。即**「函数签名 → pydantic 模型 → JSON Schema」**。下一节拆开这条链。

4. func_metadata:签名怎么变成 schema

这是「零 schema」的心脏,在 server/mcpserver/utilities/func_metadata.py:191 func_metadata。

它要解决的小问题

MCP 协议要求每个工具带一份 JSON Schema 描述入参,好让模型知道怎么填。手写 schema 又臭又长又易错。能不能从 Python 类型标注自动生成?

思路

pydantic 本来就能「模型类 ↔ JSON Schema」互转。那就动态为每个函数造一个 pydantic 模型,字段 = 函数参数,字段类型 = 参数标注。

原理演示

# 示意,非源码:核心就这几步
import inspect
from pydantic import create_model

def make_arg_model(func):
    sig = inspect.signature(func, eval_str=True)   # 拿到参数和标注
    fields = {}
    for p in sig.parameters.values():
        ann = p.annotation                          # 比如 int
        if p.default is inspect.Parameter.empty:
            fields[p.name] = (ann, ...)             # 必填
        else:
            fields[p.name] = (ann, p.default)       # 有默认值
    return create_model(f"{func.__name__}Arguments", **fields)

Model = make_arg_model(add)          # add(a:int,b:int)
print(Model.model_json_schema())     # 直接得到 {a:int, b:int} 的 JSON Schema

真实实现

真源码就是这个骨架,只是处理了一堆边界(func_metadata.py:191-283):

• inspect.signature(func, eval_str=True) 求值字符串标注;失败抛 InvalidSignature。
• 无标注的参数默认当成 str(WithJsonSchema({"title":..., "type":"string"}))。
• 参数名撞上 BaseModel 自带属性时,用 alias 改名避免 pydantic 告警(func_metadata.py 内 hasattr(BaseModel, field_name) 分支)。
• 下划线开头的参数名直接拒绝。
• create_model(f"{func.__name__}Arguments", __base__=ArgModelBase, **params) 造出参数模型。

结构化输出(返回值也能有 schema)

structured_output 控制返回类型是否也生成 schema(func_metadata.py:283 往后):

返回标注	处理
BaseModel 子类	直接当输出模型
基本类型(str/int/…)	包一层 {"result": ...} 模型
TypedDict / dataclass	转成 pydantic 模型
list / dict / Union 等泛型	包一层 {"result": ...}
CallToolResult	用户要自己控制,不校验

巧妙细节:如果返回是 T | InputRequiredResult 这种联合,InputRequiredResult 是控制流信号不是数据,会被剥掉,只用剩下的 arm 推 schema(func_metadata.py 内 residual = tuple(a for a in args if not _is_input_required_type(a)))。多轮追问机制见第 5 章。

5. 一次工具调用:注册的元数据怎么被用

注册时拆解,调用时组装。当 tools/call 到来,MCPServer._handle_call_tool(server/mcpserver/server.py:403)造一个 Context,交给 Tool.run(tools/base.py):

  tools/call {name, arguments}
        │
        ▼
  _handle_call_tool ── 造 Context ──►  call_tool ──► ToolManager.call_tool
        │                                                   │ 按 name 查 Tool
        ▼                                                   ▼
  Tool.run:  ① 若有 Resolve 参数,先跑解析器(依赖注入)
             ② fn_metadata.call_fn_with_arg_validation:
                   用参数模型校验 arguments → 调 fn(**validated)
             ③ convert_result: 把返回值转成 CallToolResult

错误处理有讲究(tools/base.py run 末尾):

• 工具体抛 MCPError(含子类)→ 原样上抛,让内核作为顶层 JSON-RPC 协议错误返回。
• 工具体抛别的异常 → 包成 ToolError,最终变成 CallToolResult(is_error=True),即「执行失败」而非「协议错误」。

这条区分很重要:协议层错误(方法不存在、参数非法)和业务层错误(工具跑崩了)走不同通道,客户端能区分对待。

6. Context 注入:工具怎么拿到「当前请求」

工具想报告进度、读别的资源、向用户追问时,需要一个把手。做法:在签名里加一个 Context 类型的参数,SDK 自动注入(utilities/context_injection.py:12 find_context_parameter):

@mcp.tool()
async def long_job(n: int, ctx: Context) -> str:   # ← ctx 会被自动填
    await ctx.report_progress(0.5, 1.0, "halfway")   # 报进度
    data = await ctx.read_resource("file://x")        # 读资源
    return "done"

find_context_parameter 用 typing.get_type_hints 找出哪个参数标注是 Context(或 Optional[Context]),把它记进 context_kwarg,并从参数 schema 里排除(客户端看不到这个参数)。Context 本身是每请求现造的,内含 report_progress / read_resource / elicit / log 等方法(server/mcpserver/context.py:32)。

7. 资源与提示词:同一套路

• @mcp.resource("greeting://{name}"):URI 里有 {name} 占位符就当模板资源,否则当静态资源(server/mcpserver/server.py:720 resource)。URI 模板解析在 shared/uri_template.py。
• @mcp.prompt():Prompt.from_function 同样用签名生成参数(prompts/)。

三类东西共享同一个心智模型:函数即声明,类型即 schema,docstring 即描述。

8. 本章要点

• MCPServer 是糖衣,真派发在 lowlevel Server + ServerRunner(下一章)。
• 「零 schema」= func_metadata 用 create_model 把签名变 pydantic 模型,再 model_json_schema()。
• 注册时解剖(Tool.from_function),调用时组装(Tool.run)。
• 协议错误 vs 业务错误走不同通道(MCPError 上抛 vs CallToolResult(is_error=True))。

→ 下一章进入全书主线:请求进来后,内核和 Dispatcher 到底怎么处理。