深入解析 Model Context Protocol (MCP) 协议设计、server/client 实现、与 LangGraph 集成模式,包含 IHUI-AI 实战经验。
# MCP 协议实现指南:从零构建生产级 AI 工具生态
> 当 LLM 不再只是「聊天机器人」,而是要查数据库、发邮件、调内部 API 时,工具调用就成了 AI 应用的命脉。Anthropic 在 2024 年开源的 Model Context Protocol (MCP),正在成为 AI 工具生态的「USB-C 接口」。本文是 IHUI-AI 在生产环境落地 MCP 的完整工程总结。
---
Model Context Protocol(简称 MCP)是 Anthropic 于 2024 年 11 月开源的开放协议,目标是**标准化 AI 应用与外部工具/数据源之间的连接方式**。
在 MCP 出现之前,每个 AI 应用都要为每个工具写一份适配代码:
┌─────────┐ 自定义协议 A ┌──────────┐ │ App 1 │ ──────────────────▶│ Tool GPT │ └─────────┘ └──────────┘ ┌─────────┐ 自定义协议 B ┌──────────┐ │ App 2 │ ──────────────────▶│ Tool GPT │ └─────────┘ └──────────┘
M 个应用 × N 个工具 = M×N 份适配代码,这是典型的「集成地狱」。
MCP 的解法是引入一个标准协议层:
┌─────────┐ ┌──────────┐ │ App 1 │ ──┐ ┌──▶│ Tool A │ └─────────┘ │ │ └──────────┘ ┌─────────┐ │ ┌───────┐ ┌──────────┐ │ App 2 │───┼─▶│ MCP │──▶│ Tool B │ └─────────┘ │ └───────┘ └──────────┘ ┌─────────┐ │ │ ┌──────────┐ │ App 3 │ ──┘ └──▶│ Tool C │ └─────────┘ └──────────┘
M + N 份适配代码,复杂度从 O(M×N) 降到 O(M+N)。
MCP 定义了三种核心原语(primitives):
1. **Tools(工具)**:可被 LLM 调用的函数,类似 function calling,但有标准 schema
2. **Resources(资源)**:可被读取的数据源,如文件、数据库记录、API 响应
3. **Prompts(提示模板)**:可复用的 prompt 模板,支持参数化
MCP 支持两种传输方式:
---
很多人第一反应是:「这不就是 function calling 换个名字吗?」差远了。
OpenAI 在 2023 年推出 function calling 时,设计目标是「让 LLM 输出结构化 JSON」。它解决的是**单次调用**问题,不是**工具生态**问题:
| 维度 | function calling | MCP |
| ---------- | ------------------------- | ---------------------------- |
| 工具发现 | 硬编码在 prompt 里 | 运行时动态发现(list_tools) |
| 工具复用 | 每个 app 重写一遍 | 一次实现,处处可用 |
| 跨厂商 | OpenAI/Anthropic 格式不同 | 协议层统一 |
| 状态管理 | 无状态,每次重传 | 有 session,支持长连接 |
| 权限模型 | 无 | 内置 capability negotiation |
| 流式响应 | 不支持 | 支持(SSE) |
在 IHUI-AI 这种多 agent 平台里,我们需要:
1. **动态接入新工具**:用户上传一个 MCP server 配置,系统自动发现工具,无需重启
2. **多 agent 共享工具**:聊天 agent 和任务 agent 都能调用同一个 `search_knowledge_base`
3. **权限隔离**:免费用户只能用只读工具,付费用户能用写工具
4. **审计日志**:每次工具调用都要记录 who/when/what/result
这些 function calling 都做不到,而 MCP 的协议设计天然支持。
---
MCP Server 是工具的提供方。下面用 Python 和 TypeScript 各实现一个最小可用的 MCP server。
Anthropic 官方提供了 `fastmcp` 库,把 MCP server 实现简化到装饰器级别:
# apps/ai-service/app/mcp/servers/knowledge_base.py
from fastmcp import FastMCP
from typing import Optional
from app.core.vector_store import pgvector_search
from app.core.auth import verify_tenant
mcp = FastMCP("knowledge-base")
@mcp.tool()
async def search_kb(
query: str,
top_k: int = 5,
tenant_id: str = "",
knowledge_base_id: Optional[str] = None,
) -> dict:
""// artículos relacionados