第八章：MCP 协议入门

发表于 分类于 ， Changyan： 本文字数： 2k 阅读时长 ≈ 7 分钟

MCP是AI工具调用的标准化协议，解决了Function Call碎片化问题。本节课带你理解MCP的设计思想、核心概念和实际应用。

MCP（Model Context Protocol）是 AI 工具调用的标准化协议，解决了 Function Call 的碎片化问题。

8.1 什么是 MCP？

MCP（Model Context Protocol，模型上下文协议） 是 Anthropic 于 2024 年底推出的开放标准协议，让 LLM 用统一接口连接外部工具和数据源。

类比 Web 开发：

1
2
Function Call = 早期 Ajax（每个浏览器实现不同）
MCP           = REST API 标准（统一的接口规范）

就像 Language Server Protocol（LSP）统一了编程语言支持，MCP 统一了 AI 工具集成。

8.2 MCP 解决了什么问题？

问题没有 MCP有了 MCP
工具发现手动注册，硬编码客户端自动发现服务器的工具列表
接口格式每个厂商不同统一 JSON-RPC 2.0 协议
跨应用共享不可能一个 MCP Server 可被多个客户端使用
上下文传递不支持支持 Roots、Sampling 等上下文机制
安全控制应用层自行处理协议级用户授权

8.3 MCP 架构：Client-Server 模型

graph TD
subgraph Host["Host 宿主应用"]
    C1["MCP Client (连接A)"]
    C2["MCP Client (连接B)"]
    C3["MCP Client (连接C)"]
end

C1 -->|JSON-RPC| S1["MCP Server (文件系统)"]
C2 -->|JSON-RPC| S2["MCP Server (数据库)"]
C3 -->|JSON-RPC| S3["MCP Server (GitHub)"]

style Host fill:#e3f2fd,stroke:#1565c0
style S1 fill:#fff9c4,stroke:#f9a825
style S2 fill:#fff9c4,stroke:#f9a825
style S3 fill:#fff9c4,stroke:#f9a825

MCP 通信序列：

sequenceDiagram
participant Host as Host 应用
participant Client as MCP Client
participant Server as MCP Server

Host->>Client: 初始化连接
Client->>Server: initialize 请求
Server-->>Client: 返回服务器能力
Client->>Server: list_tools
Server-->>Client: 返回工具列表
Client->>Server: list_resources
Server-->>Client: 返回资源列表
Client->>Server: list_prompts
Server-->>Client: 返回提示词模板

rect rgb(232, 245, 233)
    Note over Host,Server: 用户触发工具调用
    Host->>Client: 调用工具
    Client->>Server: call_tool(name, arguments)
    Server->>Server: 执行工具逻辑
    Server-->>Client: 返回执行结果
    Client-->>Host: 返回给 LLM
end

三个角色：

角色说明类比
HostLLM 应用，发起连接浏览器
ClientHost 内部的连接器浏览器内的 Tab
Server提供工具和数据的服务Web API 服务器

8.4 MCP 三大原语（Primitives）

1. Tools（工具）

LLM 可以调用的函数 — 与 Function Call 直接对应。

1
2
3
4
5
6
7
8
9
10
11
12
13
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("SQL Assistant")

@mcp.tool()
def execute_query(sql: str) -> str:
    """执行 SQL 查询并返回结果"""
    return db.execute(sql)

@mcp.tool()
def list_tables() -> str:
    """列出数据库中所有表"""
    return str(db.get_table_names())

2. Resources（资源）

LLM 可以读取的数据 — 类似 REST API 的 GET。

1
2
3
4
5
6
7
8
9
@mcp.resource("schema://{table_name}")
def get_table_schema(table_name: str) -> str:
    """获取指定表的结构"""
    return db.get_schema(table_name)

@mcp.resource("config://database")
def get_db_config() -> str:
    """获取数据库配置信息"""
    return f"Database: {db.name}, Dialect: {db.dialect}"

3. Prompts（提示词模板）

预定义的提示词模板 — 类似 REST API 的固定查询模式。

1
2
3
4
5
6
7
8
9
10
11
12
@mcp.prompt()
def sql_expert(question: str, dialect: str = "sqlite") -> str:
    """生成 SQL 专家提示词"""
    return f"""你是一个 {dialect} 数据库专家。
请根据以下问题编写 SQL 查询：

问题：{question}

要求：
- 只使用 SELECT 语句
- 限制返回 10 条结果
- 使用正确的 {dialect} 语法"""

8.5 快速开始：创建你的第一个 MCP Server

安装

1
pip install "mcp[cli]"

创建 Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My First MCP Server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def search_city(name: str) -> str:
    """Search for a city by name"""
    cities = {"北京": "中国首都，人口2189万", "上海": "中国最大城市，人口2487万"}
    return cities.get(name, f"未找到城市: {name}")

@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"你好, {name}!"

@mcp.prompt()
def help_prompt(topic: str) -> str:
    """Generate a help prompt"""
    return f"请详细介绍 {topic} 的使用方法"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")  # 生产推荐
    # mcp.run()  # 默认 stdio，适合本地开发

运行 Server

1
2
3
4
5
6
7
8
# 方式一：直接运行
python server.py

# 方式二：用 MCP CLI 调试
uv run mcp dev server.py

# 方式三：安装到 Claude Desktop
uv run mcp install server.py

创建 Client 连接 Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# client.py
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

async def main():
    # 连接 MCP Server
    async with streamable_http_client("http://localhost:8000/mcp") as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 列出可用工具
            tools = await session.list_tools()
            print(f"可用工具: {[t.name for t in tools.tools]}")

            # 调用工具
            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
            print(f"5 + 3 = {result.content[0].text}")

            # 列出资源
            resources = await session.list_resources()
            print(f"可用资源: {[r.uri for r in resources.resources]}")

            # 读取资源
            from pydantic import AnyUrl
            greeting = await session.read_resource(AnyUrl("greeting://世界"))
            print(f"问候: {greeting}")

            # 列出提示词
            prompts = await session.list_prompts()
            print(f"可用提示词: {[p.name for p in prompts.prompts]}")

