第 12 课：MCP 协议集成 —— 让你的 Agent 被 Cursor/Claude Desktop 调用

本节目标：理解 MCP 协议的核心概念、掌握 ToolRegistry → MCP Server 的三层架构、实现风险三级安全模型，以及让外部 AI（Cursor、Claude Desktop）即插即用地调用你的 Agent。

学完本课后，推荐阅读 附12：MCP 与 A2A 协议深度 —— MCP 核心架构与三大能力、MCP vs Function Calling 本质区别、A2A Agent 间协作协议。

前 11 课 Agent 只能在自己的 API 里工作。但外部 AI（Cursor、Claude Desktop）不认识你的 API → 无法调用你的 Agent 当工具。MCP 改变了这一点。

1. 什么是 MCP？

MCP = Model Context Protocol（模型上下文协议），由 Anthropic 提出，已成为 AI 工具生态的事实标准。

Code

传统模式：
  你的 Agent ──(自定义 API)──→ 外部工具
  外部 AI ──(不认识你的 API)──→ 无法调用你的 Agent

MCP 模式：
  你的 Agent ──(MCP 标准)──→ 外部工具
  外部 AI ──(MCP 标准)──→ 直接调用你的 Agent 当工具！

角色	谁扮演	做什么
MCP Client	Cursor / Claude Desktop / Windsurf	发现工具、调用工具
MCP Server	你的 Knowledge Agent	暴露工具列表、接收调用、返回结果

类比：MCP 就像 USB 协议——USB 之前每个设备自己的接口（并口、PS2），USB 之后统一一个口，插啥都能用。

2. MCP 的核心概念

协议三要素：

code

1. Tools（工具）  —— Agent 能做什么      名字 + 描述 + 参数 Schema + 执行
2. Resources（资源）—— Agent 有什么数据    URI + MIME type + 内容
3. Prompts（提示模板）—— 推荐的交互方式    名字 + 参数 + 模板内容

课程聚焦 Tools——最核心、最实用的部分。

工具暴露的标准格式：

json

{
  "tools": [{
    "name": "search_knowledge_base",
    "description": "在公司内部知识库中搜索相关内容...",
    "input_schema": {
      "type": "object",
      "properties": {
        "query": {"type": "string", "description": "搜索关键词"},
        "top_k": {"type": "integer", "default": 3}
      },
      "required": ["query"]
    }
  }]
}

和 OpenAI Function Calling 的 schema 几乎一致——MCP 就是在 Function Calling 基础上标准化的。

3. 三层架构

Code

┌─────────────────────────────────────────────────────┐
│              MCP Server（我们的系统）                  │
│                                                     │
│  GET  /api/v1/mcp/tools          → 列出可用工具       │
│  POST /api/v1/mcp/tools/{name}/invoke → 调用工具      │
│                                                     │
│       ┌──────────────┐                              │
│       │ tool_registry │ ← 所有工具的 source of truth │
│       └──────┬───────┘                              │
│    ┌─────────┼─────────────┐                        │
│  LOW 工具  LOW 工具    HIGH 工具                      │
│  暴露    暴露     不暴露                     │
└─────────────────────────────────────────────────────┘

第 1 层：ToolDefinition —— 工具的身份证

python

@dataclass
class ToolDefinition:
    name: str                                    # 唯一标识
    description: str                             # 告诉 AI "什么时候用我"
    parameters: dict[str, Any]                   # JSON Schema 定义输入
    func: Callable[..., Any] = field(repr=False) # 实际执行函数
    risk: RiskLevel = RiskLevel.LOW              # 出厂属性，不准运行时改
    tags: list[str] = field(default_factory=list)

    def to_openai_schema(self) -> dict[str, Any]:
        """转成 OpenAI 兼容的 tools 元素."""
        return {"type": "function", "function": {
            "name": self.name, "description": self.description,
            "parameters": self.parameters}}

第 2 层：ToolRegistry —— 装饰器即注册

python

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, ToolDefinition] = {}

    def register(self, *, name, description, parameters,
                 risk=RiskLevel.LOW, tags=None):
        """装饰器：定义即注册."""
        def _decorator(func):
            self._tools[name] = ToolDefinition(
                name=name, description=description, parameters=parameters,
                func=func, risk=risk, tags=list(tags or []))
            return func
        return _decorator

python

@tool_registry.register(
    name="calculate",
    description="当用户需要做数学运算时调用...",
    parameters={"type":"object","properties":{"expression":{"type":"string"}},"required":["expression"]},
    risk=RiskLevel.LOW,
)
def calculate(expression: str) -> str: ...

定义即注册——写函数的同时完成注册，不需要单独维护工具列表。

第 3 层：MCP Server 路由

Code

