MCP协议：智能体的标准化工具接口

2026-07-05 10408 words 21 minutes

Contents

1. 引言：MCP解决了什么问题

在前面几章中，我们反复使用了一个关键能力——Function Calling（函数调用）。无论是查数据库、调API，还是操作文件系统，智能体都需要通过"函数"这一抽象来触达外部世界。然而，当工具数量膨胀、应用场景复杂化时，一个令人头疼的问题浮现出来：

工具集成碎片化（Tool Integration Fragmentation）。

具体表现为：

协议不统一：每家大模型厂商都定义了自己的Function Calling格式。OpenAI、Anthropic、Google、阿里通义各有各的工具描述规范，同一份工具定义要写多遍。
上下文割裂：工具本身没有"资源"概念。Agent要读取一份文件、一段数据库Schema，都得靠开发者手动拼接到Prompt里，缺乏统一的资源访问接口。
生态重复造轮子：搜数据库、查天气、读GitHub，每个Agent项目都在重复封装相同的工具。A项目写的工具，B项目无法直接复用。
运维耦合：工具进程和Agent进程常常绑死在一起，工具升级、权限管理、多工具编排都很难解耦。

这就好比每个Agent都自带一个"私有的工具箱"，彼此不通，每次都要从零打造。2024年11月，Anthropic发布了Model Context Protocol（MCP，模型上下文协议），给出了一个令人眼前一亮的解法：

把工具、资源、提示模板从Agent中解耦出来，变成可独立部署、可被任意支持MCP的客户端发现的"Server"，Agent则作为"Client"按需连接，即插即用。

用一个类比：MCP之于AI工具生态，就像USB之于硬件外设。在USB出现前，键盘是PS/2口、打印机是并口、鼠标是串口；USB统一接口后，外设生态爆发。MCP正是希望成为"AI Agent的USB"。

本章将从协议原理讲到Server/Client开发，最后用一个天气查询Agent串起完整链路。

2. MCP协议概述

2.1 协议的诞生与定位

MCP（Model Context Protocol）由Anthropic于2024年11月开源发布，是一个开放、基于JSON-RPC 2.0的应用层协议。其设计目标是：

标准化：任何MCP Server都能被任何MCP-compatible的Host（如Claude Desktop、Cursor、各类Agent框架）发现并调用。
解耦：Server与Client通过协议而非代码耦合，可独立部署、独立升级。
可组合：一个Host可同时连接多个Server，工具数量线性扩展。
安全：Server运行在本地或受控环境，Host控制权限边界。

2.2 三大组件：Host / Client / Server

MCP架构由三个核心角色构成：

Host（宿主应用）：最终用户使用的应用，比如Claude Desktop、Cursor IDE，或我们自己开发的Agent程序。Host负责管理多个Client实例、汇总能力、向用户暴露交互入口。
Client（协议客户端）：Host内部的一个连接器，每个Client与一个Server保持1:1的会话。Client负责协议握手、能力协商、消息路由。
Server（服务提供方）：独立运行的进程，向Client暴露三类能力：Tools（工具）、Resources（资源）、Prompts（提示模板）。

它们的关系如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


┌─────────────────────────────┐
│           Host (Agent)      │
│  ┌────────┐  ┌────────┐     │
│  │Client A│  │Client B│ ... │
│  └───┬────┘  └───┬────┘     │
└──────┼───────────┼──────────┘
       │           │
   stdio/SSE   stdio/SSE
       │           │
   ┌───▼────┐  ┌───▼────┐
   │Server A│  │Server B│
   │ 文件系统 │  │ 数据库  │
   └────────┘  └────────┘

2.3 与传统Function Calling的区别

维度	Function Calling	MCP
工具定义位置	写死在Agent代码或Prompt中	独立Server，运行时动态发现
复用性	每个项目重写	一次开发，多方复用
上下文资源	无统一抽象，手动拼接	内置Resources概念，按URI访问
提示模板	无	内置Prompts，可被Host列出并调用
多模型兼容	各厂商格式不同	协议中立，与具体模型解耦
部署形态	与Agent同进程	可独立进程，本地或远程

一个直观的差异：在Function Calling模式下，你换一个模型厂商，工具描述格式可能就要改一遍；而在MCP下，Server完全不用动，Host换底层模型即可。

