#MCP 实战：从 50 行 Server 到企业 API 封装

阅读提示 本笔记面向已经读完 MCP 协议原理、想动手写 MCP Server 的读者。主线是：先写出最小 Server → 客户端怎么连上它 → 把公司已有的 REST API 包成 MCP → 不走 MCP 手写 Tool 有什么不同 → 上线还差什么。

全部代码取自课程 mcp_examples/ 的真实文件，示例中标注了来源。

⚠️ 装依赖前必须锁版本：MCP Python SDK v2 改了名字 错误操作： pip install mcp langchain-mcp-adapters langchain httpx python-dotenv，不带版本约束。

实际结果： 装到 SDK v2 之后，本文所有代码第一行就崩：

ImportError: cannot import name 'FastMCP' from 'mcp.server.fastmcp'

原因： v2 是破坏性升级，且旧路径是直接移除而不是标记废弃：

v1	v2
from mcp.server.fastmcp import FastMCP	from mcp.server import MCPServer
mcp.server.fastmcp.*	mcp.server.mcpserver.*
属性 camelCase（result.isError、tool.inputSchema）	属性 snake_case（result.is_error、tool.input_schema）——线上 JSON 仍是 camelCase，只有 Python 属性名变了
类型定义在 mcp 里	拆成独立分发包 mcp-types，import mcp_types

正确做法： 本文代码基于 v1，装的时候把版本钉死：

pip install "mcp>=1.28,<2" langchain-mcp-adapters langchain httpx python-dotenv

v2 现在是什么状态（核验于 2026-08-13）：mcp 2.0.0 已于 2026-07-28 正式发布，对应新的 2026-07-28 规范；同日发布的 1.29.0 是 v1 主线，官方说明 v1.x 仍在单独分支上收关键 bug 修复和安全补丁。所以本文钉 v1 依然是可行选择，但新项目应当直接上 v2。

要用 v2 的话，装饰器风格的工具定义基本原样保留，但迁移不止是改导入名——官方把 v2 定性为 "a major rework of the SDK"，除了上表那几条，还要注意：

• 服务方式挪了位置：host、port、stateless_http、json_response、端点路径、transport security 这些不再挂在构造函数上，改由 run() 和 app builder 承接；
• 低层 Server 接口变化很大：handler 从装饰器改成构造函数参数，返回值不再自动包装；
• 同步函数改在 worker 线程跑：def 写的工具不再阻塞事件循环（以前 def 里做阻塞 IO 会卡死整个 Server）；
• 手动序列化必须带 by_alias=True——Python 属性是 snake_case，线上 JSON 仍是 camelCase，忘了这个参数发出去的报文对方解析不了。

照着官方迁移指南走一遍，别混着用。

#目录

• 一、50 行写出第一个 MCP Server
• 二、客户端的三种连法
• 三、把企业 REST API 封装成 MCP Server
• 四、对照路线：不走 MCP，手动把 REST 封装成 Tool
• 五、上线还差的五件事
• 六、实战陷阱
• 速查表
• 复习重点

核心概念 写 MCP Server 的工作量比想象中小：FastMCP + @mcp.tool() 装饰器，一个函数就是一个工具，docstring 就是给模型看的接口文档。

真正需要动脑的不是怎么写，而是暴露哪些能力、暴露到什么粒度、写操作怎么设闸。

⚠️ 两点范围说明：

1. 本文代码基于 MCP Python SDK v1（mcp>=1.28,<2）。v2 把 FastMCP 改名为 MCPServer、模块路径和属性命名全变，照抄会导入失败——装依赖时务必锁版本，详见下面的警告框。
2. 本文只讲 Tool。 一个完整的 MCP Server 还可以暴露 Resource（只读数据，由应用层决定何时塞进上下文）和 Prompt（标准化流程模板），Server 还能反过来用 Sampling / Elicitation 请求 Client 做事。这些见 MCP 协议原理。Tool 是兼容性最好、也最常用的一类，但它不等于 MCP 的全部。

#一、50 行写出第一个 MCP Server

先看完整的最小实现（来自 local_weather_server.py，去掉数据后就是全部骨架）：

from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("LocalWeather")          # 类似 app = FastAPI()
 
