1. 为什么要把 REST API 封装成 MCP 服务
手里有一套跑了很久的 REST API,接口稳定、文档齐全、调用方也不少,但最近半年我越来越频繁地遇到同一个尴尬:想让 AI 助手直接帮我操作这些接口,结果要么是手动写一堆胶水代码,要么是让模型自己拼 URL 和参数,十次里有三次拼错字段名。REST API 是给程序员的 HTTP 客户端设计的,不是给模型设计的,这两者的“使用姿势”根本不在一个频道上。
MCP(Model Context Protocol)解决的正是这个错位问题。它本质上是一套让模型能够以结构化方式发现和调用外部能力的协议:模型不需要猜你的接口长什么样,而是通过 MCP 服务暴露出来的工具描述,明确知道“有这么个工具、它接受哪些参数、每个参数是什么类型、返回什么”。这跟 REST 里靠 OpenAPI 文档 + 人工阅读的方式完全不同,MCP 的工具描述是直接喂给模型的上下文,模型可以自主决策调用哪个工具、传什么参数。
把现有 REST API 封装成 MCP 服务,核心价值有三个。第一是复用,你不需要重写业务逻辑,REST 层继续对外服务,MCP 层只是加了一个“翻译适配器”。第二是可控,哪些接口暴露给模型、参数怎么校验、返回怎么裁剪,全在你手里,而不是把整个 API 网关裸奔给模型。第三是可观测,MCP 服务的调用日志、参数校验失败、超时重试,都能按你的规范统一处理。
这篇文章适合两类人:一类是手里有成熟 REST API、想快速接入 AI 工具链的后端或平台工程师;另一类是想理解 MCP 服务到底怎么落地、不想只看概念介绍的技术负责人。我会用 FastMCP + Pydantic 这套组合,把封装过程拆到能直接抄作业的程度,包括参数映射、错误处理、鉴权透传、流式输出这些实际会踩坑的地方。
需要先说明一点:MCP 本身还在快速演进,不同客户端对协议版本、传输方式的支持程度不一样。我下面讲的是基于当前主流实践的方案,具体到你自己的环境,传输层选型(stdio 还是 HTTP)需要根据客户端能力来定,这一点后面会专门讲。
2. 封装前的整体设计与关键取舍
2.1 先想清楚:MCP 服务不是 REST 的镜像
很多人第一反应是“我有 20 个 REST 接口,那就暴露 20 个 MCP 工具”。这个思路在接口少的时候没问题,但接口一多就会出问题:模型的上下文窗口是有限的,工具描述太多会挤占推理空间,而且模型在几十个相似工具里选错的概率会明显上升。
我的做法是按业务场景聚合,而不是按 REST 端点一一对应。举个例子,假设你有一套订单相关的 REST API:创建订单、查询订单、取消订单、修改地址。这四个接口在 REST 世界里是四个端点,但在 MCP 世界里,更合理的做法可能是暴露两个工具:order_manage(接受 action 参数区分 create/cancel/update_address)和order_query(按订单号或条件查询)。这样模型只需要理解两个工具的语义,而不是四个。
当然,聚合也有代价:参数会变复杂,Pydantic 模型需要处理条件必填。所以我的经验法则是:如果两个接口的调用场景高度重叠、参数有 60% 以上重合,就考虑合并;否则保持独立。这个阈值不是拍脑袋,是我在实际项目里试出来的——合并太多会导致参数校验逻辑爆炸,合并太少又回到工具过多的问题。
2.2 传输层选型:stdio 还是 HTTP
MCP 支持多种传输方式,最常见的是 stdio(标准输入输出)和 HTTP(包括 SSE 和 Streamable HTTP)。这个选择直接决定了你的服务怎么部署、怎么被客户端发现。
stdio 的优点是简单、无需网络配置、进程生命周期由客户端管理,适合本地工具类场景,比如你在 IDE 里让 AI 助手调用本地脚本。缺点是没法多客户端共享,每个客户端都要自己拉起一个进程。
HTTP 的优点是服务化、可多客户端复用、方便做鉴权和限流,适合团队内部共享的服务。缺点是需要处理网络、认证、超时这些额外问题。
我的建议是:如果这个 MCP 服务是给团队多人用的,直接上 HTTP;如果只是个人本地辅助,stdio 更省事。我自己的项目里,对外共享的服务统一走 Streamable HTTP,本地调试用 stdio 快速验证逻辑。
2.3 工具描述怎么写才不让模型犯迷糊
工具描述(tool description)是 MCP 里最容易被忽视、但影响最大的部分。模型能不能正确调用你的工具,80% 取决于描述写得好不好。
我见过太多人把 REST 的接口文档直接复制过来当描述,结果模型根本不知道怎么用。REST 文档是写给人类看的,里面有大量“参见 XX 章节”“需先调用 YY 接口”这种上下文依赖,模型看不到这些上下文。
好的工具描述应该包含三部分:这个工具做什么(一句话)、什么时候该用(触发场景)、参数怎么填(每个参数的业务含义,不是类型)。比如不要写“order_id: string”,而要写“order_id: 订单编号,格式为 ORD 开头的 16 位字符串,可从 order_query 工具获取”。后者模型一看就知道该去哪里拿这个值。
2.4 参数校验:Pydantic 是你的第一道防线
REST API 通常有自己的参数校验,但 MCP 层必须再做一次,原因有两个:一是模型生成的参数不可信,可能缺字段、类型错、超范围;二是 MCP 层的校验失败可以返回结构化的错误信息给模型,让模型自己纠正重试,而不是直接抛 500。
Pydantic 在这里几乎是标配。它能把参数定义、类型校验、默认值、描述一次性搞定,而且 FastMCP 原生支持 Pydantic 模型作为工具参数。我后面会给出具体的模型定义方式,包括条件必填、枚举约束、嵌套对象这些实际会用到的写法。
3. 核心细节解析与实操要点
3.1 环境准备与依赖选型
先把依赖理清楚。核心是三个包:fastmcp(MCP 服务框架)、pydantic(参数模型)、httpx(异步 HTTP 客户端,用来调你的 REST API)。如果你用的是同步的 requests,也可以,但 MCP 服务本身是异步的,混用同步客户端会阻塞事件循环,所以强烈建议用 httpx 的异步模式。
pip install fastmcp pydantic httpx版本上,FastMCP 迭代很快,建议锁一个近期稳定版本,不要用 latest 裸奔。我当前用的是 fastmcp 2.x 系列,pydantic 2.x。如果你还在 pydantic 1.x,部分写法(比如Field的某些参数)会有差异,建议先升级。
注意:FastMCP 和官方 MCP Python SDK 是两个东西。FastMCP 是更高层的封装,写起来更简洁;官方 SDK 更底层,控制力更强。新手直接用 FastMCP,遇到它不支持的场景再降级到官方 SDK。
3.2 把 REST 接口映射成工具函数
假设你有一个查询用户信息的 REST 接口:GET /api/v1/users/{user_id},返回 JSON。封装成 MCP 工具的第一步是定义参数模型:
from pydantic import BaseModel, Field class GetUserInput(BaseModel): user_id: str = Field( ..., description="用户唯一标识,格式为 U 开头的 12 位字符串,例如 U123456789012" ) include_profile: bool = Field( default=False, description="是否返回详细档案信息,默认 false 只返回基础信息" )这里有几个细节值得说。Field(...)里的...表示必填,这是 Pydantic 的写法。description一定要写清楚业务含义和格式,这是给模型看的。include_profile这种可选参数给默认值,模型不传也能正常工作。
然后是工具函数本身:
from fastmcp import FastMCP import httpx mcp = FastMCP("user-service") BASE_URL = "https://your-api.example.com" @mcp.tool() async def get_user(input: GetUserInput) -> dict: """查询用户基础信息或详细档案。 当需要获取用户昵称、注册时间、账号状态时使用此工具。 如果需要用户的历史订单,请改用 order_query 工具。 """ async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( f"{BASE_URL}/api/v1/users/{input.user_id}", params={"include_profile": input.include_profile} ) resp.raise_for_status() return resp.json()注意 docstring 的写法。FastMCP 会把 docstring 作为工具描述传给模型,所以这里不是写给人看的注释,而是写给模型看的说明书。我特意加了一句“如果需要用户的历史订单,请改用 order_query 工具”,这是在帮模型做工具选择,减少误调用。
3.3 错误处理:别让 REST 的错误码直接透传
REST API 返回 404、403、500 的时候,如果你直接把状态码和原始错误体抛给模型,模型大概率不知道怎么处理。MCP 层的错误处理要做两件事:把技术错误翻译成业务语义,以及告诉模型下一步该怎么做。
from fastmcp.exceptions import ToolError @mcp.tool() async def get_user(input: GetUserInput) -> dict: """...""" try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(...) if resp.status_code == 404: raise ToolError( f"用户 {input.user_id} 不存在。" "请确认 user_id 是否正确,或先用 user_search 工具按昵称查找。" ) if resp.status_code == 403: raise ToolError( "当前凭证无权访问该用户信息。" "请检查是否使用了正确的 API Key,或联系管理员开通权限。" ) resp.raise_for_status() return resp.json() except httpx.TimeoutException: raise ToolError("查询用户信息超时,请稍后重试。如果持续超时,可能是上游服务异常。")ToolError是 FastMCP 提供的异常类型,抛出来后模型会收到结构化的错误信息,并且可以基于错误信息决定是否重试或换工具。这比直接抛 HTTP 异常友好太多。
实操心得:错误信息里不要暴露内部实现细节,比如数据库表名、内部服务地址。模型可能会把这些信息复述给用户,造成信息泄露。错误信息要面向“业务操作”而不是“技术排查”。
3.4 鉴权透传:API Key 怎么安全地传给 REST 层
你的 REST API 大概率需要鉴权。MCP 服务作为中间层,鉴权信息怎么传是个关键问题。有三种常见做法:
第一种是服务端固定凭证,MCP 服务启动时从环境变量读取 API Key,所有请求都用这个 Key。适合内部服务、权限统一的场景。
第二种是客户端透传,MCP 客户端在调用工具时把用户的凭证传进来,MCP 服务原样转发给 REST API。适合多租户、权限差异大的场景。
第三种是令牌交换,MCP 服务用自己的凭证去换取用户级令牌。适合有统一认证中心的场景。
我自己的项目里,内部工具用第一种,对外服务用第二种。第二种的实现方式是在工具参数里加一个auth_token字段,但这个字段不应该出现在工具描述里让模型去填——它应该由客户端在协议层注入。FastMCP 支持通过 context 获取请求元数据,具体做法依赖你用的传输方式,stdio 模式下通常从环境变量读,HTTP 模式下从请求头读。
from fastmcp import Context @mcp.tool() async def get_user(input: GetUserInput, ctx: Context) -> dict: """...""" api_key = ctx.request_context.meta.get("api_key") if ctx.request_context else None if not api_key: raise ToolError("缺少 API Key,无法调用用户服务。") headers = {"Authorization": f"Bearer {api_key}"} # ... 后续请求带上 headers注意:不要把 API Key 写进工具描述或参数默认值里,那等于把密钥暴露给模型上下文。密钥只能通过协议层或环境变量传递。
4. 完整实操流程与关键环节实现
4.1 从零搭建一个可运行的 MCP 服务
我把完整流程拆成六步,每一步都有明确的产出物。
第一步:确定工具清单。列出你要暴露的 REST 接口,然后按 2.1 节的聚合原则合并。产出物是一张表:工具名、对应 REST 端点、参数、返回。
第二步:定义 Pydantic 模型。每个工具一个 Input 模型,必要时定义 Output 模型。Output 模型不是必须的,但定义了能让返回结构更稳定,模型解析起来更准。
第三步:实现工具函数。每个函数做四件事:校验参数(Pydantic 自动做)、构造 REST 请求、处理错误、裁剪返回。
第四步:配置传输层。stdio 模式直接mcp.run();HTTP 模式配置 host、port、路径。
第五步:本地验证。用 FastMCP 自带的测试客户端或 MCP Inspector 验证工具能被正确发现和调用。
第六步:接入真实客户端。把服务注册到你的 AI 客户端,跑几个真实场景验证。
4.2 参数映射的三种典型场景
REST 参数和 MCP 参数不是一一对应的,我遇到过的映射场景主要有三类。
场景一:路径参数 + 查询参数混合。REST 里GET /users/{id}?detail=true,MCP 里统一成一个 Input 模型,函数内部拆开拼 URL。这个前面已经演示过。
场景二:POST body 嵌套结构。REST 的 body 可能是嵌套 JSON,MCP 的 Input 模型也要用嵌套 Pydantic 模型对应:
class Address(BaseModel): province: str = Field(..., description="省份,例如 广东省") city: str = Field(..., description="城市,例如 深圳市") detail: str = Field(..., description="详细地址,不超过 200 字") class CreateOrderInput(BaseModel): user_id: str = Field(..., description="下单用户 ID") items: list[str] = Field(..., description="商品 ID 列表,至少一个") address: Address = Field(..., description="收货地址") remark: str = Field(default="", description="订单备注,可选")嵌套模型的好处是模型能理解结构层次,不会把 address 拍平成一个字符串。
场景三:枚举值约束。REST 接口经常有状态字段,取值有限。用 Pydantic 的Literal或Enum约束,模型就不会传非法值:
from typing import Literal class UpdateOrderInput(BaseModel): order_id: str = Field(..., description="订单编号") action: Literal["cancel", "confirm", "modify_address"] = Field( ..., description="操作类型:cancel 取消订单,confirm 确认收货,modify_address 修改地址" )用Literal而不是str,模型在生成参数时会受到约束,非法值的概率大幅降低。
4.3 返回结果裁剪:别把整个 JSON 塞回去
REST API 返回的 JSON 往往包含大量模型不需要的字段,比如内部 ID、创建时间戳、审计信息。这些字段塞进模型上下文,既浪费 token,又可能干扰模型判断。
我的做法是在 MCP 层做一次字段裁剪,只保留业务相关的字段。比如用户接口返回 30 个字段,MCP 层只返回 8 个:
RAW_FIELDS = ["user_id", "nickname", "status", "register_time", "profile"] @mcp.tool() async def get_user(input: GetUserInput) -> dict: """...""" raw = await _call_rest(...) return {k: raw.get(k) for k in RAW_FIELDS if k in raw}如果返回是列表,还要考虑分页。REST 可能一次返回 100 条,MCP 层应该默认只返回前 10 条,并在描述里说明“如需更多请用 offset 参数”。这是防止上下文爆炸的关键。
4.4 流式输出:长任务怎么让模型边收边处理
有些 REST 接口是流式的,比如导出报表、生成内容。MCP 支持流式返回,FastMCP 里可以用yield逐步返回内容。这个能力在“使用 mcp 工具流式输出内容到文件”这类场景里特别有用。
@mcp.tool() async def export_report(input: ExportInput): """导出报表,流式返回内容块。""" async with httpx.AsyncClient(timeout=60.0) as client: async with client.stream("GET", f"{BASE_URL}/reports/{input.report_id}") as resp: async for chunk in resp.aiter_text(): yield chunk客户端收到的是逐步到达的内容块,可以边收边写文件或边展示。注意流式工具的返回类型和普通工具不同,具体写法要参考你用的 FastMCP 版本,不同版本 API 有差异。
4.5 本地验证与调试
写完服务别急着接客户端,先用 MCP Inspector 或 FastMCP 的测试客户端跑一遍。我通常验证四件事:工具列表能不能正确列出、每个工具的参数 schema 对不对、正常调用能不能返回、错误场景能不能返回可读的错误信息。
# 快速验证脚本 import asyncio from fastmcp import Client async def main(): async with Client("user_service.py") as client: tools = await client.list_tools() for t in tools: print(t.name, t.description[:50]) result = await client.call_tool("get_user", {"input": {"user_id": "U123456789012"}}) print(result) asyncio.run(main())这个脚本能跑通,基本就没大问题了。跑不通的话,八成是参数模型定义和调用方式不匹配,仔细看报错信息。
5. 常见问题与排查技巧实录
5.1 模型不调用我的工具,怎么办
这是最高频的问题。模型不调用工具,通常有三个原因。
第一是工具描述太模糊。如果描述写的是“获取用户信息”,模型不知道什么时候该用。改成“当用户询问账号状态、昵称、注册时间时使用”,触发场景明确了,调用率会明显上升。
第二是工具太多,模型选择困难。前面说过,工具数量控制在 10 个以内比较稳妥。超过的话考虑合并或分组。
第三是参数描述缺失。模型不知道某个参数该填什么,就会放弃调用。每个参数的 description 都要写清楚格式和来源。
5.2 参数校验总是失败
模型生成的参数不符合 Pydantic 模型,常见原因和对策我整理成了一张表:
| 现象 | 常见原因 | 对策 |
|---|---|---|
| 必填字段缺失 | 描述里没强调必填 | 在 description 里加“必填”字样 |
| 类型错误(字符串传成数字) | 模型对类型理解偏差 | 用更严格的类型,或在描述里给示例 |
| 枚举值非法 | 没用 Literal 约束 | 改用 Literal 或 Enum |
| 嵌套对象结构错 | 嵌套模型描述不清 | 给嵌套模型每个字段都写描述 |
| 日期格式错 | 没指定格式 | 描述里写明“格式 YYYY-MM-DD” |
实操心得:Pydantic 的校验错误信息默认比较技术化,模型不一定看得懂。可以在工具函数里捕获
ValidationError,转成更口语化的提示再抛ToolError。
5.3 调用超时或卡死
MCP 服务调 REST API 超时,排查顺序是:先看 REST API 本身是否正常(用 curl 直接测),再看 MCP 服务的超时设置是否合理,最后看是不是同步阻塞了事件循环。
最常见的坑是用了同步的requests库。MCP 服务是异步的,同步调用会阻塞整个事件循环,导致其他请求全部卡住。一定要用httpx.AsyncClient或aiohttp。
5.4 鉴权信息丢失
HTTP 传输模式下,鉴权信息通常放在请求头里。如果 MCP 服务读不到,检查两点:一是客户端有没有正确配置请求头,二是 MCP 服务的传输层有没有把请求头透传到 context。stdio 模式下没有请求头概念,鉴权信息只能从环境变量或启动参数读。
5.5 返回内容太大导致上下文溢出
这个问题的根源是没做返回裁剪。对策有三层:第一层是字段裁剪,只返回必要字段;第二层是条数限制,列表默认返回前 N 条;第三层是分页,提供 offset 和 limit 参数让模型自己控制。三层做完,基本不会溢出。
6. 工具选型与扩展思路
6.1 FastMCP vs 官方 SDK vs 自研
FastMCP 适合快速落地,API 简洁,Pydantic 集成好,大部分场景够用。官方 SDK 适合需要精细控制协议细节的场景,比如自定义传输、自定义能力协商。自研只建议在你有非常特殊的需求时考虑,比如要嵌入到已有框架里,否则维护成本太高。
我的选择是:新项目一律 FastMCP 起步,遇到它解决不了的问题再降级到官方 SDK。这样前期开发效率最高,后期也有退路。
6.2 和现有 API 网关的关系
MCP 服务不应该替代 API 网关,而应该和网关配合。网关负责限流、鉴权、路由,MCP 服务负责协议转换和工具描述。我通常把 MCP 服务部署在网关后面,MCP 服务调网关,网关再调后端服务。这样网关的策略对 MCP 调用同样生效,不用重复实现。
6.3 后续可以扩展的方向
服务跑起来之后,有几个方向可以继续做。一是加缓存,对读多写少的接口在 MCP 层加一层缓存,减少 REST 调用。二是加审计,记录每次工具调用的参数和结果,方便排查问题。三是加工具分组,工具多了之后按业务域拆成多个 MCP 服务,客户端按需加载。四是加健康检查,MCP 服务启动时探测 REST API 是否可用,不可用就快速失败而不是让模型等超时。
我个人在实际操作中的体会是,MCP 封装这件事,技术难度不高,难的是“站在模型的角度想问题”。你写的每一个描述、每一个参数、每一条错误信息,都是模型理解你服务的唯一途径。把模型当成一个聪明但对你系统一无所知的新同事,把该交代的都交代清楚,封装出来的服务就好用。反过来,如果只是把 REST 文档机械翻译一遍,模型用起来就会磕磕绊绊,最后你还是得手动兜底。