引言

随着大语言模型（LLM）能力的不断增强，AI Agent 从“聊天机器人”进化为能够执行实际任务的智能体。Agent 需要查询数据库、调用 API、操作本地文件等，这一切都依赖于工具调用（Function Calling）。然而，如果每个 AI 平台都定义一套私有的工具描述规范，开发者就会陷入适配地狱。为此，Anthropic 在 2024 年底开源了 Model Context Protocol（MCP），旨在统一 LLM 与外部工具、数据源之间的交互方式。

本文将从核心概念出发，通过一个完整的 Python 实战项目，带你用 MCP 构建标准化的工具服务，并模拟一个 AI Agent 完成工具调用。读完你会理解：

• MCP 的架构与设计思想
• 如何创建 MCP Server 并暴露工具
• Agent 如何通过 MCP Client 发现并调用工具
• 实际开发中的关键注意事项

一、核心概念：什么是 MCP？

MCP 的官方定义是：一种开放协议，用于标准化大型语言模型与外部服务之间的交互。它类似于“AI 世界的 USB-C 接口”——只要硬件支持该接口，任何设备都能即插即用。在 MCP 出现之前，每个 LLM 平台（OpenAI、Cohere、Anthropic）都有自己的一套函数调用格式，工具开发者需要为不同平台编写不同的适配代码。

MCP 采用 客户端-服务器架构：

• MCP Host：AI 应用本身，例如 Claude Desktop、IDE 插件或你自己的 Agent 程序。
• MCP Client：嵌入在 Host 中，负责与 MCP Server 建立连接、发送请求。
• MCP Server：轻量级的服务，通过标准协议暴露工具（Tools）、资源（Resources）、提示模板（Prompts）等能力。

一次典型的工具调用流程如下：

1. Host（Agent）通过 Client 向 Server 请求可用的能力列表（如工具的名称、描述、参数 schema）。
2. Server 返回结构化的元数据。
3. Agent 根据任务选择合适的工具并提供参数，通过 Client 发起调用。
4. Server 执行对应函数，将结果返回给 Client，最终送达 Agent。

MCP 底层支持多种传输协议（JSON-RPC over STDIO、HTTP+SSE、WebSocket 等），这使其既能嵌入本地进程，也能搭建远程服务。本文实战将使用 STDIO 传输，简单高效，非常适合本地开发或代码沙箱场景。

二、环境准备

我们使用 Python 的官方 MCP SDK：mcp。它同时提供构建客户端和服务端的工具。

pip install mcp

另外，因为模拟的 Agent 需要模拟“大模型选择工具”的决策，为了示例简洁，我们不接入任何 LLM API，而是手动指定要调用的工具及其参数，以此演示完整链路。你可以在真实项目中用 LangChain、Semantic Kernel 或直接调用 OpenAI API 生成工具调用请求。

项目结构：

mcp_demo/
├── weather_server.py   # MCP Server，提供天气查询和数学运算工具
└── agent.py            # 模拟 Agent，通过 MCP Client 调用工具

三、实战：构建 MCP Server

我们将创建一个 Server，提供两个工具：
- add: 执行两数相加运算
- get_current_weather: 返回指定城市的模拟天气信息

3.1 完整代码（weather_server.py）

import json
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 1. 创建 MCP Server 实例
server = Server("weather-and-math-server")

# 2. 注册工具列表处理器：当客户端请求工具列表时返回可用工具
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="add",
            description="执行两数之和运算",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "第一个数字"},
                    "b": {"type": "number", "description": "第二个数字"}
                },
                "required": ["a", "b"]
            }
        ),
        Tool(
            name="get_current_weather",
            description="获取指定城市的实时天气（模拟）",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称，如 Beijing"}
                },
                "required": ["city"]
            }
        )
    ]