@mcp.tool()
def get_weather(city: str) -> str:
    """查询指定城市的天气信息。
 
    Args:
        city: 城市名称，如"北京"、"上海"、"深圳"
    """
    weather_db = {"北京": "晴，25°C，空气质量良好", "上海": "多云，28°C，有轻微雾霾"}
    return weather_db.get(city, f"暂无 {city} 的天气数据")
 
@mcp.tool()
def get_air_quality(city: str) -> str:
    """查询指定城市的空气质量。
 
    Args:
        city: 城市名称，如"北京"、"上海"
    """
    aqi_db = {"北京": "AQI 65 - 良好", "上海": "AQI 82 - 轻度污染"}
    return aqi_db.get(city, f"暂无 {city} 的空气质量数据")
 
if __name__ == "__main__":
    mcp.run()                           # 默认走 stdio

跑起来就一句 python local_weather_server.py。注意它不会打印任何东西——stdout 已经被协议占用了（原因见 MCP 协议原理 的陷阱一节）。

#三个关键点

FastMCP 与 FastAPI 的心智对应能省很多理解成本：

类型注解不是可选的。 city: str 会被 FastMCP 转成 JSON Schema 里的参数定义，再经桥接层变成 Pydantic 模型。不写注解，模型就不知道该传什么。

docstring 就是接口文档，而且读者是模型。 上面 get_weather 的 docstring 里那句「如"北京"、"上海"、"深圳"」不是写给人看的注释，它会原样进入 tools/list 的 description，模型靠它判断这个工具收的是中文城市名而不是拼音或 adcode。

docstring 的写法规则 按「这个工具做什么 + 什么时候该用它 + 每个参数的格式和取值示例」三段写。

反面例子："""查天气""" —— 模型不知道传城市名还是坐标，也不知道和另一个天气工具有什么区别。

正面例子就是上面代码里的写法：动作说清楚、参数给示例。

#二、客户端的三种连法

三段配置来自 mcp_demo.py，区别只在配置字典。

#连法一：本地 stdio

import os, asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
 
async def run():
    client = MultiServerMCPClient({
        "weather": {
            "command": "python",
            "args": [os.path.join(os.path.dirname(__file__), "local_weather_server.py")],
            "transport": "stdio",
        },
    })
    tools = await client.get_tools()
    print(f"加载了 {len(tools)} 个工具")           # 2 个
 
    agent = create_agent(model, tools=tools,
                         system_prompt="你是一个智能助手，请使用工具来回答用户问题。")
    result = await agent.ainvoke({"messages": [{"role": "user", "content": "上海天气怎么样？"}]})
    print(result["messages"][-1].content)
 
asyncio.run(run())

args 里用绝对路径拼接，不要写相对路径——子进程的工作目录不一定是你以为的那个。

#连法二：远程 SSE

换成 url 就行，其余代码一模一样：

client = MultiServerMCPClient({
    "amap": {
        "url": f"https://mcp.amap.com/sse?key={amap_key}",
        "transport": "sse",
    },
})
tools = await client.get_tools()      # 高德官方提供 11 个工具

高德官方 Server 提供的工具包括 amap_maps_geo（地理编码）、amap_maps_regeo（逆地理编码）、amap_maps_weather、amap_maps_search（POI 搜索）、amap_maps_around_search（周边搜索）、四种 amap_maps_direction_*（步行/驾车/骑行/公交路径规划）、amap_maps_distance、amap_maps_ip_location。

Key 从 高德开放平台 申请，服务平台必须选 Web 服务（选错类型申请出来的 Key 调不通）。

#连法三：本地 + 远程混合

一个 Client 管多个 Server，工具自动合并：

client = MultiServerMCPClient({
    "weather": {"command": "python", "args": [...], "transport": "stdio"},          # 本地
    "amap":    {"url": f"https://mcp.amap.com/sse?key={amap_key}", "transport": "sse"},  # 远程
})
tools = await client.get_tools()      # 2 + 11 = 13 个，扁平一个列表

模型看到的是一个 13 个工具的列表，不知道也不关心哪个来自哪个 Server。这既是便利也是风险，重名问题见 MCP 协议原理。

#三、把企业 REST API 封装成 MCP Server