asyncio.run(main())

8.6 MCP 高级特性

结构化输出

1
2
3
4
5
6
7
8
9
10
11
from pydantic import BaseModel, Field

class WeatherData(BaseModel):
    temperature: float = Field(description="温度（摄氏度）")
    humidity: float = Field(description="湿度百分比")
    condition: str = Field(description="天气状况")

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """获取天气信息 — 返回结构化数据"""
    return WeatherData(temperature=22.5, humidity=45.0, condition="sunny")

进度报告

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from mcp.server.fastmcp import Context, FastMCP

mcp = FastMCP("Progress Demo")

@mcp.tool()
async def long_task(task_name: str, ctx: Context, steps: int = 5) -> str:
    """执行耗时任务，报告进度"""
    for i in range(steps):
        await ctx.report_progress(
            progress=(i + 1) / steps,
            total=1.0,
            message=f"步骤 {i + 1}/{steps}",
        )
    return f"任务 '{task_name}' 完成"

用户交互（Elicitation）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from pydantic import BaseModel, Field

class BookingConfirm(BaseModel):
    check_alternative: bool = Field(description="是否查看其他日期？")
    alternative_date: str = Field(default="2026-05-10", description="备选日期")

@mcp.tool()
async def book_room(date: str, ctx: Context) -> str:
    """预订房间"""
    if date == "2026-05-07":
        result = await ctx.elicit(
            message="该日期已满，是否查看其他日期？",
            schema=BookingConfirm,
        )
        if result.action == "accept" and result.data and result.data.check_alternative:
            return f"已预订 {result.data.alternative_date}"
    return f"已预订 {date}"

生命周期管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class AppContext:
    db_connection: Database

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    db = await Database.connect()
    try:
        yield AppContext(db_connection=db)
    finally:
        await db.disconnect()

mcp = FastMCP("App", lifespan=app_lifespan)

@mcp.tool()
def query_data(sql: str, ctx: Context) -> str:
    """查询数据"""
    db = ctx.request_context.lifespan_context.db_connection
    return db.query(sql)

8.7 MCP vs Function Call 对比

维度Function CallMCP
定义代码中的函数独立运行的服务
协议无（各厂商不同）JSON-RPC 2.0（标准化）
发现手动注册客户端自动 list_tools
跨应用不可能一个 Server 服务多个 Client
上下文需要手动传递内置 Roots、Sampling
资源不支持支持 Resources（数据读取）
提示词不支持支持 Prompts（模板）
运行方式进程内函数调用独立进程 / 网络服务
部署随应用一起独立部署和扩展
适用规模单应用跨应用、跨团队

选型建议：

1
2
3
4
5
6
7
8
9
10
11
12
Function Call 适合：
  - 快速原型
  - 工具数量少（<10）
  - 不需要跨应用共享
  - 简单的单 Agent 场景

MCP 适合：
  - 工具需要跨应用共享
  - 多个 Agent/应用需要同一组工具
  - 需要动态发现工具
  - 企业级部署，工具独立维护
  - 第三方工具集成

8.8 在 DeepAgents 中使用 MCP

DeepAgents 本身不直接内置 MCP 客户端，但你可以通过两种方式集成：

方式一：将 MCP 工具包装为 DeepAgents 工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from deepagents import create_deep_agent

async def fetch_mcp_tools():
    """从 MCP Server 获取工具列表，转换为 DeepAgents 工具"""
    async with streamable_http_client("http://localhost:8000/mcp") as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools_response = await session.list_tools()
            return tools_response.tools, session

# 使用示例
mcp_tools, session = asyncio.run(fetch_mcp_tools())

# 将 MCP 工具包装为 DeepAgents 可用的函数
def make_tool_func(mcp_tool, mcp_session):
    async def tool_func(**kwargs):
        result = await mcp_session.call_tool(mcp_tool.name, arguments=kwargs)
        return result.content[0].text
    
    tool_func.__name__ = mcp_tool.name
    tool_func.__doc__ = mcp_tool.description
    return tool_func

# 创建 Deep Agent 并使用 MCP 工具
tools = [make_tool_func(t, session) for t in mcp_tools]
agent = create_deep_agent(model="openai:gpt-4o", tools=tools)

方式二：通过 LangChain MCP 适配器

1
2
3
4
5
6
7
8
9
10
11
from langchain_mcp import MCPToolkit
from deepagents import create_deep_agent

# LangChain 提供了 MCP 适配器
mcp_toolkit = MCPToolkit(url="http://localhost:8000/mcp")
tools = mcp_toolkit.get_tools()

agent = create_deep_agent(
    model="openai:gpt-4o",
    tools=tools,
)

8.9 传输方式

传输方式适用场景说明
stdio本地开发标准输入输出，最简单
SSE旧版网络通信Server-Sent Events
Streamable HTTP生产推荐HTTP 双向通信，支持流式
1
2
3
4
5
6
7
8
# stdio（本地开发）
mcp.run()

# Streamable HTTP（生产推荐）
mcp.run(transport="streamable-http")

# 指定端口
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

8.10 安全模型

MCP 的四个安全原则：

  1. 用户授权：所有数据访问和操作必须用户明确同意
  2. 数据隐私：Host 不得在未经同意的情况下传输用户数据
  3. 工具安全：工具 = 任意代码执行，必须用户审批
  4. 采样控制：用户必须批准 LLM 采样请求，控制服务器可见内容