Model Context Protocol（MCP）原理解析与实战

MCP 是 Anthropic 提出的标准化协议，让 AI 模型与外部工具和数据源安全交互。

MCP 协议架构

┌─────────────────┐        ┌─────────────────┐
│   AI Application │        │   MCP Client    │
│  (Claude App)   │◄──────►│   (SDK)         │
└─────────────────┘   JSON-RPC  └────────┬────────┘
                                          │
                                   ┌──────┴──────┐
                                   ▼             ▼
                            ┌──────────┐   ┌──────────┐
                            │ MCP Host │   │ MCP Host │
                            └────┬─────┘   └────┬─────┘
                                 │              │
                            ┌────┴────┐   ┌────┴────┐
                            │Local Tool│   │Remote API│
                            │ Server   │   │ Server   │
                            └──────────┘   └──────────┘

核心概念

1. Resources（资源）

python复制
# 服务端：暴露资源
@mcp.resource("document://{doc_id}")
def get_document(doc_id: str) -> dict:
    """返回文档内容"""
    return {"id": doc_id, "content": load_doc(doc_id)}

# 客户端：订阅资源变化
async def on_document_change(uri: str):
    print(f"文档 {uri} 发生了变化")

2. Tools（工具）

python复制
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("我的服务器")

@mcp.tool()
def search_knowledge_base(query: str) -> str:
    """在知识库中搜索"""
    return search(query)

@mcp.tool()
def run_python(code: str) -> str:
    """执行 Python 代码"""
    result = subprocess.run(["python3", "-c", code], capture_output=True)
    return result.stdout or result.stderr

@mcp.tool()
def get_current_weather(city: str) -> dict:
    """获取城市天气"""
    return {"city": city, "temp": 25, "condition": "晴"}

3. Prompts（提示模板）

python复制
@mcp.prompt()
def code_review(code: str, language: str) -> list[dict]:
    return [
        {"role": "system", "content": "你是一个代码审查专家"},
        {"role": "user", "content": f"请审查以下 {language} 代码：\n{code}"}
    ]

4. Sampling（采样回调）

python复制
# 服务端可以请求 LLM 采样
@mcp.prompt()
def analyze_with_llm(text: str) -> str:
    return f"请分析：{text}"

# 服务端实现采样
mcp.set sampling_handler(lambda request: llm.generate(request.prompt))

JSON-RPC 消息格式

python复制
# 请求
{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "search_knowledge_base",
        "arguments": {"query": "RAG 是什么"}
    }
}

# 响应
{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "content": [
            {"type": "text", "text": "RAG 是检索增强生成..."}
        ],
        "isError": False
    }
}

Transport 传输层

1. Stdio（本地进程通信）

bash复制
# 启动本地 MCP 服务器
python -m my_mcp_server --stdio

# Claude App 通过 stdio 通信

2. HTTP + SSE（远程通信）

python复制
from fastapi import FastAPI
from sse_starlette import EventSourceResponse

app = FastAPI()

@app.get("/sse")
async def sse_endpoint():
    async def event_generator():
        while True:
            message = await message_queue.get()
            yield {"event": "message", "data": json.dumps(message)}
    return EventSourceResponse(event_generator())

@app.post("/mcp")
async def mcp_endpoint(request: Request):
    body = await request.json()
    response = await mcp.handle_request(body)
    return response

MCP Server 开发实战

python复制
# server.py
from mcp.server.fastmcp import FastMCP
from pydantic import Field

mcp = FastMCP("企业知识库助手")

@mcp.tool()
def search_docs(
    query: str = Field(description="搜索关键词"),
    doc_type: str | None = Field(default=None, description="文档类型过滤")
) -> list[dict]:
    """搜索企业文档"""
    results = db.query(
        "SELECT * FROM docs WHERE content LIKE %s",
        (f"%{query}%",)
    )
    if doc_type:
        results = [r for r in results if r["type"] == doc_type]
    return results[:10]

@mcp.resource("config://company")
def get_company_config() -> dict:
    """返回公司配置"""
    return {
        "name": "某科技公司",
        "founded": "2020",
        "employees": 500
    }

if __name__ == "__main__":
    mcp.run(transport="stdio")

MCP 安全性

python复制
# 1. 工具权限控制
@mcp.tool(
    permissions=["read:documents", "search:knowledge_base"],
    description="搜索知识库，需要 read 权限"
)
def search(...): ...

# 2. 资源访问控制
@mcp.resource("secret://{key}")
def get_secret(key: str, ctx: Context) -> str:
    if not ctx.has_permission("read:secrets"):
        raise PermissionError("无权限读取密钥")
    return secrets[key]

# 3. 审计日志
@mcp.on_request(async def log_request(req):
    audit_log.info(f"{ctx.user_id} 调用 {req.method} at {time.time()}")
)

常见面试问题

Q1: MCP 和 OpenAI Function Calling 的区别？

Q2: MCP 的优势？

1. 标准化：一个协议对接所有工具
2. 安全：权限隔离，审计日志
3. 双向：Server 可以主动推送
4. 可组合：多个 MCP Server 协同

Q3: MCP 的限制？

1. 协议较新，生态还在完善
2. Claude 生态优先，其他模型支持有限
3. 远程 MCP 需要额外安全配置