这是 MCP 在公司里最常见的用法：已有系统一行代码不改，外面包一层 MCP，AI 就能用了。 下面以课程里的学生管理系统为例（student_management_mcp_demo.py），它是一个跑在 127.0.0.1:8000 的 FastAPI 服务。

三层的职责边界很清楚：业务系统不改、MCP 层是你唯一要新写的代码、Agent 层只负责决策。下面逐层看。

#第一层：API 客户端——把 HTTP 细节关在这里

class StudentManagementAPI:
    BASE_URL = "http://127.0.0.1:8000"
 
    def __init__(self):
        self._access_token: str | None = None
 
    def login(self, username: str, password: str) -> bool:
        """登录获取 JWT Token"""
        with httpx.Client(timeout=10.0) as client:
            resp = client.post(f"{self.BASE_URL}/auth/login",
                               json={"username": username, "password": password})
            result = resp.json()
            if result.get("status_code") == 1:
                self._access_token = result["data"]["access_token"]
                return True
            return False
 
    def _get_headers(self) -> dict:
        if not self._access_token:
            raise RuntimeError("未登录，请先调用 login() 方法")
        return {"Authorization": f"Bearer {self._access_token}"}
 
    def get_student(self, student_id: int) -> dict | None:
        with httpx.Client(timeout=10.0) as client:
            resp = client.get(f"{self.BASE_URL}/students/{student_id}",
                              headers=self._get_headers())
            if resp.status_code == 401:      # Token 过期
                return None
            result = resp.json()
            return result["data"] if result.get("status_code") == 1 else None
 
    def add_student(self, payload: dict) -> dict:
        """新增学生。返回业务层原始响应（含 status_code / message / data）。"""
        with httpx.Client(timeout=10.0) as client:
            resp = client.post(f"{self.BASE_URL}/students",
                               json=payload, headers=self._get_headers())
            if resp.status_code == 401:
                return {"status_code": 0, "message": "认证失败或 Token 已过期"}
            return resp.json()

⚠️ 别漏掉 API 层的方法 下一层的 add_student Tool 会调 student_api.add_student(payload)。如果 API 类里没定义这个方法（本笔记早先版本就漏了），Tool 本身能正常注册、tools/list 里也看得见，但模型一旦真的调用它，就会抛：

AttributeError: 'StudentManagementAPI' object has no attribute 'add_student'

而且这个错只在运行时、只在模型选中这个工具时才出现——写完 Server 手工测查询功能是完全测不到的。这正是上面说的「工具与 API 层要成对写、成对测」。

这一层的职责是把协议细节吸收掉：Token 存哪、Header 怎么拼、超时多久、401 和 403 分别代表什么。上层不该看到这些。

注意 status_code 是这个系统自己的业务状态码（1 表示成功），和 HTTP 状态码是两回事——企业内部 API 常见这种双层状态，封装时要区分开。

#第二层：MCP Tool——决定暴露什么，怎么说人话

from mcp.server.fastmcp import FastMCP
 
def create_mcp_server(student_api: StudentManagementAPI):
    mcp = FastMCP("学生管理系统")
 
    @mcp.tool()
    def query_student(student_id: int) -> str:
        """查询指定学生的详细信息，包括姓名、班级、专业、入学时间等。
        参数 student_id 为学生 ID（整数），如 1。
        """
        student = student_api.get_student(student_id)
        if not student:
            return f"未找到学号 {student_id} 的记录。请确认学号是否正确。"
        return (
            f"学生 {student.get('name', '未知')}（ID: {student.get('id')}）详情：\n"
            f"  班级：{student.get('class_id', '未分配')}\n"
            f"  专业：{student.get('major', '未登记')}\n"
            f"  入学时间：{student.get('enrollment_time', '未登记')}"
        )
 
    @mcp.tool()
    def add_student(name: str, class_id: int, major: str = "", education: str = "") -> str:
        """新增一名学生记录。必填参数：name（姓名）、class_id（班级 ID，整数）。
        可选参数：major（专业）、education（学历）。
        """
        payload = {"name": name, "class_id": class_id}
        for key, value in {"major": major, "education": education}.items():
            if value not in ("", 0):                    # 只提交填了的字段
                payload[key] = value
        result = student_api.add_student(payload)
        if result.get("status_code") == 1:
            new_id = result.get("data", {}).get("id", "未知")     # 逐层 get，别直接下标
            return f"✓ 成功添加学生 '{name}'，系统分配 ID: {new_id}。"
        return f"✗ 添加学生失败：{result.get('message', '未知错误')}"
 
    return mcp

