1. 从 REST 到 MCP：为什么需要这层封装
手里有一套跑得好好的 REST API，接口文档齐全，鉴权、限流、日志都配好了，突然要把它接进大模型工具链里，让模型能直接调用——这件事听起来简单，做起来坑不少。直接让模型去发 HTTP 请求？模型不知道你的接口长什么样，参数怎么填，返回结构是什么，它只能瞎猜。把接口文档塞进提示词里？上下文窗口撑不住，接口一多就爆了。
MCP（Model Context Protocol）就是来解决这个问题的。它本质上是一套标准化的协议，让模型能够以结构化的方式发现和调用外部能力。你可以把它理解成"给模型看的 API 说明书 + 调用通道"，模型通过 MCP 服务暴露的 tool 列表知道有哪些能力可用，每个 tool 需要什么参数，然后按格式发起调用，服务端执行完把结果返回。
那为什么不直接写个 MCP 服务，非要从现有 REST API 封装？因为绝大多数团队的核心业务逻辑已经在 REST API 里了——订单查询、用户管理、数据统计、文件处理，这些接口经过生产环境验证，有完整的错误处理、权限控制、审计日志。重新写一套 MCP 服务等于把业务逻辑复制一遍，维护成本翻倍，还容易出现两边行为不一致的问题。封装而不是重写，是工程上更务实的选择。
这篇文章面向的是有一定后端经验、手里有现成 REST API、想把它们接入大模型工具链的开发者。我会用 FastMCP 这个框架来演示，因为它是目前 Python 生态里封装 MCP 服务最顺手的工具之一，配合 Pydantic 做参数校验，整个链路非常清晰。全文会覆盖从接口梳理、工具定义、参数映射、错误处理到部署调试的完整流程，每个环节都会解释为什么这么做，以及我实际踩过的坑。
先说一个核心认知：MCP 封装不是简单的 HTTP 转发。如果你只是把 REST 请求原样包一层，那模型用起来会非常痛苦——参数命名不直观、返回结构太啰嗦、错误信息看不懂。好的封装要做的是"翻译"，把面向程序员的 API 翻译成面向模型的工具，这中间有大量的设计决策要做。
2. 封装前的接口盘点与工具粒度设计
2.1 先搞清楚哪些接口值得暴露给模型
不是所有 REST API 都适合封装成 MCP tool。我见过有人把整个后台管理系统的接口全量暴露，结果模型面对两百多个 tool 直接懵了，选错工具的概率极高。工具数量和质量之间要平衡。
判断一个接口是否适合暴露，我会问三个问题：
- 模型能自主决定调用时机吗？比如"查询订单状态"这种，模型在对话中能判断出用户想查订单，可以暴露。"删除用户"这种高危操作，模型不应该自主决定，要么不暴露，要么加二次确认机制。
- 参数模型能填对吗？如果接口需要传一个内部加密后的 token 或者复杂的嵌套结构，模型很难正确构造，这种就不适合直接暴露，需要在封装层做转换。
- 返回结果模型能理解吗？如果返回的是一大坨嵌套 JSON，字段名还是缩写，模型解析起来很吃力。这种情况需要在封装层做裁剪和重命名。
我一般的做法是先列出所有接口，按业务域分组，每个域挑出 3 到 8 个高频、低危、参数简单的接口作为第一批。跑通之后再逐步扩展。
2.2 工具粒度：一个接口一个 tool，还是合并
这是封装时第一个要做的设计决策。举个例子，你的 REST API 里有这些接口：
GET /api/orders/{id} GET /api/orders?user_id=xxx&status=xxx POST /api/orders PUT /api/orders/{id}/status四种做法：
做法一：一个接口一个 tool。四个 tool 分别叫get_order_by_id、list_orders、create_order、update_order_status。优点是映射直接，缺点是 tool 数量膨胀快。
做法二：按资源合并。一个manage_ordertool，通过一个action参数区分操作。优点是 tool 少，缺点是模型容易在 action 上选错，而且参数定义会变得很复杂。
做法三：按读写信道合并。查询类合并成一个query_orders，写入类合并成一个modify_order。这是我在多数场景下的选择，读写分离符合模型的使用习惯，tool 数量可控。
做法四：按业务场景合并。比如"查订单详情"和"查订单物流"合并成一个get_order_full_info，一次调用返回模型需要的所有信息。这种适合模型经常需要组合查询的场景。
我的经验是：读操作可以适度合并，写操作尽量独立。因为读操作合并后参数用可选字段控制，模型理解成本低；写操作合并后风险集中，一旦模型选错 action 可能造成误操作。
2.3 参数命名：别把 API 的字段名直接搬过来
REST API 的字段名往往是给程序员看的，比如usr_id、ord_stat、crt_tm。这些名字模型能猜，但猜不准。封装时要重命名成自然语言风格的字段：
| REST 字段 | MCP 参数名 | 说明 |
|---|---|---|
| usr_id | user_id | 补全缩写 |
| ord_stat | order_status | 明确语义 |
| crt_tm | created_after | 从名词改成条件表达 |
| pg_sz | page_size | 补全缩写 |
重命名的原则是：让模型只看参数名就能猜到该填什么。created_after比crt_tm好，因为前者暗示了这是一个时间过滤条件，后者只是个字段名。
3. 用 FastMCP 搭骨架：从零到能跑通的最小服务
3.1 环境准备与依赖选择
FastMCP 是 MCP 官方 Python SDK 之上的一个高层封装，把 tool 注册、参数校验、传输层这些繁琐的事情都简化了。安装很直接：
pip install fastmcp httpx pydantic这里解释一下三个依赖的分工：fastmcp负责 MCP 协议层，httpx负责调用你的 REST API（比 requests 更适合异步场景），pydantic负责参数定义和校验。如果你用的是同步的 requests，也能跑，但 FastMCP 本身支持异步，用 httpx 能发挥全部性能。
注意：FastMCP 的版本迭代比较快，API 偶有变动。建议锁定一个稳定版本，比如
fastmcp==2.x，避免线上环境因为自动升级出问题。
3.2 最小可运行服务
先写一个最简单的版本，把"查询订单"这个接口封装成 tool：
from fastmcp import FastMCP import httpx from pydantic import Field mcp = FastMCP("order-service") REST_BASE = "https://your-api.example.com" API_KEY = "your-api-key" @mcp.tool() async def get_order( order_id: str = Field(description="订单ID，格式为 ORD 开头的字符串") ) -> dict: """根据订单ID查询订单详情，返回订单状态、金额、创建时间等信息。""" async with httpx.AsyncClient() as client: resp = await client.get( f"{REST_BASE}/api/orders/{order_id}", headers={"Authorization": f"Bearer {API_KEY}"}, timeout=10.0 ) resp.raise_for_status() return resp.json() if __name__ == "__main__": mcp.run()跑起来之后，模型就能通过 MCP 协议发现get_order这个 tool，知道它需要一个order_id参数，调用后能拿到订单详情。
但这段代码离"工业级"还差得远。它没有错误处理，没有参数校验，没有日志，没有超时重试，返回结构也没做裁剪。接下来几节逐个补齐。
3.3 为什么用 Field 描述参数而不是写在 docstring 里
FastMCP 支持两种方式描述参数：一种是在 docstring 里写，一种是用 Pydantic 的Field(description=...)。我强烈推荐后者，原因有两个：
第一，Field的描述会直接进入 tool 的 JSON Schema，模型看到的是结构化的参数说明，比从 docstring 里解析更可靠。第二，Field支持更多约束，比如min_length、pattern、ge、le，这些约束会体现在 schema 里，模型在构造参数时能直接看到边界。
from pydantic import Field @mcp.tool() async def list_orders( user_id: str = Field(description="用户ID"), status: str = Field( default="all", description="订单状态过滤，可选值：pending/paid/shipped/completed/all", pattern="^(pending|paid|shipped|completed|all)$" ), page_size: int = Field( default=20, description="每页返回数量，范围1-100", ge=1, le=100 ) ) -> dict: ...这样模型在调用时就知道status只能填那几个值，page_size不能超过 100。实测下来，加了约束之后模型填错参数的概率明显下降。
4. 参数映射与返回结构裁剪：让模型看得懂
4.1 参数映射层：REST 和 MCP 之间的翻译
REST API 的参数风格和 MCP tool 的参数风格往往不一致。REST 喜欢用 query string 加 body 的组合，MCP tool 是一组扁平参数。中间需要一个映射层。
我一般会写一个独立的映射函数，把 MCP 参数转成 REST 请求需要的格式：
def build_order_query_params( user_id: str, status: str, page_size: int, page: int = 1 ) -> dict: """把 MCP 参数映射为 REST API 的 query 参数""" params = { "usr_id": user_id, "pg_sz": page_size, "pg_num": page, } if status != "all": params["ord_stat"] = status.upper() return params这样做的好处是：映射逻辑集中在一处，REST API 字段变了只改这一个函数；MCP tool 的参数定义保持稳定，模型侧无感知。
4.2 返回结构裁剪：别把原始 JSON 直接丢给模型
REST API 的返回往往包含大量模型不需要的字段。比如一个订单查询接口可能返回：
{ "code": 0, "message": "success", "data": { "order_id": "ORD123", "usr_id": "U456", "ord_stat": "PAID", "amt": 29900, "crt_tm": "2024-01-15T10:30:00Z", "ext_info": {...}, "internal_flags": {...}, "audit_log": [...] } }模型真正需要的可能只有订单号、状态、金额、时间。ext_info、internal_flags、audit_log这些字段不仅浪费上下文，还可能干扰模型判断。
裁剪的做法是写一个转换函数：
def normalize_order(raw: dict) -> dict: data = raw.get("data", {}) return { "order_id": data.get("order_id"), "status": data.get("ord_stat"), "amount_yuan": data.get("amt", 0) / 100, "created_at": data.get("crt_tm"), }注意这里把金额从分转成了元，字段名也改成了模型友好的形式。这种"翻译"是封装的核心价值之一。
4.3 分页信息的处理
列表类接口的分页信息怎么返回，也有讲究。REST API 通常返回total、page、page_size，模型需要知道"还有没有下一页"才能决定是否继续查。
我的做法是在返回结构里显式加一个has_more字段：
def normalize_order_list(raw: dict) -> dict: data = raw.get("data", {}) items = [normalize_order({"data": item}) for item in data.get("list", [])] total = data.get("total", 0) page = data.get("pg_num", 1) page_size = data.get("pg_sz", 20) return { "orders": items, "total": total, "page": page, "has_more": page * page_size < total }has_more这个字段看起来多余，但实测下来模型对它的理解比对total和page做算术要可靠得多。
5. 错误处理与重试：工业级封装的必修课
5.1 错误分类：哪些该抛给模型，哪些该吞掉
REST API 的错误五花八门，但不是所有错误都该原样抛给模型。我一般分三类处理：
第一类：参数错误（4xx）。这类错误应该抛给模型，因为模型可以修正参数重试。但错误信息要翻译成人话，不能直接把 REST 的报错丢过去。
第二类：服务端错误（5xx）。这类错误模型无能为力，重试也没用。应该吞掉细节，返回一个统一的"服务暂时不可用"提示，同时记录日志。
第三类：业务错误。比如"订单不存在"、"余额不足"，这类错误 REST API 通常用 200 状态码加错误码返回。需要解析错误码，翻译成模型能理解的描述。
class OrderNotFoundError(Exception): pass async def call_rest_api(method: str, url: str, **kwargs) -> dict: async with httpx.AsyncClient() as client: try: resp = await client.request(method, url, timeout=10.0, **kwargs) except httpx.TimeoutException: raise RuntimeError("调用后端服务超时，请稍后重试") except httpx.RequestError as e: raise RuntimeError(f"网络异常：{e}") if resp.status_code == 404: raise OrderNotFoundError("订单不存在，请确认订单ID是否正确") if resp.status_code >= 500: raise RuntimeError("后端服务暂时不可用，请稍后重试") if resp.status_code >= 400: raise RuntimeError(f"请求参数有误：{resp.text[:200]}") body = resp.json() if body.get("code") != 0: raise RuntimeError(f"业务错误：{body.get('message', '未知错误')}") return body5.2 重试策略：不是所有错误都值得重试
超时和 5xx 可以重试，4xx 重试没意义。重试次数我一般设 2 次，加上首次一共 3 次尝试。重试间隔用指数退避，避免雪崩。
import asyncio async def call_with_retry(method, url, max_retries=2, **kwargs): last_exc = None for attempt in range(max_retries + 1): try: return await call_rest_api(method, url, **kwargs) except (httpx.TimeoutException, httpx.RequestError) as e: last_exc = e if attempt < max_retries: await asyncio.sleep(2 ** attempt) raise RuntimeError(f"重试{max_retries}次后仍失败：{last_exc}")注意：重试只对幂等操作安全。GET 可以随便重试，POST 创建订单这种操作重试可能导致重复下单。写操作的重试要配合幂等键，或者干脆不重试。
5.3 错误信息的人话翻译
模型看到{"code": 40001, "message": "invalid param"}这种错误，基本没法自我修正。封装层要做的是把它翻译成"参数 user_id 格式不正确，应该是 U 开头的字符串"这种模型能直接用的信息。
我维护了一个错误码映射表：
ERROR_MAP = { 40001: "参数格式错误，请检查参数类型和格式", 40002: "缺少必填参数", 40003: "参数值超出允许范围", 40101: "认证失败，请检查API密钥", 40301: "没有权限执行此操作", 40401: "请求的资源不存在", 42901: "请求过于频繁，请稍后重试", } def translate_error(code: int, raw_msg: str) -> str: return ERROR_MAP.get(code, f"未知错误（代码{code}）：{raw_msg}")这张表是封装工作里最容易被忽略但价值最高的部分。花时间把常见错误码都翻译一遍，模型的自愈能力会明显提升。
6. 鉴权、限流与可观测性
6.1 鉴权信息的传递
MCP 服务调用 REST API 需要鉴权，鉴权信息从哪来？三种常见方案：
方案一：服务端固定密钥。MCP 服务配置里写死一个 API Key，所有调用都用这个。简单，但无法区分不同用户的权限。
方案二：透传用户身份。MCP 协议支持在请求里带上下文，可以把用户身份透传到 REST API。适合多租户场景，但实现复杂。
方案三：按 tool 配置不同密钥。不同 tool 用不同的密钥，权限隔离更细。适合工具权限差异大的场景。
我多数项目用方案一，因为 MCP 服务通常部署在内网，面向的是可信的模型调用方。如果面向多用户，方案二是必须的。
import os API_KEY = os.environ.get("REST_API_KEY") if not API_KEY: raise RuntimeError("缺少 REST_API_KEY 环境变量") HEADERS = {"Authorization": f"Bearer {API_KEY}"}密钥绝对不要硬编码在代码里，用环境变量或者配置中心。
6.2 限流：保护后端也保护模型
MCP 服务被模型调用时，调用频率可能很高。如果每个 tool 调用都直接打到 REST API，后端可能扛不住。需要在 MCP 层加限流。
简单的做法是用令牌桶：
import time from collections import deque class RateLimiter: def __init__(self, max_calls: int, window_seconds: float): self.max_calls = max_calls self.window = window_seconds self.calls = deque() async def acquire(self): now = time.monotonic() while self.calls and self.calls[0] < now - self.window: self.calls.popleft() if len(self.calls) >= self.max_calls: wait = self.calls[0] + self.window - now raise RuntimeError(f"调用过于频繁，请等待{wait:.1f}秒后重试") self.calls.append(now) limiter = RateLimiter(max_calls=10, window_seconds=1.0)在每个 tool 入口调用await limiter.acquire()。限流阈值根据后端承载能力设置，我一般从每秒 10 次开始，观察后端压力再调整。
6.3 日志：出问题时能查得清
MCP 服务的日志要记录四件事：谁调用了哪个 tool、传了什么参数、调用了哪个 REST 接口、返回了什么结果或错误。这四件事串起来，出问题才能定位。
import logging import json logger = logging.getLogger("mcp.order") @mcp.tool() async def get_order(order_id: str) -> dict: logger.info("tool_call", extra={ "tool": "get_order", "params": {"order_id": order_id} }) try: raw = await call_with_retry("GET", f"{REST_BASE}/api/orders/{order_id}", headers=HEADERS) result = normalize_order(raw) logger.info("tool_result", extra={"tool": "get_order", "result": result}) return result except Exception as e: logger.error("tool_error", extra={"tool": "get_order", "error": str(e)}) raise日志里不要记录敏感信息，比如完整的用户 token、身份证号。参数里的敏感字段要做脱敏。
7. 调试与验证：怎么确认封装是对的
7.1 用 MCP Inspector 做交互式调试
FastMCP 自带一个开发模式，可以启动一个 Inspector 界面，手动调用 tool 看返回。这是调试阶段最常用的工具：
fastmcp dev server.py启动后浏览器打开 Inspector，能看到所有注册的 tool，手动填参数调用，实时看返回。这个阶段重点验证三件事：参数能不能正确传到 REST API、返回结构是不是预期的、错误处理是不是按预期触发。
7.2 写单元测试覆盖映射逻辑
映射函数和裁剪函数是纯函数，最适合写单元测试。我一般会覆盖这几类用例：
def test_normalize_order(): raw = {"code": 0, "data": {"order_id": "ORD1", "ord_stat": "PAID", "amt": 29900}} result = normalize_order(raw) assert result["order_id"] == "ORD1" assert result["status"] == "PAID" assert result["amount_yuan"] == 299.0 def test_build_query_params_excludes_all_status(): params = build_order_query_params("U1", "all", 20) assert "ord_stat" not in params def test_build_query_params_includes_status(): params = build_order_query_params("U1", "paid", 20) assert params["ord_stat"] == "PAID"这些测试跑得快，能在重构时快速发现回归。
7.3 端到端验证：从模型侧真实调用
单元测试过了不代表模型用起来没问题。最终验证一定要从模型侧真实调用一遍。我会准备一组典型对话场景，比如"帮我查一下订单 ORD123 的状态"，看模型能不能正确选择 tool、填对参数、理解返回。
这一步经常能发现意料之外的问题。比如模型可能把order_id填成了"订单ORD123"这种带前缀的字符串，或者把status填成了中文"已支付"。发现这类问题后，要么在参数描述里加更明确的说明，要么在封装层做容错处理。
8. 部署与运维：让服务稳定跑起来
8.1 传输方式的选择
MCP 支持多种传输方式，常见的是 stdio 和 HTTP。stdio 适合本地集成，模型客户端直接启动 MCP 服务进程，通过标准输入输出通信。HTTP 适合远程部署，多个客户端共享一个服务。
选择依据：如果 MCP 服务和模型客户端在同一台机器，用 stdio 最简单，不用管端口和网络。如果服务要独立部署、多客户端共享，用 HTTP。
# stdio 模式 mcp.run(transport="stdio") # HTTP 模式 mcp.run(transport="http", host="0.0.0.0", port=8000)8.2 健康检查与优雅关闭
HTTP 模式下要加健康检查接口，方便运维监控：
@mcp.custom_route("/health", methods=["GET"]) async def health(request): return JSONResponse({"status": "ok"})优雅关闭也很重要。服务收到停止信号时，要等在途请求处理完再退出，避免模型调用到一半连接断开。
8.3 版本管理与向后兼容
MCP tool 的参数一旦暴露给模型，改动就要谨慎。模型可能缓存了旧的 tool 定义，参数改名或删除会导致调用失败。
我的做法是：新增参数用可选字段，不破坏现有调用；删除参数先标记 deprecated，保留一段时间再移除；tool 改名时保留旧名字作为别名，指向新实现。
@mcp.tool() async def get_order( order_id: str = Field(description="订单ID"), include_items: bool = Field(default=False, description="是否包含订单商品明细") ) -> dict: ...include_items这种新增的可选参数，老调用不受影响，新调用可以按需使用。
9. 几个我踩过的坑
第一个坑是参数类型不匹配。REST API 期望的page_size是整数，但模型有时会传字符串"20"。FastMCP 配合 Pydantic 会自动做类型转换，但如果 REST API 那边对类型敏感，最好在映射层再显式转一次。
第二个坑是返回内容过大。有个接口返回的列表默认 100 条，每条还有几十个字段，一次调用返回几万 token，直接把模型的上下文撑爆。后来改成默认返回 20 条，并且只保留必要字段，问题解决。列表类接口一定要控制默认返回量。
第三个坑是错误信息泄露内部细节。REST API 的报错里有时会带堆栈信息、数据库表名、内部 IP，这些直接抛给模型既不安全也没用。封装层一定要过滤，只保留对模型有用的信息。
第四个坑是tool 描述写得太简略。一开始我觉得参数名够清楚就不用写描述，结果模型经常选错 tool。后来给每个 tool 都写了详细的 docstring，说明这个 tool 做什么、什么时候用、和相邻 tool 的区别，模型的选择准确率明显提升。tool 描述是给模型看的"使用说明书"，值得花时间打磨。
第五个坑是忽略了并发。MCP 服务可能同时处理多个调用，如果用了全局的 httpx client 但没配连接池，高并发下会出问题。正确做法是用httpx.AsyncClient的上下文管理器，或者配置一个带连接池的全局 client。
_client = None async def get_client() -> httpx.AsyncClient: global _client if _client is None: _client = httpx.AsyncClient( timeout=10.0, limits=httpx.Limits(max_connections=50, max_keepalive_connections=10) ) return _client这套封装方案我在几个项目里跑下来，从最初的两三个 tool 扩展到几十个，整体稳定。核心经验就一条：把 MCP 封装当成一次面向模型的产品设计，而不是简单的协议转换。参数怎么命名、返回怎么裁剪、错误怎么翻译，每个决策都影响模型用起来顺不顺。多从模型的角度想"我看到这个 tool 会怎么用"，很多设计问题自然就有答案了。