@router.get("/tools", summary="MCP：列举对外可用工具（过滤 HIGH）")
def list_tools() -> dict[str, list[dict[str, Any]]]:
    tools = []
    for t in tool_registry.list_all():
        if t.risk == RiskLevel.HIGH:
            continue  # HIGH 工具永不对外暴露
        tools.append({
            "name": t.name, "description": t.description,
            "input_schema": t.parameters, "risk": t.risk.value,
            "tags": list(t.tags),
        })
    return {"tools": tools}

@router.post("/tools/{name}/invoke", summary="MCP：调用工具")
def invoke_tool(name: str, payload: dict[str, Any]) -> dict[str, Any]:
    tool = tool_registry.get(name)  # KeyError → 404
    if tool.risk == RiskLevel.HIGH:
        raise HTTPException(403, detail=f"tool {name!r} is HIGH-risk")
    result = tool_registry.execute(name, payload.get("arguments", {}))
    return {"result": result, "risk": tool.risk.value}

两个 endpoint，极简：GET /tools 列出能力，POST /tools/{name}/invoke 执行能力。

4. 安全设计：风险三级

python

class RiskLevel(str, Enum):
    LOW = "low"        # 只读、无副作用 — 查时间、算数、搜索
    MEDIUM = "medium"  # 有副作用但可逆 — 修改草稿、更新标签
    HIGH = "high"      # 不可逆、对外通信 — 删文档、发邮件

等级	例子	MCP 暴露？	Agent 内部？
LOW	查时间、算数、搜索	直接暴露	自动执行
MEDIUM	修改草稿、更新标签	注意：需要认证	注意：需确认
HIGH	删文档、发邮件	禁止：永不暴露	禁止：必须人审

纵深防御：

python

# 第 1 道：list_tools 时过滤 — 列表里看不到
if t.risk == RiskLevel.HIGH: continue

# 第 2 道：invoke 时拒绝 — 知道名字也调不了
if tool.risk == RiskLevel.HIGH: raise HTTPException(403)

# 第 3 道：Agent 内部也需要人审（Ch14 审批流）
if risk == RiskLevel.HIGH: create_approval_request(...)

三道防线：看不到 → 调不了 → 内部也要审批。

5. 工具注册的内置工具 + 安全计算

5 个内置工具：

工具	风险	功能
`get_current_time`	LOW	拿当前时间
`calculate`	LOW	算数学表达式（AST 白名单，不用 eval）
`search_knowledge_base`	LOW	检索知识库
`delete_document`	HIGH	删除知识库文档（不可逆）
`send_email`	HIGH	对外发邮件（不可逆/对外通信）

安全计算：AST 白名单而不是 eval：

Code

# eval — LLM 可能注入恶意代码
eval("__import__('os').system('rm -rf /')")  # 

# AST 白名单 — 只允许数字和四则运算
_safe_eval(ast.parse("1+2*3", mode="eval"))  # → 7
_safe_eval(ast.parse("__import__('os')", mode="eval"))  # → ValueError!

execute 的兜底设计：

python

def execute(self, name: str, arguments) -> str:
    """任何异常都不抛出，统一以 [ERROR] ... 字符串回注 LLM."""
    try:
        tool = self.get(name)
    except KeyError as e:
        return f"[ERROR] {e}"

    # 容忍 LLM 多传字段：只挑 schema 声明的参数进去
    sig = inspect.signature(tool.func)
    allowed = set(sig.parameters.keys())
    filtered = {k: v for k, v in kwargs.items() if k in allowed}

    try:
        result = tool.func(**filtered)
        text = str(result) if not isinstance(result, str) else result
        return text[:8000] if len(text) > 8000 else text  # 截断防撑爆 context
    except Exception as e:
        return f"[ERROR] tool '{name}' failed: {e}"

四个防御：参数容错（LLM 常多传参数）、异常不抛（[ERROR] 回注 LLM）、结果截断（≤8000 字符）、格式兼容（JSON 字符串和 dict 都接受）。

6. MCP 和 Agent 内部工具的关系

Code

┌──────────────────────────────────────────┐
│          tool_registry (唯一真相源)        │
│  get_current_time │ calculate │ search_kb │
│  delete_document  │ send_email            │
└───────┬──────────────────┬───────────────┘
        │                  │
   ┌────┴─────┐      ┌─────┴──────┐
   │ MCP 对外  │      │ Agent 内部  │
   │ 过滤 HIGH │      │ HIGH→审批   │
   │ 只暴露 LOW│      │ LOW→自动执行│
   └──────────┘      └────────────┘

同一个 tool_registry，两种使用方式——工具定义一次，两处都能用。

7. 测试 MCP

python

