🔌 MCP 协议

Model Context Protocol 概述

MCP（Model Context Protocol）是由 Anthropic 于 2024 年 11 月发布的开放标准协议，旨在为 AI 模型与外部数据源、工具之间提供统一的连接方式。可以把它理解为 AI 世界的"USB-C 接口"——一个标准，让所有工具和所有模型都能即插即用。

在 MCP 之前，每个 AI 应用都需要为每个工具编写定制集成代码，形成 M×N 的集成难题。MCP 将其简化为 M+N：工具提供者实现 MCP Server，AI 应用实现 MCP Client，双方通过标准协议对话。

为什么需要 MCP？

Client-Server 架构

MCP 采用经典的 Client-Server 架构，其中 Host 应用（如 Claude Desktop）内嵌 MCP Client，通过标准化协议与一个或多个 MCP Server 通信。

关键角色

Tools — 可执行操作

Tools 让 LLM 能够通过 Server 执行操作。模型发现并调用 Tool，但需要用户确认（人在回路）。

# MCP Server 定义 Tool
@server.tool()
async def create_issue(
    title: str,
    body: str,
    repo: str
) -> str:
    """在 GitHub 仓库创建 Issue。
    
    Args:
        title: Issue 标题
        body: Issue 正文描述
        repo: 仓库名称（格式：owner/repo）
    """
    result = await github.create_issue(repo, title, body)
    return f"Issue #{result.number} 创建成功"

Resources — 结构化数据

Resources 让 Server 向 Client 暴露结构化数据，类似 REST API 的 GET 端点。Client（而非模型）控制何时读取。

@server.resource("config://app-settings")
async def get_settings() -> str:
    """获取应用配置信息"""
    return json.dumps(load_config())

# 带 URI 模板的动态 Resource
@server.resource("db://users/{user_id}/profile")
async def get_user_profile(user_id: str) -> str:
    """获取用户档案"""
    user = await db.get_user(user_id)
    return json.dumps(user.to_dict())

Prompts — 提示词模板

Prompts 让 Server 提供可复用的提示词模板，帮助用户更好地与 AI 交互。

@server.prompt()
async def code_review(code: str, language: str) -> list:
    """代码审查提示词模板"""
    return [
        {
            "role": "user",
            "content": f"""请审查以下 {language} 代码：
```{language}
{code}
```
关注：安全性、性能、可读性、最佳实践。"""
        }
    ]

三大原语对比

两种传输方式

通信协议：JSON-RPC 2.0

MCP 使用 JSON-RPC 2.0 作为消息格式，支持三种消息类型：

// Request（请求）
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "create_issue",
    "arguments": { "title": "Bug fix", "repo": "org/repo" }
  }
}

// Response（响应）
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [{ "type": "text", "text": "Issue #42 created" }]
  }
}

// Notification（通知，无需响应）
{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": { "uri": "config://app-settings" }
}

Python SDK 快速开始

# pip install mcp
from mcp.server.fastmcp import FastMCP

# 创建 Server 实例
mcp = FastMCP("my-weather-server")

# 定义 Tool
@mcp.tool()
async def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称，例如"北京"、"上海"
    """
    data = await weather_api.fetch(city)
    return f"{city}: {data.temp}°C, {data.desc}"

# 定义 Resource
@mcp.resource("weather://forecast/{city}")
async def get_forecast(city: str) -> str:
    """获取 7 天天气预报"""
    return json.dumps(await weather_api.forecast(city))

# 定义 Prompt
@mcp.prompt()
async def travel_advice(destination: str) -> str:
    """旅行建议提示词"""
    weather = await get_weather(destination)
    return f"我计划去{destination}旅行，当前天气：{weather}，请给出穿衣和行程建议。"

# 运行
if __name__ == "__main__":
    mcp.run()

TypeScript SDK

// npm install @modelcontextprotocol/sdk
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from
  "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({ name: "weather", version: "1.0.0" });

server.tool(
  "get_weather",
  "Get weather for a city",
  { city: z.string().describe("City name") },
  async ({ city }) => ({
    content: [{ type: "text", text: `${city}: 25°C, Sunny` }]
  })
);

const transport = new StdioServerTransport();
await server.connect(transport);

Python Client 示例

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 连接到 MCP Server
params = StdioServerParameters(
    command="python",
    args=["weather_server.py"]
)

async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        # 初始化连接
        await session.initialize()
        
        # 列出可用工具
        tools = await session.list_tools()
        print("可用工具:", [t.name for t in tools.tools])
        
        # 调用工具
        result = await session.call_tool(
            "get_weather",
            arguments={"city": "北京"}
        )
        print(result.content[0].text)
        # → "北京: 15°C, 晴"

与 LangChain 集成

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

# 连接多个 MCP Server
async with MultiServerMCPClient({
    "weather": {"command": "python", "args": ["weather_server.py"]},
    "github": {"command": "npx", "args": ["@modelcontextprotocol/server-github"]},
}) as client:
    tools = client.get_tools()
    agent = create_react_agent(llm, tools)
    
    result = await agent.ainvoke({
        "messages": [("human", "北京天气怎么样？")]
    })

主要 MCP Server 生态

支持 MCP 的 Host 应用

Server 开发指南

• 工具描述清晰：用自然语言详细描述每个 Tool 的功能、参数含义和返回值，这直接影响 LLM 的调用准确率
• 参数校验严格：使用 Pydantic/Zod 定义参数 Schema，避免注入和异常输入
• 错误处理优雅：返回结构化错误信息，帮助 LLM 理解失败原因并重试
• 最小权限原则：Server 只暴露必要的功能，避免过度授权
• 幂等性：写操作尽量保证幂等，防止重复调用导致数据异常

安全考量