2.4 2026年生态状态

截至2026年中，MCP已被主流Agent框架广泛采纳：

官方/社区MCP Server数量已突破5000+，覆盖文件系统、Git、数据库（PostgreSQL/MySQL/SQLite）、Slack、GitHub、Notion、Linear、Sentry、Brave Search、Puppeteer等场景。
主流Host支持：Claude Desktop、Cursor、Windsurf、Zed、Cline等均原生支持MCP；LangChain、LlamaIndex、AutoGen也提供了MCP适配层。
多语言SDK：官方提供了TypeScript、Python SDK，社区贡献了Go、Rust、Java、C#等实现。
远程MCP：基于HTTP+SSE的远程Server部署模式逐渐成熟，出现了MCP Gateway、MCP Hub等托管平台。

可以说，MCP已经从"Anthropic的提案"成长为事实标准。

3. MCP架构详解

3.1 Host：宿主应用

Host是用户视角的"Agent应用"。它的职责：

创建并管理多个Client实例（一个Server对应一个Client）。
汇总所有Server暴露的Tools/Resources/Prompts，统一呈现给LLM或用户。
在LLM决定调用某个工具时，路由请求到对应Client，再把结果回传给LLM。
控制权限边界：是否允许某个Server读取某资源、是否允许某工具执行，由Host把关。

Claude Desktop就是典型Host：用户可以在配置文件中声明若干MCP Server，应用启动时拉起这些Server进程，并把工具接入对话。

3.2 Client：协议客户端

Client是Host内部维护的"会话管理器"。每个Client对应一个Server，负责：

初始化握手：发送initialize请求，交换协议版本与能力声明。
能力协商：Server声明自己支持哪些Tools/Resources/Prompts，Client声明自己支持哪些回调（如采样sampling、根目录roots）。
消息收发：所有调用基于JSON-RPC 2.0，Client既是请求方，也接收Server发起的通知与反向请求。
生命周期管理：维持会话、处理断连、优雅关闭。

3.3 Server：服务提供方

Server是能力提供者，可暴露三类原语：

原语	语义	谁主导	典型例子
Tools	可执行的函数，有副作用	LLM主动调用	发邮件、查数据库、运行Shell
Resources	可读取的数据，无副作用	应用/用户主动拉取	文件内容、DB Schema、日志
Prompts	预制的提示模板	用户主动选择	“代码审查"模板、“周报"模板

注意三者的控制权差异：Tools是模型自主决定调用；Resources和Prompts更多是应用层主动获取后注入上下文。这种分工避免了LLM滥用只读资源，也让权限管理更清晰。

3.4 通信协议：JSON-RPC over stdio/SSE

MCP传输层支持两种主流模式：

stdio（标准输入输出）：本地进程间通信。Host以子进程方式拉起Server，通过stdin/stdout交换JSON-RPC消息。延迟低、部署简单，是本地集成的主流方式。
HTTP + SSE（Server-Sent Events）：远程通信。Server作为HTTP服务暴露，Client通过POST发请求、SSE接收推送。适合跨机器、云托管场景。新版本也支持Streamable HTTP，简化了SSE的双向通道。

消息格式示例（Client调用工具）：

1
2
3
4
5
6
7
8
9


{
  "jsonrpc": "2.0",
  "id": 42,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {"path": "/tmp/note.md"}
  }
}

Server响应：

1
2
3
4
5
6
7
8


{
  "jsonrpc": "2.0",
  "id": 42,
  "result": {
    "content": [{"type": "text", "text": "笔记内容..."}],
    "isError": false
  }
}

3.5 生命周期：初始化→能力交换→工具调用→关闭

一次完整的Client-Server会话：

Initialize：Client发送initialize，携带本端支持的协议版本和能力。Server回应当前版本及自身能力。
Initialized 通知：Client发送notifications/initialized，握手完成。
能力列举：Client调用tools/list、resources/list、prompts/list获取清单。
运行期调用：Client按需调用tools/call、resources/read、prompts/get；Server也可反向发起sampling/createMessage让Host代为调用LLM。
关闭：Client发送关闭信号或直接结束stdio通道，Server清理资源退出。