@pytest.mark.unit
def test_mcp_list_excludes_high_risk(client: TestClient) -> None:
    r = client.get("/api/v1/mcp/tools")
    assert r.status_code == 200
    names = {t["name"] for t in r.json()["tools"]}
    assert "get_current_time" in names
    assert "delete_document" not in names      # HIGH 不可见
    assert "send_email" not in names           # HIGH 不可见

@pytest.mark.unit
def test_mcp_invoke_low_tool(client: TestClient) -> None:
    r = client.post("/api/v1/mcp/tools/calculate/invoke",
                    json={"arguments": {"expression": "1+2"}})
    assert r.status_code == 200
    assert "3" in r.json()["result"]

@pytest.mark.unit
def test_mcp_invoke_high_tool_forbidden(client: TestClient) -> None:
    r = client.post("/api/v1/mcp/tools/delete_document/invoke",
                    json={"arguments": {"doc_id": "x"}})
    assert r.status_code == 403

@pytest.mark.unit
def test_mcp_invoke_unknown_tool_404(client: TestClient) -> None:
    r = client.post("/api/v1/mcp/tools/no_such_tool/invoke", json={})
    assert r.status_code == 404

覆盖四条关键路径：列表过滤 HIGH / LOW 正常调用 / HIGH 被 403 / 未知工具 404。

8. 连接 Cursor / Claude Desktop

Cursor 配置（.cursor/mcp.json）：

json

{
  "mcpServers": {
    "knowledge-agent": {
      "url": "http://localhost:8000/api/v1/mcp",
      "transport": "streamableHttp"
    }
  }
}

Claude Desktop 配置（claude_desktop_config.json）：

json

{
  "mcpServers": {
    "knowledge-agent": {
      "command": "npx",
      "args": ["-y", "mcp-remote", "http://localhost:8000/api/v1/mcp"]
    }
  }
}

连接后，用户在 Cursor 里问"公司请假流程是什么？"→ Cursor 自动调用 search_knowledge_base → Agent 检索知识库 → 返回结果 → Cursor 用结果回答。你的 Agent 变成了 Cursor 的知识库插件！

9. 5 条工程教训

教训 1：工具描述决定 AI 会不会调用

python

# 差的描述
description="搜索知识库"
# 好的描述 — 写"什么时候用"和"什么时候不用"
description="在公司知识库中搜索。当用户询问内部流程、规章、产品文档时使用。公开知识不要调用。"

教训 2：风险等级是"出厂属性"，不准运行时改 — 写在代码里 = 需要 code review 才能改 = 安全审计可追溯。

教训 3：execute 永远不抛异常 — 异常变成 [ERROR] ... 字符串回注 LLM，LLM 理解后换方式回答。不导致整个请求崩溃。

教训 4：MCP 是对外边界——必须有独立的安全层 — 外部 Client 可能没认证、可能恶意。不暴露 HIGH、结果截断、Rate limit。

教训 5：to_openai_schema() 保证一致性 — 同一个 ToolDefinition：对内 to_openai_schema() → OpenAI Function Calling；对外 input_schema → MCP Client。定义一次两处使用，永远一致。

10. 小练习（5 题）

练习 1：如果要新增一个 MEDIUM 风险的工具 update_document_tags（打标签，有副作用但可逆），MCP 应该暴露它吗？当前代码需要怎么改才能支持 MEDIUM 的差异化处理？

练习 2：execute() 用 inspect.signature(tool.func) 过滤 LLM 多传的参数。如果工具函数声明了 **kwargs，allowed 集合会包含什么？会导致问题吗？

练习 3：MCP 的 list_tools 返回的 input_schema 和 OpenAI 的 to_openai_schema() 里的 parameters 是同一个 dict 对象。如果 MCP Client 修改了返回的 schema，会影响 Agent 内部的 Function Calling 吗？

练习 4：当前 MCP Server 没有认证。如果要给它加认证（只允许持有 API Key 的 Client 调用），在 FastAPI 里最小改动方案是什么？

练习 5（最重要）：假设 Cursor 连接了你的 MCP Server，用户问"帮我删除文档 X"。Cursor 会看到 delete_document 吗？如果用户坚持要删，整个交互链路是什么？

答案与解析

练习 1：MEDIUM 工具的 MCP 暴露策略

当前代码只区分 "HIGH vs 非 HIGH"，MEDIUM 和 LOW 一样直接暴露。生产版需要差异化：

python

# list_tools 里加标记
tools.append({
    ...,
    "requires_confirmation": t.risk == RiskLevel.MEDIUM,  # Client 可展示确认 UI
})

# invoke 里加审计日志
if tool.risk == RiskLevel.MEDIUM:
    logger.info("[mcp-audit] MEDIUM tool %s invoked by %s", name, user)

策略矩阵：LOW → 直接执行；MEDIUM → 标记 requires_confirmation；HIGH → 不可见 + 403。

