第07章 FastMCP 把检索封装成 Agent 工具工单知识库已经能在 Python 进程内被普通函数调用但要让外部 Agent、Web 后端或其他语言的客户端使用这份能力函数级别的接口不够缺少协议、缺少描述、缺少跨进程通讯。MCPModel Context Protocol正是为这种场景设计的标准化协议它把工具的描述、参数、返回值与传输层一并标准化便于 Agent 自动发现和调用。本章用 FastMCP 把上一章的语义检索函数变成一个标准 MCP Tool并在另一进程里写客户端验证调用。完成本章后读者将理解 MCP 的核心概念并能把任意 Python 函数升级为可被 Agent 直接消费的工具。7.1 MCP 协议与 FastMCP理解 MCP 之前读者只需要知道它解决的核心问题让大语言模型在对话过程中按需调用外部工具并以一致方式获取工具返回的结构化数据。FastMCP 是 Python 生态中实现这一协议的轻量框架。7.1.1 MCP 的角色与传输方式一次完整的 MCP 调用涉及三方用户、模型、工具。三方协作过程“如图7-1”所示。读者从图中可以看到模型并不直接执行工具而是产出“我想调用工具 X参数是 Y”的意图由 MCP 客户端把意图转换为标准协议消息发送给 MCP Server再把工具的结果回填给模型继续生成回答。这种分工让模型只负责语言层面的决策工具只负责确定性的数据访问。MCP 协议在传输层支持多种方式包括 stdio、Server-Sent Events 与最新的 streamable HTTP。FastMCP 默认推荐 streamable HTTP本书也采用这一方式便于把 MCP Server 放在独立进程或独立服务器上。7.1.2 FastMCP 的核心抽象FastMCP 把 MCP 协议中的概念抽象为几个 Python 对象常见抽象与作用“如表7-1”所示。表 7-1 FastMCP 的核心抽象抽象在 MCP 协议中的对应在 Python 中的体现FastMCP 实例Server一个 MCP 服务进程mcp.tool 装饰器Tool 注册把普通函数注册为可被发现的工具函数签名Tool 描述与参数类型注解与 docstring 自动生成 Schema函数返回值Tool 响应字符串或结构化内容FastMCP 充分利用 Python 类型注解与 docstring函数签名映射为工具参数 Schemadocstring 映射为工具描述。这种“代码即文档”的方式让维护成本接近写一个普通函数。注意FastMCP 自动生成的 Tool Schema 基于类型注解缺少注解会导致模型看不到参数类型影响调用准确度建议所有工具参数都标注类型与默认值。7.2 把检索函数注册为 MCP Tool上一章实现的 search_tickets_semantic 已经是合适的工具形态输入自然语言查询、输出 JSON 字符串。本节给它加上 mcp.tool 装饰器让 FastMCP 接管协议层。7.2.1 FastMCP 实例与工具注册只需要两个动作实例化 FastMCP把函数加上 mcp.tool 装饰器即可。fromfastmcpimportFastMCP mcpFastMCP(电商工单向量检索服务-HTTP)mcp.tool()defsearch_tickets_semantic(query:str,n_results:int5)-str: 语义搜索工单 - 使用向量相似度 Args: query: 搜索查询自然语言描述 n_results: 返回结果数量默认5条 # ... 与上一章相同的检索实现读者注意 docstring 的写法函数主要描述写在第一行Args 段落按 Google 风格列出每个参数的含义。FastMCP 把这些信息自动转换为 MCP 协议中的 Tool 描述模型在决定是否调用工具时正是依据这些描述判断的。7.2.2 第二个工具重建向量库工具的颗粒度建议按业务动作划分而非按数据表划分。除了语义检索再注册一个重建向量库的工具便于运维场景下让 Agent 一句话触发数据重建。mcp.tool()defrebuild_vector_store()-str:重建向量库清空后重新初始化示例数据try:all_idstickets_collection.get()[ids]ifall_ids:tickets_collection.delete(idsall_ids)init_sample_data()returnjson.dumps({success:True,message:f向量库重建完成当前共{tickets_collection.count()}条数据,},ensure_asciiFalse)exceptExceptionase:returnjson.dumps({error:f重建失败:{str(e)}},ensure_asciiFalse)两个工具一个偏读、一个偏写覆盖了知识库的最小可用工具集。后续如果有按工单号查询、按状态筛选等需求按相同模式注册即可。注意MCP 工具的副作用写库、重建应在返回值中明确说明执行结果不要依赖异常退出告诉模型“出错了”模型对 JSON 内容的解读能力远好于堆栈跟踪。7.2.3 启动 MCP Server把以上代码与初始化逻辑组装到一起启动一个 streamable HTTP 模式的 MCP Server。if__name____main__:init_sample_data()print(\n启动电商工单向量检索 MCP Server...)print(服务端点: http://localhost:8001/mcp)print(\n提供工具:)print( • search_tickets_semantic - 语义搜索向量检索)print( • rebuild_vector_store - 重建向量库)mcp.run(transportstreamable-http,port8001,path/mcp)mcp.run 接收三个关键参数transport 指定传输协议、port 指定端口、path 指定路径。streamable HTTP 把 MCP 协议消息封装在 HTTP 流式请求里与本书后端的 SSE 思路是同一族便于跨语言客户端接入。启动后服务在 localhost:8001/mcp 监听整个服务进程的能力清单与端点关系“如表7-2”所示。表 7-2 MCP Server 的访问端点与工具清单端点作用调用方式http://localhost:8001/mcpMCP 协议入口通过 MCP 客户端发起 streamable HTTP 调用search_tickets_semantic语义检索工具客户端通过 call_tool 调用rebuild_vector_store重建向量库工具客户端通过 call_tool 调用读者可以把 MCP Server 想象成一台微型 REST 服务区别在于它不暴露任意自定义路径所有调用都走 /mcp 入口、由协议字段区分目的。7.3 用 MCP 客户端调用工具服务端就绪后客户端的工作很简单建立 streamable HTTP 连接、初始化会话、按工具名调用、解析返回内容。本节用 agent_client.py 中的实现展开讲解。7.3.1 建立会话并调用工具MCP 客户端的核心动作是 ClientSession.call_tool。它接收工具名与参数字典返回结构化结果。importasyncioimportjsonfrommcpimportClientSessionfrommcp.client.streamable_httpimportstreamable_http_clientasyncdefcall_agent_tool(tool_name:str,arguments:dictNone):ifargumentsisNone:arguments{}asyncwithstreamable_http_client(urlhttp://localhost:8001/mcp)as(read,write,_session_id_callback,):asyncwithClientSession(read,write)assession:awaitsession.initialize()resultawaitsession.call_tool(tool_name,arguments)forcontentinresult.content:ifcontent.typetext:returncontent.textreturnjson.dumps({error:No text content in result})代码使用嵌套的 async with外层管理传输连接内层管理会话生命周期。session.initialize 是 MCP 协议规定的握手动作必须在 call_tool 之前调用。result.content 是一个 list每项可能是文本、图片、二进制等多种类型本书的工具只返回文本所以只需要遍历找到第一个 type “text” 的内容即可。7.3.2 包装常用工具的便捷函数每次调用都要写工具名和参数字典容易出错建议为每个工具包一个便捷函数。asyncdefcall_search_tickets_semantic(query:str,n_results:int5):returnawaitcall_agent_tool(search_tickets_semantic,{query:query,n_results:n_results},)便捷函数的好处不仅是代码简洁更重要的是把工具名与参数 schema 集中在一处管理将来工具签名变化时只需要改这一个函数业务代码无感知。7.3.3 客户端联调验证把 agent_client.py 单独作为脚本运行可以快速验证 MCP Server 是否可用。if__name____main__:asyncdeftest():print(测试语义搜索...)resultawaitcall_search_tickets_semantic(退款问题,3)print(result)asyncio.run(test())读者先启动 agent_server.py再运行 agent_client.py应能看到返回的 JSON 字符串中包含三条与退款相关的工单。整条链路从客户端发起、经 MCP 协议、进入向量检索、回到客户端跑通即说明 MCP 工具封装成功。注意streamable_http_client 与 ClientSession 都是异步上下文管理器必须在 async 函数中使用写到同步代码里会立即抛错。7.4 工具设计的几条经验把函数变成 MCP Tool 不只是加装饰器那么简单工具的命名、返回结构、错误处理直接影响 Agent 调用质量。本节小结几条来自本书项目的经验。7.4.1 工具名与描述的可读性工具名既给程序看也给模型看。命名应满足两个要求动词在前突出动作意图名词在后明确操作对象。本书 search_tickets_semantic 即“搜索-工单-语义”rebuild_vector_store 即“重建-向量-库”模型从工具名就能理解功能。描述来自 docstring应包含三个层面信息功能一句话总结、典型适用场景、参数含义。模型在多个相似工具间做选择时主要靠描述区分。7.4.2 返回值结构的稳定JSON 返回值的字段名应保持稳定模型一旦学会某个工具的返回结构后续调用都会按此结构解析。新增字段是安全的、修改或删除字段则可能破坏已写好的提示模板应通过版本字段或新增工具等方式向后兼容。7.4.3 错误的显式表达Python 异常在 MCP 调用链中可能被吞掉或转成模型不易理解的形式。建议工具内部捕获异常把错误转为标准 JSON 字段。returnjson.dumps({error:f搜索失败:{str(e)}},ensure_asciiFalse)模型看到 error 字段后会自然采取重试或降级策略远好于面对空白响应或 500 错误时的无所适从。7.5 本章小结本章把语义检索能力从普通 Python 函数升级为 MCP Tool并实现了对应的客户端封装。读者现在掌握的不仅是 FastMCP 用法更是一种通用的“工具化”思路把确定性的数据访问能力沉淀为标准工具让模型只承担语言决策。接下来的工作是把 MCP 客户端嵌入到完整后端服务里给前端用户提供一个体验流畅的流式问答接口。下一章笔者将搭建 FastAPI 服务串起向量检索、RAG 提示工程、Ollama 流式生成、SSE 推流四个环节把链路推进到面向用户的最后一公里。作者光谷老亢配套源码https://github.com/kang-airtc/agent-ollama-book