这一层有三条设计原则，每条都有实际后果：

1. 一个 Tool 只做一件事（最小权限）。 学生管理系统可能有二十个接口，但只有查询和新增对 AI 有价值。不要为了省事写一个 call_api(path, method, body) 的万能工具——那等于把整个系统的写权限交给模型。

2. 参数要有文档、有默认值。 major: str = "" 加默认值，模型就知道这是可选的；docstring 里逐个说明参数含义，模型才填得对。

3. 给模型看的用自然语言，给程序用的走 structuredContent。

「返回自然语言而不是 JSON」这条流传很广的建议，只说对了一半。它成立的部分是：给模型读的那份文本，写成人话确实比 json.dumps() 更不容易被漏读、误读：

return json.dumps(student)          # 只有这一份的话，模型转述时容易漏字段、编造字段
return f"学生 {name}（ID: {id}）……"  # 模型直接引用，不需要二次解释

失败情况也一样——返回 "未找到学号 9999 的记录。请确认学号是否正确。" 比返回 None 或抛异常好得多，模型能直接把这句话转达给用户。

⚠️ 「MCP Tool 只能返回自然语言」已经不对了 过时的前提： 早期 MCP 的 CallToolResult 只有 content（文本/图片/资源），想传结构化数据只能塞进文本里让模型再解析一遍。所以那时候「返回人话」是唯一合理选择。

现在的情况： 协议已经支持 outputSchema（在 tools/list 里声明返回值的 JSON Schema）和 structuredContent（在 CallToolResult 里返回真正的结构化数据）。两者可以同时给：

• content → 给模型读的人话摘要；
• structuredContent → 给程序用的结构化数据，类型有保证、字段不会被模型改写。

为什么这很重要： 只返回自然语言时，下游想拿到学号就只能从模型转述的句子里再抽一次——模型完全可能把 ID 写错、把「未登记」写成 null。有了 structuredContent，程序直接读原始字段，绕开模型这一环。

正确做法： 声明 outputSchema 并返回结构化结果，同时保留一份人话摘要。SDK 里通常的写法是让工具函数返回一个带类型注解的模型/TypedDict，SDK 会自动生成 outputSchema 并填 structuredContent——具体 API 名随版本变化，查你装的版本的文档。

判断标准：这个返回值只是给模型念给用户听，还是下游代码要拿它做事？ 后者一律走 structuredContent。

#第三层：接入——两种方式，用途完全不同

方式 A：代码直调

r = asyncio.run(mcp_server.call_tool("query_student", {"student_id": 1}))
text = r[0][0].text        # ⚠️ 这是旧 SDK 的返回结构，见下方两个警告

⚠️ 进程内直调不是** MCP 集成，别拿它当验证** 错误操作： 写完 Server 后用 mcp_server.call_tool(...) 跑通了，就认为「MCP 集成没问题」。

实际结果： 这行代码只是在同一个进程里直接调用那个 Python 函数。它完全绕过了：

绕过了什么	后果
传输层（stdio / HTTP）	序列化问题、编码问题、stdout 污染全都测不出来
initialize 握手与能力协商	版本不兼容、能力没声明的问题测不出来
Session 与生命周期	会话状态、并发、断线重连测不出来
认证与授权	鉴权根本没跑过——本地全通，上线全挂
JSON-RPC 序列化	参数里有 datetime、Decimal、自定义对象这类不可序列化的东西，本地正常、走协议就炸

也就是说：本地全绿，换成真实客户端连可能一条都跑不通，而且失败点全在你没测过的那几层。

正确做法： 直调只适合当单元测试（验证业务逻辑对不对）。真正的集成验证必须走完整协议栈——用 mcp dev 起 Inspector 手工点一遍，或者用一个真实 Client（Claude Desktop、MultiServerMCPClient）连上去跑，至少覆盖握手、工具列表、一次调用、一次错误。