# 3. 注册工具调用处理器：收到调用请求时执行具体逻辑
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "add":
        a = arguments["a"]
        b = arguments["b"]
        result = a + b
        return [TextContent(type="text", text=f"计算结果：{a} + {b} = {result}")]

    elif name == "get_current_weather":
        city = arguments["city"]
        # 模拟天气查询，实际应调用真实 API
        weather_data = {
            "Beijing": "晴天，22°C，湿度45%",
            "Shanghai": "多云，26°C，湿度65%",
            "Shenzhen": "阵雨，30°C，湿度80%"
        }
        info = weather_data.get(city, "未知城市，无法查询")
        return [TextContent(type="text", text=f"{city}天气：{info}")]

    else:
        raise ValueError(f"未知工具：{name}")

# 4. 启动服务（STDIO 传输）
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationCapabilities(
                sampling=NotificationOptions(),
                logging=NotificationOptions(),
                experimental=NotificationOptions(),
            ),
        )

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

代码重点说明：
- server.list_tools() 装饰器用于定义工具清单，必须返回包含 name、description、inputSchema 的 Tool 对象列表。
- inputSchema 遵循 JSON Schema 格式，这是 MCP 要求的标准参数描述，LLM 可以据此自动生成参数。
- server.call_tool() 装饰器是实际执行函数的地方，返回 TextContent 列表（也支持图片、资源等）。
- 通过 stdio_server() 建立标准输入输出传输，这将使 Server 能作为子进程被 Client 启动并通信。

四、实战：构建 Agent 模拟客户端

Agent 程序负责启动 MCP Server 作为子进程，通过 STDIO 传输连接，获取工具列表，选择工具并调用。我们手动指定工具和参数，模拟 LLM 的决策。

4.1 完整代码（agent.py）

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

async def run_agent():
    # 1. 定义 Server 启动参数（以子进程方式启动 Python 脚本）
    server_params = StdioServerParameters(
        command="python",      # 启动命令
        args=["weather_server.py"]  # 服务器脚本
    )

    print("Agent 启动，正在连接 MCP Server...")
    # 2. 建立 STDIO 客户端通道
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 3. 初始化会话（握手、协商能力等）
            await session.initialize()
            print("连接成功！\n")

            # 4. 获取工具列表
            tools_result = await session.list_tools()
            tools = tools_result.tools
            print("可用工具：")
            for tool in tools:
                print(f"- {tool.name}: {tool.description}")

            # 5. 模拟 Agent 决策：依次调用两个工具
            print("\n--- 模拟 Agent 调用 add 工具 ---")
            # 假设大模型决定调用 add(3, 5)
            result = await session.call_tool("add", arguments={"a": 3, "b": 5})
            print("返回结果：", result.content[0].text)

            print("\n--- 模拟 Agent 调用 get_current_weather 工具 ---")
            result = await session.call_tool("get_current_weather", arguments={"city": "Beijing"})
            print("返回结果：", result.content[0].text)

            # 6. 可以尝试调用未定义的工具，观察错误处理
            print("\n--- 尝试调用未知工具 ---")
            try:
                await session.call_tool("delete_file", {})
            except Exception as e:
                print(f"预期的错误：{e}")

    print("\nAgent 任务完成。")

if __name__ == "__main__":
    asyncio.run(run_agent())

运行方式：确保 weather_server.py 和 agent.py 在同一目录，然后执行：

python agent.py

输出示例（省略部分）：

Agent 启动，正在连接 MCP Server...
连接成功！

可用工具：
- add: 执行两数之和运算
- get_current_weather: 获取指定城市的实时天气（模拟）

--- 模拟 Agent 调用 add 工具 ---
返回结果： 计算结果：3 + 5 = 8

--- 模拟 Agent 调用 get_current_weather 工具 ---
返回结果： Beijing天气：晴天，22°C，湿度45%

--- 尝试调用未知工具 ---
预期的错误：Tool not found