理解这一生命周期对调试MCP连接问题非常重要——多数"工具不出现"的故障都源于握手或能力交换阶段出错。

4. 开发MCP Server

4.1 安装Python SDK

1

pip install mcp

mcp是官方维护的Python SDK，同时支持stdio与SSE两种传输方式，并提供了高级声明式API。下面我们用它开发一个文件系统MCP Server，暴露读取文件、列出目录两个工具。

4.2 定义Tools：工具

一个Tool由三部分组成：

name：唯一标识，小写、下划线分隔。
description：给LLM看的自然语言说明，决定模型何时选用此工具。
inputSchema：JSON Schema描述的参数表，LLM据此生成调用参数。

4.3 定义Resources：资源

Resource以URI形式暴露，例如file:///tmp/note.md。Client可通过resources/list和resources/read访问。Resources适合"被动数据”——LLM不直接调用，而是Host按需拉取后塞入上下文。

4.4 定义Prompts：提示模板

Prompts是参数化的Prompt片段，用arguments描述可填的占位符。用户在UI里选择"周报模板"并填入参数，Host再调用prompts/get获取渲染后的文本。

4.5 完整示例：文件系统MCP Server

下面是完整可运行的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
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160


# filesystem_mcp_server.py
# 文件系统MCP Server示例：提供读取文件、列出目录两个工具
# 运行：python filesystem_mcp_server.py
# 依赖：pip install mcp

import os
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

# 创建Server实例，名称与版本用于握手阶段的能力声明
app = Server("filesystem-mcp")


@app.list_tools()
async def list_tools() -> list[types.Tool]:
    """向Client列举本Server提供的所有工具"""
    return [
        types.Tool(
            name="read_file",
            description="读取指定路径的文本文件内容。路径必须是绝对路径。",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "目标文件的绝对路径，例如 /tmp/note.md"
                    }
                },
                "required": ["path"]
            }
        ),
        types.Tool(
            name="list_directory",
            description="列出指定目录下的文件与子目录名称。",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "目标目录的绝对路径"
                    }
                },
                "required": ["path"]
            }
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """实际执行工具调用，返回文本内容列表"""
    if name == "read_file":
        path = arguments["path"]
        try:
            with open(path, "r", encoding="utf-8") as f:
                content = f.read()
            return [types.TextContent(type="text", text=content)]
        except FileNotFoundError:
            # 通过返回文本承载错误信息，避免抛异常打断协议
            return [types.TextContent(type="text", text=f"错误：文件不存在 {path}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"错误：{e}")]

    elif name == "list_directory":
        path = arguments["path"]
        try:
            entries = os.listdir(path)
            # 区分文件与目录，便于LLM理解结构
            result = []
            for entry in entries:
                full = os.path.join(path, entry)
                kind = "目录" if os.path.isdir(full) else "文件"
                result.append(f"{entry} ({kind})")
            return [types.TextContent(type="text", text="\n".join(result))]
        except Exception as e:
            return [types.TextContent(type="text", text=f"错误：{e}")]

    else:
        return [types.TextContent(type="text", text=f"错误：未知工具 {name}")]


@app.list_resources()
async def list_resources() -> list[types.Resource]:
    """列举可被读取的资源（此处把/tmp下的.md文件暴露为资源）"""
    resources = []
    tmp_dir = "/tmp"
    if os.path.isdir(tmp_dir):
        for entry in os.listdir(tmp_dir):
            if entry.endswith(".md"):
                full = os.path.join(tmp_dir, entry)
                resources.append(
                    types.Resource(
                        uri=f"file://{full}",
                        name=entry,
                        description=f"/tmp 下的Markdown文件：{entry}",
                        mimeType="text/markdown"
                    )
                )
    return resources


@app.read_resource()
async def read_resource(uri: str) -> str:
    """根据URI读取资源内容"""
    # file:///tmp/note.md -> /tmp/note.md
    if uri.startswith("file://"):
        path = uri[7:]
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    raise ValueError(f"不支持的URI：{uri}")


@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    """列举可用的提示模板"""
    return [
        types.Prompt(
            name="summarize_file",
            description="生成一段提示词，让模型总结指定文件内容",
            arguments=[
                types.PromptArgument(
                    name="path",
                    description="要总结的文件路径",
                    required=True
                )
            ]
        )
    ]


@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
    """渲染提示模板为完整消息列表"""
    if name == "summarize_file":
        path = arguments.get("path", "")
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"请阅读以下文件内容，并用三句话总结要点：\n\n路径：{path}\n\n请总结。"
                    )
                )
            ]
        )
    raise ValueError(f"未知模板：{name}")