⚠️ r[0][0].text 是旧 SDK 的返回结构 问题： 这个下标写法既不可读，也不适用于当前的返回类型，更重要的是它把错误情况直接吞掉了。

现在 call_tool 返回的是一个 CallToolResult，该看的是三样东西：

result = await mcp_server.call_tool("query_student", {"student_id": 1})
 
if result.isError:                       # ① 先判错（v2 里写 result.is_error）
    raise RuntimeError(extract_text(result.content))
 
if result.structuredContent is not None: # ② 有结构化结果就优先用
    data = result.structuredContent
else:                                    # ③ 否则拼接文本内容
    data = "".join(c.text for c in result.content if getattr(c, "type", None) == "text")

三个要点：isError 必须判（工具执行失败时它是 True，而 content 里装的是错误信息——按老写法会把报错当成正常结果用）；content 是个列表，可能有多项、也可能含非文本项（图片、资源引用），别假定只有一个 text；structuredContent 优先（见上一节）。

方式 B：包成 LangChain Tool 交给 Agent

from langchain_core.tools import tool
 
@tool
def query_student(student_id: int) -> str:
    """查询学生详情。student_id 为学号（整数），如 1"""
    r = asyncio.run(mcp_server.call_tool("query_student", {"student_id": student_id}))
    return r[0][0].text
 
agent = create_agent(model, tools=[query_student, add_student], system_prompt=(
    "你是教务 AI 助手，可以访问学生管理系统。\n"
    "回复原则：\n"
    "1. 先查后说：涉及学生信息时，先调工具再回答\n"
    "2. 录入确认：新增学生时，确认关键信息（姓名、班级）后再调用\n"
    "3. 友好简洁：用自然语言回复，不要直接输出 JSON"
))

两种方式该怎么选：

混合用法 生产里常见的是两者叠加：确定性流程（每天导出报表）走方式 A，用户自由提问走方式 B，同一套 MCP Tool 复用。这正是把能力封成 MCP 而不是直接写死在 Agent 里的好处。

#四、对照路线：不走 MCP，手动把 REST 封装成 Tool

同样是「让模型用上已有 REST API」，还有一条不需要 MCP 的路：直接用 LangChain 的 @tool 装饰器包 requests 调用。课程里的 gaode_api_tool_demo.py 走的就是这条路——它不连高德官方 MCP Server，而是手动调高德 Web Service REST API。

from langchain_core.tools import tool
import requests
 
AMAP_BASE = "https://restapi.amap.com/v3"
 
def _amap_get(path: str, params: dict) -> dict:
    params["key"] = AMAP_KEY
    params["output"] = "JSON"
    return requests.get(f"{AMAP_BASE}/{path}", params=params, timeout=10).json()
 
@tool
def amap_geo(address: str) -> str:
    """地理编码：将地址转换为经纬度坐标。"""
    data = _amap_get("geocode/geo", {"address": address})
    if data.get("status") != "1" or not data.get("geocodes"):
        return f"未找到地址「{address}」的坐标信息"
    g = data["geocodes"][0]
    return f"地址: {g['formatted_address']}\n坐标: {g['location']} (经度,纬度)"

结构和 MCP 版几乎一样：一个函数、一段 docstring、返回人话。区别只是装饰器从 @mcp.tool() 换成了 @tool，少了一层进程/HTTP 通信。

#Skill 的雏形：组合多个 API 的那个工具

同一个文件里的 amap_travel_planner 值得单独看，它在一个 Tool 内部串了三类 API：

@tool
def amap_travel_planner(city: str, interests: str = "景点,美食", route_type: str = "walking") -> str:
    """智能旅游规划：搜索城市兴趣点并规划游览路线。"""
    # 1. 地理编码拿城市中心坐标
    geo_data = _amap_get("geocode/geo", {"address": city})
    location = geo_data["geocodes"][0]["location"]
 
    # 2. 按兴趣类型逐个搜 POI
    all_pois = []
    for t in [x.strip() for x in interests.split(",")]:
        poi_data = _amap_get("place/text", {"keywords": t, "city": city, "offset": 3})
        for p in poi_data.get("pois", [])[:2]:
            all_pois.append({"name": p["name"], "location": p["location"]})
 
    # 3. 相邻景点之间算步行路线
    for i in range(min(len(all_pois) - 1, 3)):
        route = _amap_get("direction/walking",
                          {"origin": all_pois[i]["location"], "destination": all_pois[i+1]["location"]})
        # …拼装成一段文字返回