至此，一个完整的 MCP 工具调用链路就跑通了。Agent 并没有直接调用任何天气 API，也没有关心函数实现，它只通过 MCP 标准接口发现并使用工具，实现了完美的解耦。

五、常见问题与注意事项

实际项目落地时，以下问题需多加留意。

5.1 工具描述与 Schema 设计

MCP Server 暴露的工具描述是给模型看的，描述越清晰、约束越准确，模型的决策就越可靠。inputSchema 中建议：
- 为每个参数明确指定 type 和 description。
- 使用 required 数组标明必填项。
- 对于枚举值，可用 enum 字段限制可选范围，如 "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}。

5.2 异步执行与并发

MCP 基于 asyncio，Server 和 Client 理论上可以同时处理多个请求。如果你在 Server 的 call_tool 中执行耗时 I/O（如网络请求），请确保使用异步库（aiohttp 等），避免阻塞事件循环。SDK 内部已经为每个工具调用创建了协程任务，无需额外处理。

5.3 安全与权限控制

工具调用可能涉及文件系统、数据库、外部 API，务必小心：
- 对传入参数进行严格校验，防止注入攻击。
- 对于敏感操作（如删除文件），可在 Client 端增加确认交互，或由 LLM 判断后要求用户批准。MCP 支持 notifications/requests 机制实现交互式确认。
- 在生产环境中，Server 应运行在权限受限的容器或沙箱中，STDIO 传输本身隔离性较好，但远程传输务必加密（HTTPS/SSE + TLS）。

5.4 传输方式选择

• STDIO：适合本地开发、IDE 插件、单机 Agent，延迟极低，无需网络配置。
• HTTP+SSE：适合将工具服务化为远程 API，多 Agent 共享，需处理认证鉴权。
• WebSocket：适用于需要双向长连接、流式返回的场景。

本例使用 STDIO 是入门首选，官方 SDK 已完美支持。

5.5 与现有框架的集成

目前 LangChain、LlamaIndex、CrewAI 等主流 Agent 框架都开始支持 MCP。LangChain 提供了 MCPToolkit，可以直接将 MCP Server 暴露的工具加载为 LangChain 的 Tool 对象。这样一来，你既可以享受 MCP 的标准化，又不必改变现有 Agent 构建习惯。

from langchain_mcp import MCPToolkit
toolkit = MCPToolkit(server_params=...)
tools = toolkit.get_tools()

5.6 资源与提示模板

除了工具，MCP 还定义了 Resources（向模型提供上下文数据，如文件内容、数据库记录）和 Prompts（预定义的提示模板）。在构建复杂 Agent 时，善用这些能力可以让模型获得更丰富的环境感知。例如，你可以将用户文档作为 Resource 暴露，让模型直接引用，而不需要把所有内容塞进 prompt。

总结

本文从零讲解了 Model Context Protocol 的核心理念，并给出了一个完整可运行的 Python 示例。通过 MCP Server 暴露工具，并由 Agent 客户端调用，我们实现了 LLM 与外部工具的标准化集成。

MCP 的价值在于解耦与统一：工具开发者只需实现一次 MCP Server，就能被任何遵循协议的 AI 平台使用；Agent 构建者也无需重复编写胶水代码，直接发现并调用即可。这大大降低了构建复杂 AI 应用的工程成本。

随着生态发展，MCP 将在 AI Agent 的“感知 - 决策 - 执行”循环中扮演越来越重要的角色。建议开发者尽早尝试将内部工具封装为 MCP 服务，提前享受标准化带来的红利。

下一步，你可以尝试：
- 为你的日常工具（如搜索引擎、数据库查询）编写 MCP Server。
- 将 MCP 融入到 LangChain 或 AutoGPT 等 Agent 框架中。
- 探索 MCP 的 Resources 和 Prompts 机制，构建更智能的上下文感知 Agent。

希望本文能帮你打开 MCP 实践的大门，欢迎在评论区交流你的想法与踩坑经验！