第 2 章:函数如何自动长出 JSON schema(核心魔法)
这是全项目最有含金量的一章。FastMCP 的整个卖点——「你只写函数,schema 和校验白送」——就靠这一章的机制。看完你会明白它没有任何魔法,只是把 pydantic 用到了极致。
2.1 它要解决的小问题
MCP 协议要求:每个工具都要向客户端声明一个 JSON schema,描述它接受什么参数。比如 add(a: int, b: int) 对应:
{"type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, "required": ["a", "b"]}
手写 MCP 服务器就得手敲这段 JSON,而且要和函数签名保持同步——改一次参数就得改两处。FastMCP 的目标:从函数签名自动生成它,永不失同步。
2.2 思路:把「函数」本身交给 pydantic
关键洞察:pydantic 的 TypeAdapter 不止能吃类型,还能吃一个可调用对象。 给它一个函数,它会:
- 读函数签名,把每个参数当成一个字段;
- 生成整个「参数对象」的 JSON schema;
- 提供
validate_python(args)——校验入参、并直接调用该函数。
也就是说,一个 TypeAdapter(fn) 同时是「schema 生成器」和「带校验的调用器」。FastMCP 就是围着这个双重身份搭起来的。工具函数在 utilities/types.py:45(get_cached_typeadapter)——带缓存,因为同一个函数会被反复 adapt。
一句话直觉: FastMCP 不是「解析你的函数」,而是「把你的函数塞给 pydantic,让 pydantic 当它是一个待校验的数据结构」。schema 是副产品,校验是副产品,调用也是副产品。
2.3 原理演示(示意,非源码)
下面这段用纯 pydantic 演示 FastMCP 的核心把戏,帮你建立直觉:
# 示意,非源码。演示「TypeAdapter 吃一个函数」的双重身份
from pydantic import TypeAdapter
def add(a: int, b: int) -> int:
return a + b
adapter = TypeAdapter(add) # 把函数交给 pydantic
# 身份一:产出 JSON schema(这就是发给 MCP 客户端的东西)
print(adapter.json_schema())
# → {'properties': {'a': {'type':'integer'}, 'b': {'type':'integer'}}, ...}
# 身份二:校验入参 + 直接调用函数
result = adapter.validate_python({"a": 1, "b": 2}) # 校验通过后执行 add(1,2)
print(result) # → 3
# 校验失败示例:
adapter.validate_python({"a": "oops", "b": 2}) # 抛 pydantic.ValidationError
重点看: json_schema() 和 validate_python() 来自同一个 adapter,所以 schema 和校验规则天然一致,不可能失同步。FastMCP 生产代码就是这个模式的工业化版本。
2.4 真实实现:ParsedFunction.from_function
注册期解析函数的主函数是 tools/function_parsing.py:178(ParsedFunction.from_function)。它比上面的演示多做了几件真实世界必须处理的事,按顺序:
① 拒绝 *args / **kwargs。 无法映射成固定 schema,直接报错(function_parsing.py:187 起)。
② 抽取 docstring 里的参数描述。 用 parse_docstring(utilities/docstring_parsing.py)解析函数文档,把每个参数的说明后面注入 schema。真实代码:
# tools/function_parsing.py 内 —— docstring 描述注入 schema
if parsed_docstring.parameters:
properties = input_schema.get("properties", {})
for param_name, param_desc in parsed_docstring.parameters.items():
if param_name in properties and "description" not in properties[param_name]:
properties[param_name]["description"] = param_desc
注意 "description" not in ... 这个守卫:如果你已经用 Annotated[int, Field(description=...)] 显式写了描述,它不会被 docstring 覆盖——显式注解优先。
③ 把 Context 参数变成依赖注入。 工具函数如果有个 ctx: Context 参数,它不是模型要填的入参,而是运行时注入的。解析前先调 transform_context_annotations(fn) 把它转成依赖(第 3 章详述),再用 without_injected_parameters(fn) 剥掉这些参数——这样生成 schema 时 ctx 不会出现在给模型看的参数列表里(function_parsing.py 内,约 :234-:245)。
④ 生成并压缩 schema。 核心两行:
# tools/function_parsing.py —— schema 生成的心脏
input_type_adapter = get_cached_typeadapter(wrapper_fn)
input_schema = input_type_adapter.json_schema()
input_schema = compress_schema(input_schema, prune_params=..., prune_titles=True)
compress_schema(utilities/json_schema.py:688)做清理:内联 pydantic 生成的 $defs 引用、剪掉冗余的 title 字段,让发给模型的 schema 更紧凑干净。
⑤ 从返回类型注解生成输出 schema。 读 sig.return_annotation,如果是具体类型就生成 output_schema。但有几类不能生成:bytes(不能表示成结构化 JSON,_contains_bytes_type 在 function_parsing.py:39)、prefab UI 组件类型等,会被替换成「不可序列化」哨兵跳过。
2.5 结果:一个 FunctionTool
解析完,FunctionTool.from_function(tools/function_tool.py:211)把 ParsedFunction 的产物组装成最终对象:name、description(优先用显式的,否则用 docstring 首段)、parameters(输入 schema)、output_schema、tags、超时、鉴权等(function_tool.py:363 起的 return cls(...))。
发给客户端时,再由 to_mcp_tool(tools/base.py:222)转成官方 SDK 的 MCPTool 线格式:
# tools/base.py:222 to_mcp_tool —— 转成 MCP 线格式
mcp_tool = MCPTool(
name=..., title=..., description=...,
inputSchema=self.parameters, # ← 就是上面生成的输入 schema
outputSchema=self.output_schema,
...
)
2.6 巧妙之处 / 坑
- 同源保证一致。 schema 和运行时校验共享同一个
TypeAdapter,结构上杜绝了「声明的参数和实际校验的参数不一致」这类 bug。 - 超时 + 同步函数的矛盾被提前拒绝。 如果给一个同步函数设了
timeout又设run_in_thread=False,from_function直接报错(function_tool.py:326起):同步内联执行没有可取消的检查点,anyio.fail_after拦不住它,超时会静默失效——与其埋雷,不如注册期就让你二选一。这是很典型的「把不可能正确的组合在最早的时机拒掉」。 - callable 类当工具时,描述取类 docstring、参数描述取
__call__的 docstring(function_parsing.py:212起有一段注释解释:类 docstring 的 Args 通常描述__init__,拿来注入__call__的 schema 会串味)。这种细节体现了它对真实用法的打磨。
代码地图
| 主题 | 文件路径 | 符 号名 |
|---|---|---|
| 解析函数主入口 | fastmcp_slim/fastmcp/tools/function_parsing.py | ParsedFunction.from_function |
| 带缓存的 TypeAdapter | fastmcp_slim/fastmcp/utilities/types.py | get_cached_typeadapter |
| schema 压缩/清理 | fastmcp_slim/fastmcp/utilities/json_schema.py | compress_schema |
| docstring 解析 | fastmcp_slim/fastmcp/utilities/docstring_parsing.py | parse_docstring |
| 组装 FunctionTool | fastmcp_slim/fastmcp/tools/function_tool.py | FunctionTool.from_function |
| bytes/prefab 排除 | fastmcp_slim/fastmcp/tools/function_parsing.py | _contains_bytes_type、_contains_prefab_type |
| 转 MCP 线格式 | fastmcp_slim/fastmcp/tools/base.py | Tool.to_mcp_tool |