这就是 Skill（技能）的雏形：模型只看到「旅游规划」一个工具，内部的三步编排被封在函数里。它和另一种做法——把八个原子 Tool 都给模型、靠 system_prompt 引导它自己串——是同一个问题的两种解法：

#MCP 封装 vs 手动 @tool：怎么选

选择规则

1. 只有你自己这一个 Python 程序要用 → 手动 @tool，别引入 MCP 的复杂度。
2. 要给 Claude Desktop / Cursor / 别的团队用 → MCP，这是它存在的理由。
3. 工具需要独立部署、独立扩容（比如它很重、或者要放在能访问内网的机器上）→ MCP。
4. 还在探索阶段、需求天天变 → 先手动 @tool 跑通，稳定后再包成 MCP。接口签名不变的话，迁移成本很低。

#五、上线还差的五件事

Demo 能跑不等于能上线。按重要性排：

1. 认证鉴权。 Demo 里 Token 是登录时存在 StudentManagementAPI 实例变量 _access_token 上的，生产要考虑 Token 过期刷新、多用户隔离。别把一个管理员账号的 Token 写死给所有人用。

⚠️ Demo 那个 _access_token 在多用户下会串身份 StudentManagementAPI 通常是跟着 Server 进程活一辈子的单例，_access_token 是它的实例属性。这意味着所有请求共用同一个 Token：A 用户登录后写进去，B 用户的请求就带着 A 的身份打到后端——不报错，数据照返回，只是返回的是别人的权限范围。stdio 本地单人用没问题，一旦挂成 HTTP 多人共用，这就是一个越权漏洞。

正确做法： 凭据必须按请求/按会话隔离，不能挂在长生命周期对象上——每次调用从当前请求的认证上下文取，或者为每个用户单独建客户端实例。

下面这种写法是反例，它把凭据做成了工具参数——先记住结论：认证信息永远不进模型的视野，完整原因见本节末尾那个 auth_token 的 danger 框：

@mcp.tool()
def query_student(student_id: int, auth_token: str) -> str:   # ❌ 别这么写
    if not verify_token(auth_token):
        return "认证失败，请重新登录"
    ...

2. 按角色控权。 学生管理系统里查询要 user 角色、新增要 operator 角色，这个约束必须在 Tool 层再检查一遍——不能指望模型不去调它不该调的工具。

3. 日志审计。 每次 Tool 调用记录「谁、什么操作、什么参数、什么结果」。AI 调用出问题时，这是唯一能复盘的东西。

logging.info(f"Tool call: query_student(student_id={student_id}) by {user}")

4. 输入校验。 模型生成的参数可能不合规——姓名空串、日期格式乱、班级 ID 不存在。用 Pydantic 或手工校验都行，但必须校验。

5. 速率限制。 Agent 循环可能在几秒内连发几十次调用（尤其是它「想不明白」反复重试的时候），令牌桶或滑动窗口挡一下，别让 AI 把内部系统打垮。

#两种部署形态

stdio 本地部署（开发、个人使用）——Client 和 Server 同机，配置写进客户端：

{
  "mcpServers": {
    "student-system": {
      "command": "python",
      "args": ["student_mcp_server.py"]
    }
  }
}

Streamable HTTP 远程部署（生产）——Server 部署在能访问内网的机器上，走 HTTPS：

{
  "mcpServers": {
    "student-system": {
      "url": "https://api.school.com/mcp/students",
      "transport": "streamable_http",
      "headers": {"Authorization": "Bearer <token>"}
    }
  }
}

⚠️ 新部署不要再选 "transport": "sse" 旧的 HTTP+SSE 传输已被 Streamable HTTP 取代（见 四、stdio、SSE、Streamable HTTP 该选哪个？）。SSE 需要维护收发两个通道、断线重连麻烦，Streamable HTTP 单端点就够了。只有在连别人已经部署好的老 SSE 端点时才用 sse。