async def main():
    """以stdio模式启动Server"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

几个值得注意的细节：

错误处理风格：MCP推荐"返回文本错误"而非抛异常，这样LLM能看到错误描述并自行决定是否重试或换工具。
Resource的URI：必须符合协议://路径形式，file://是最常用的，也可以自定义如db://users/42。
Prompt返回的是消息列表：这样模板可以包含多轮对话（system+user），比纯字符串更灵活。

5. 开发MCP Client

5.1 Client的职责

一个完整的MCP Client需要做四件事：

连接：通过stdio或SSE与Server建立会话。
初始化：发送initialize、接收能力声明、发送initialized通知。
发现：调用tools/list等列举能力。
调用：按需调用tools/call，解析返回的content数组。

5.2 完整示例：MCP Client集成到Agent

下面这段代码启动上一节的文件系统Server，并把它的工具接入一个简易Agent循环：

  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
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101


# mcp_client_agent.py
# 通过MCP Client连接文件系统Server，并把工具接入一个简易Agent
# 依赖：pip install mcp openai
# 环境变量：OPENAI_API_KEY、OPENAI_BASE_URL（可选）

import os
import sys
import asyncio
import json

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

from openai import OpenAI


# LLM客户端：可替换为任何兼容OpenAI接口的模型服务
llm = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-xxx"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
MODEL = os.getenv("MODEL_NAME", "gpt-4o-mini")


async def run_agent(user_query: str):
    """启动MCP Server子进程，建立会话，运行一轮Agent对话"""

    # 1. 配置stdio连接参数：以子进程方式拉起Server
    server_params = StdioServerParameters(
        command=sys.executable,           # 当前Python解释器
        args=["filesystem_mcp_server.py"],# Server脚本路径
        env=os.environ.copy()            # 透传环境变量
    )

    # 2. 建立stdio连接 + 创建Client会话
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 3. 初始化握手
            await session.initialize()

            # 4. 发现工具
            tools_resp = await session.list_tools()
            mcp_tools = tools_resp.tools
            print(f"[Client] 发现 {len(mcp_tools)} 个工具：",
                  [t.name for t in mcp_tools])

            # 5. 把MCP工具转换为OpenAI Function Calling格式
            oa_tools = []
            for t in mcp_tools:
                oa_tools.append({
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description,
                        "parameters": t.inputSchema
                    }
                })

            # 6. Agent主循环：对话 -> 可能调用工具 -> 回填 -> 继续对话
            messages = [
                {"role": "system", "content": "你是一个文件助手，可以读写本地文件。"},
                {"role": "user", "content": user_query}
            ]

            for _ in range(5):  # 最多5轮，防止死循环
                resp = llm.chat.completions.create(
                    model=MODEL,
                    messages=messages,
                    tools=oa_tools
                )
                msg = resp.choices[0].message
                messages.append(msg.model_dump(exclude_none=True))

                # 没有工具调用，对话结束
                if not msg.tool_calls:
                    print(f"[Agent] 回复：{msg.content}")
                    break

                # 执行每一个工具调用
                for call in msg.tool_calls:
                    fn_name = call.function.name
                    fn_args = json.loads(call.function.arguments or "{}")
                    print(f"[Agent] 调用工具 {fn_name}，参数：{fn_args}")

                    # 通过MCP Client调用Server的工具
                    result = await session.call_tool(fn_args.get("__skip__") and "" or fn_name, fn_args)
                    # 把工具结果回填为tool角色消息
                    tool_text = "\n".join(
                        c.text for c in result.content if hasattr(c, "text")
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": tool_text
                    })
                    print(f"[Tool] 结果：{tool_text[:200]}")


if __name__ == "__main__":
    # 示例：让Agent列出/tmp目录并读取某个文件
    asyncio.run(run_agent("请列出 /tmp 目录下有哪些文件，并读取其中任意一个 .md 文件的内容。"))

注意我们做了协议桥接：MCP工具的inputSchema正好就是OpenAI Function Calling所需的JSON Schema，所以转换几乎零成本。这也是MCP设计中"协议中立"红利的体现——你可以用任意LLM厂商的API，工具层完全复用。

提示：真实项目中建议把call_tool的参数名清理一下（上面代码中有一处用于演示容错的__skip__分支，实际可直接传fn_name）。在生产代码里应去掉这种防御性分支，保持调用清晰。

6. MCP实战：构建天气查询Agent

理论清楚了，下面用一个更贴近真实场景的例子把全流程串起来。我们将：

开发一个天气MCP Server，封装开放天气API。
用MCP Client把它接入Agent，让用户用自然语言问天气。

6.1 天气MCP 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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76


# weather_mcp_server.py
# 天气查询MCP Server：暴露get_weather工具
# 依赖：pip install mcp httpx
# 环境变量：WEATHER_API_KEY（如使用需要Key的服务）

import os
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("weather-mcp")

# 这里使用免费的wttr.in服务，无需API Key，便于演示
WEATHER_URL = "https://wttr.in/{city}?format=j1"


@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="查询指定城市的当前天气，返回温度、天气状况、湿度等。",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名（英文或拼音），例如 Beijing、shanghai"
                    }
                },
                "required": ["city"]
            }
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name != "get_weather":
        return [types.TextContent(type="text", text=f"未知工具：{name}")]

    city = arguments["city"]
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(WEATHER_URL.format(city=city))
            resp.raise_for_status()
            data = resp.json()

        # 解析当前天气
        current = data["current_condition"][0]
        temp_c = current["temp_C"]
        desc = current["weatherDesc"][0]["value"]
        humidity = current["humidity"]
        wind = current["windspeedKmph"]

        text = (
            f"城市：{city}\n"
            f"温度：{temp_c}°C\n"
            f"天气：{desc}\n"
            f"湿度：{humidity}%\n"
            f"风速：{wind}km/h"
        )
        return [types.TextContent(type="text", text=text)]
    except Exception as e:
        return [types.TextContent(type="text", text=f"查询失败：{e}")]


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

6.2 Agent通过MCP调用天气工具

我们复用上一节的Client骨架，只换Server脚本与System Prompt：

 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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75


# weather_agent.py
# 天气查询Agent：通过MCP调用weather_mcp_server
# 依赖：pip install mcp openai

import os
import sys
import asyncio
import json

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

llm = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "sk-xxx"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
MODEL = os.getenv("MODEL_NAME", "gpt-4o-mini")


async def run_weather_agent(user_query: str):
    server_params = StdioServerParameters(
        command=sys.executable,
        args=["weather_mcp_server.py"],
        env=os.environ.copy()
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools_resp = await session.list_tools()
            oa_tools = [{
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.inputSchema
                }
            } for t in tools_resp.tools]

            messages = [
                {"role": "system", "content": "你是天气助手，可以查询全球城市天气。"},
                {"role": "user", "content": user_query}
            ]

            for _ in range(4):
                resp = llm.chat.completions.create(
                    model=MODEL, messages=messages, tools=oa_tools
                )
                msg = resp.choices[0].message
                messages.append(msg.model_dump(exclude_none=True))

                if not msg.tool_calls:
                    print(f"天气助手：{msg.content}")
                    return

                for call in msg.tool_calls:
                    fn_name = call.function.name
                    fn_args = json.loads(call.function.arguments or "{}")
                    print(f"[调用] {fn_name}({fn_args})")

                    result = await session.call_tool(fn_name, fn_args)
                    tool_text = "\n".join(
                        c.text for c in result.content if hasattr(c, "text")
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": tool_text
                    })


if __name__ == "__main__":
    asyncio.run(run_weather_agent("北京和上海今天分别多少度？需要带伞吗？"))

运行效果（示意）：

1
2
3
4


[调用] get_weather({'city': 'Beijing'})
[调用] get_weather({'city': 'Shanghai'})
天气助手：北京今天约28°C，晴，湿度40%；上海26°C，小雨，湿度85%。
上海有雨，建议带伞；北京晴朗，无需带伞。

可以看到，LLM自主决定调用两次工具（一次问北京、一次问上海），再把结果综合成自然语言回答。整个过程Server完全不知道用的是哪个模型，模型也不关心Server用什么语言写——这就是MCP带来的解耦价值。

7. MCP最佳实践

7.1 Server设计原则

工具粒度适中：太细（每个SQL一个工具）会让LLM选择困难；太粗（一个"do_anything"工具）又失去可组合性。建议按"一个完整用户意图"为粒度。
描述写给LLM看：description不仅是文档，更是模型决策依据。写清楚"何时用、不用、输入输出含义”，比堆砌功能列表更有效。
Schema严格：用required、enum、pattern等约束参数，能显著降低模型传错参数的概率。
幂等优先：读类工具尽量幂等；写类工具应在描述中明确副作用，便于Host做权限审批。

7.2 安全与权限

最小权限：Server只暴露必要能力。文件系统Server应限制在指定根目录，避免暴露/etc/passwd。
路径校验：对路径参数做规范化与越权检查，防止../穿越。
Host审批：生产环境Host应实现"工具调用确认"机制，敏感操作需用户点击确认后再下发。
远程Server鉴权：基于HTTP的远程MCP应使用Token、mTLS等手段鉴权，避免裸暴露。
审计日志：记录每一次tools/call的入参出参，便于事后追溯。

7.3 错误处理

返回而非抛出：工具失败时返回TextContent描述错误，让LLM有机会重试或换策略；只在协议层错误（如序列化失败）才抛异常。
结构化错误码：可在文本中嵌入[ERR_TYPE]前缀，便于Host做策略化处理（如配额耗尽自动切换Server）。
超时与重试：网络类工具设置合理超时；Client层可对瞬时错误做有限次重试。

7.4 性能优化

异步IO：Python SDK基于asyncio，工具内部应使用httpx、aiosqlite等异步库，避免阻塞事件循环。
批量接口：若工具常被连续调用多次（如批量查天气），可设计get_weather_batch减少往返。
缓存：对慢且少变的资源（如DB Schema）在Server端缓存，减少resources/read延迟。
连接复用：远程MCP Client应复用HTTP连接，避免每次调用都重建TLS。
冷启动：stdio模式下Server是子进程，启动慢会拖慢Host首屏；可在Server启动时预加载模型/连接池。

8. 小结

本章我们从"工具集成碎片化"这一痛点出发，完整走过了MCP协议的理论与实践：

MCP是什么：Anthropic提出的开放协议，用Host/Client/Server三段式架构，把工具、资源、提示模板从Agent中解耦。
为什么重要：它让工具像USB外设一样即插即用，2026年已有5000+ Server、主流框架原生支持，正在成为AI工具生态的事实标准。
怎么开发：用Python mcp SDK，几十行代码就能写出一个暴露Tools/Resources/Prompts的Server；Client侧通过stdio或SSE连接，几行代码就能把工具接入任意LLM的Function Calling循环。
怎么用对：注意工具粒度、Schema严格性、安全权限与异步性能，才能让MCP真正成为"可生产"的集成层。

一句话总结：MCP把"Agent怎么用工具"这件事从代码层升级到了协议层。掌握MCP，你写的工具就不再属于某一个项目，而属于整个生态。

9. 预告：第11章评估与可观测性

当Agent接入了越来越多的工具（无论通过Function Calling还是MCP），一个新的问题随之浮现——怎么知道它干得好不好？ 一个跑偏的工具调用、一次幻觉的回答、一段静默失败的任务，都可能让整个Agent失去可信度。

第11章我们将聚焦评估与可观测性：

评估（Evaluation）：如何用基准任务集、LLM-as-a-Judge、人工评分等手段量化Agent能力，避免"看着像对的"幻觉。
可观测性（Observability）：如何用Trace、Span、指标体系把Agent的一次完整运行"切片"展现，定位到底是Prompt、工具还是模型的问题。
工具链：介绍LangSmith、Langfuse、Phoenix、OpenTelemetry等在Agent场景的应用。

从"能跑"到"可信"，评估与可观测性是Agent走向生产不可绕过的一关。我们下一章见。