练习 2：`**kwargs` 对 `inspect.signature` 的影响

inspect.signature(get_current_time) 得到 allowed = {"_ignored"}，但 _ignored 是 **kwargs 参数（VAR_KEYWORD）。LLM 传的字段名（如 {"random_field": "x"}）不匹配 "_ignored" → 被过滤掉 → 函数被调用时无参数 → 正常工作。

但注意：如果写成 def bad_tool(**kwargs)（参数名是 kwargs），allowed = {"kwargs"} → LLM 传的 {"query": "hi"} 会被过滤 → KeyError。正确做法：对 VAR_KEYWORD 参数跳过过滤。

练习 3：schema 被修改会影响内部吗？

跨网络不会——FastAPI 返回 JSON 时 t.parameters 被序列化成 JSON 字符串发出去，Client 收到的是新 dict 对象，和 Server 进程内存无共享。

但进程内测试会——直接拿到 dict 引用，修改会污染原始对象。防御：copy.deepcopy(t.parameters) 或 Pydantic model 序列化。

练习 4：给 MCP 加认证的最小改动

python

# 新建 app/mcp/auth.py
def verify_mcp_api_key(x_mcp_key: str = Header(..., alias="X-MCP-Key")) -> str:
    if x_mcp_key not in {"mcp-key-abc123"}:  # 实际从 env/db 读
        raise HTTPException(401, detail="Invalid MCP API Key")
    return x_mcp_key

# router 上加一行
router = APIRouter(prefix="/mcp", tags=["mcp"],
                   dependencies=[Depends(verify_mcp_api_key)])

只两处改动：新建 dependency + router 加 dependencies=[...]。所有 MCP 路由自动被保护。

练习 5（最重要）：Cursor 里删除文档的完整链路

Step 1：Cursor 调 GET /mcp/tools → 返回只有 LOW 工具 → delete_document 不在列表里 → Cursor 的 LLM 看不到。

Step 2：Cursor 的 LLM 告诉用户"我只有搜索能力，无法删除文档"。

如果用户坚持要删——正确路径是通过内部 Agent 审批流：

Code

POST /api/v1/agent/ask {"question": "删除文档 policy-2024"}
→ Agent 内部 LLM 看到 delete_document（看到所有工具）
→ 决定调用 delete_document(doc_id="policy-2024")
→ risk = HIGH → 不自动执行
→ 创建 ApprovalRequest → Agent 返回 {"status": "paused"}
→ 管理员审批 → Agent resume → 执行删除

Code

┌─── Cursor (MCP Client) ───┐   ┌─── 直接 Agent ───────────┐
│ 看到工具：3 个 LOW         │   │ 看到工具：5 个（含 HIGH） │
│ 能做：查询、计算           │   │ HIGH → 审批流 → 人工批准 │
│ 安全边界：MCP 过滤         │   │ 安全边界：HITL 审批      │
└────────────────────────────┘   └──────────────────────────┘

设计哲学：MCP 过滤 + Agent 审批 = 纵深防御。外部看不到高危工具（攻击面收窄），内部不能自动执行高危操作（人工兜底）。单点都不够安全。

三句话带走第 12 课

MCP 让你的 Agent 变成"USB 设备"——任何 MCP Client（Cursor、Claude Desktop）即插即用地调用你暴露的工具。核心就 GET /tools + POST /tools/{name}/invoke。
风险三级是安全的骨架——LOW 直接暴露、MEDIUM 标记确认、HIGH 永不暴露。同一个 tool_registry 对内对外两种策略，定义一次保证一致。
MCP 过滤 + Agent 审批 = 纵深防御——外部看不到高危工具（攻击面收窄），内部也不能自动执行高危操作（人工兜底）。两层配合才行。

第 12 课：MCP 协议集成 —— 让你的 Agent 被 Cursor/Claude Desktop 调用

1. 什么是 MCP？

2. MCP 的核心概念

3. 三层架构

第 1 层：ToolDefinition —— 工具的身份证

第 2 层：ToolRegistry —— 装饰器即注册

第 3 层：MCP Server 路由

4. 安全设计：风险三级

5. 工具注册的内置工具 + 安全计算

6. MCP 和 Agent 内部工具的关系

7. 测试 MCP

8. 连接 Cursor / Claude Desktop

9. 5 条工程教训

10. 小练习（5 题）

答案与解析

练习 1：MEDIUM 工具的 MCP 暴露策略

练习 2：**kwargs 对 inspect.signature 的影响

练习 3：schema 被修改会影响内部吗？

练习 4：给 MCP 加认证的最小改动

练习 5（最重要）：Cursor 里删除文档的完整链路

三句话带走第 12 课

练习 2：`**kwargs` 对 `inspect.signature` 的影响