上面那个 headers 里塞静态 Bearer token 的写法，只适合内网、单租户、你自己控制两端的场景。对外提供服务时应当走规范定义的 OAuth 2.1 授权流程（授权码 + PKCE、动态注册、元数据发现），并且 Server 必须校验 token 的受众是自己——不能接受一个签发给别处的 token，也不能把收到的 token 原样转手去调下游（token passthrough 是被规范明令禁止的）。

⚠️ 把 auth_token 做成模型可见的 Tool 参数 前面权限那节出现过这种写法：

@mcp.tool()
def query_student(student_id: int, auth_token: str) -> str:   # ❌
    if not verify_token(auth_token):
        return "认证失败"

实际结果： 一旦 auth_token 成为工具参数，它就会出现在 inputSchema 里，于是：

• 模型必须知道这个 token 才能调用——也就是说 token 必然出现在提示词或上下文里，等于把凭据交给了模型；
• token 会被写进对话历史、日志、trace，泄露面骤增；
• 模型可以随便编一个——它会尝试填 "your_token_here"、复用它在别处见过的字符串，甚至从上下文里翻出别的用户的 token；
• 提示注入攻击可以直接诱导模型把 token 打印出来。

正确做法：认证信息永远不进模型的视野。 它应该来自：

• 传输层——HTTP 的 Authorization 头 / OAuth 流程拿到的 token，由 Client 和 Server 之间处理；
• 进程环境——stdio 场景下由 Host 通过环境变量注入；
• 然后在 Server 内部从可信的认证上下文里取，而不是从工具参数里取。

工具签名里应该只有业务参数：

@mcp.tool()
def query_student(student_id: int) -> str:      # ✅ 只有业务参数
    user = current_auth_context()               # 从可信上下文取，模型看不见也改不了
    if not user.can("student:read"):
        return "无权限执行该操作"

#六、实战陷阱

⚠️ 写操作 Tool 不设闸就交给 Agent 错误操作： 把 add_student、delete_order 这类写操作和查询工具一起丢给 Agent，靠 system_prompt 里写一句「录入前请确认」来约束。

实际结果： 模型误解意图时直接写库。用户说「这个学生信息不太对」，模型可能理解成要新建一条正确的记录，于是产生一条脏数据——而且它会礼貌地告诉你「已为您添加」。

原因： system_prompt 是建议不是约束。模型对写操作和读操作一视同仁，都是「可以调的工具」，没有任何机制阻止它调。

正确做法： 写操作的闸门必须设在 Tool 内部，不能设在提示词里。可选方案按可靠性排序：

1. 不把写操作暴露给自由对话的 Agent——只在确定性流程里用方式 A 直调。最彻底。
2. 真正的人工审批（Human-in-the-loop）——Tool 里挂起，把待执行动作交给Host / 应用层去问用户，拿到用户的批准凭据后再执行。关键在于「批准」这个信号来自模型之外。
3. 限权 + 限额——按角色控制谁能调、加额度上限、写操作走幂等键，把最坏后果控制住。

⚠️ 用 confirmed: bool = False 参数当审批闸门 错误操作： 给写操作加一个确认参数，以为这样就有了二次确认：

@mcp.tool()
def add_student(name: str, class_id: int, confirmed: bool = False) -> str:  # ❌
    if not confirmed:
        return f"即将添加学生 {name}，请确认"
    ...   # 真正写库

实际结果：这个闸门形同虚设——因为参数是模型自己填的。 模型完全可以第一次就传 confirmed=True，或者在收到「请确认」之后自己判断「用户刚才说要加，那就是确认了」，随即再调一次并传 True。整个过程中真实用户可能一次都没被问到，但日志上看起来完整走了两步确认流程——这比没有确认更危险，因为它制造了「已确认」的假象。

原因： 审批的本质是从模型之外获得一个授权信号。任何放在 inputSchema 里的字段，都在模型的控制范围内，因此不可能充当这个信号。这和上面 auth_token 那个坑是同一个错误：把安全控制项交给了被控制的一方。

正确做法： 确认必须发生在协议/应用层，而不是参数层。三条路：

• 应用层拦截：Host 或 Agent 框架在真正执行工具前挂起，把动作和参数展示给用户，等一个真实的用户输入（LangGraph 的 interrupt 就是干这个的，见 八、持久化与人工审批为什么必须成对出现？）；
• MCP 的 Elicitation：由 Server 主动向用户发起确认请求——这条请求由 Client 呈现给真人，模型无法代答；
• 带外凭据：真实确认后由应用层生成一个一次性 token，Tool 校验该 token 而非某个 bool 参数。这个 token 不能出现在模型上下文里。

判断标准很简单：问一句「模型能不能自己伪造这个信号？」 能，它就不是审批。

⚠️ Tool 直接返回原始 JSON 错误操作： return json.dumps(student_data, ensure_ascii=False)，觉得信息全给了模型总没错。

实际结果： 模型转述给用户时漏字段、把 class_id: 3 说成「三班」（实际可能是「2023 级 3 班」）、字段为 null 时编一个看起来合理的值。

原因： 模型拿到的是没有语义的键值对，它必须猜每个字段是什么意思、null 该怎么表达。猜错不会报错。

正确做法： 在 Tool 里就把 JSON 翻成带字段名的中文文本，空值显式写成「未登记」而不是留空或 null。模型只需转述，不需要解释。

⚠️ 在同步 @tool 里用 asyncio.run() 错误操作： 像上面方式 B 那样，在同步的 @tool 函数里写 asyncio.run(mcp_server.call_tool(...))。

实际结果： 单独跑测试没问题；一旦放进 FastAPI 接口、Jupyter，或改用 agent.ainvoke()，就抛 RuntimeError: asyncio.run() cannot be called from a running event loop。

原因： asyncio.run() 要求当前线程没有正在运行的事件循环。Web 框架和异步 Agent 本身就跑在事件循环里，再调一次就冲突。

正确做法： 整条链路统一成异步——用 async def 定义工具函数、await mcp_server.call_tool(...)、await agent.ainvoke(...)。教学 demo 里的 asyncio.run() 写法只在纯同步脚本里成立，别原样搬进服务。

⚠️ Tool 的名字和描述含糊 错误操作： 工具叫 search，docstring 写 """搜索"""。同一个 Agent 里还有另一个 query。

实际结果： 模型有时调对、有时调错、有时两个都调一遍。全程不报任何错，只是答案偶尔莫名其妙。这类问题极难定位，因为日志里一切正常。

原因： 模型选工具的全部依据就是 name + description（见 MCP 协议原理 第三节）。两个描述接近的工具，对模型来说就是两个几乎等价的选项。

正确做法： 名字带领域前缀（student_query 而非 query）；描述里写清楚什么时候该用它、什么时候不该用；工具超过 10 个时，检查是否有语义重叠的，能合并就合并。

#速查表

#复习重点

复习重点

1. 一句话说清封装工作：FastMCP + @mcp.tool()，函数签名生成参数 Schema，docstring 生成给模型看的说明，业务系统一行不改。
2. 最容易做错的设计决策：为了省事写一个万能的 call_api(path, method, body) 工具。这等于把整个系统交给模型，正确做法是一个 Tool 一个能力、按最小权限暴露。
3. 最容易被忽视的返回值：Tool 返回裸 JSON 会让模型猜字段含义并编造缺失值；返回带字段名的中文文本，模型只需转述。
4. 最容易在生产爆炸的写法：同步 @tool 里套 asyncio.run()，在任何已有事件循环的环境里都会抛 RuntimeError。
5. 两种接入方式的分工：确定性流程用 call_tool() 直调，自由对话交给 Agent 自选；两者可以复用同一套 Tool。
6. 要不要上 MCP 的判断：只有自己一个程序用就手动 @tool；要给别的客户端/团队用、或工具需要独立部署，才值得付出跨进程调试的代价。

一句话总结 把企业 API 包成 MCP，技术上是十几行装饰器的事，真正的工作量在于想清楚暴露哪些能力、写操作在哪里设闸、失败时对模型说什么话。前者是体力活，后者决定这套东西能不能上生产。

协议层面的机制回看 MCP 协议原理；Agent 本身的循环和组件分层见 Agent概念、Agent架构设计-lang系框架。