MCP 协议与 AI Agent 架构：从概念到实战

2025 年底，Anthropic 提出的 MCP（Model Context Protocol） 正在成为 AI Agent 生态中最受关注的协议标准之一。如果你正在构建 AI 应用，理解 MCP 的设计理念和实现方式，将直接影响你能否在未来的 Agent 生态中占据一席之地。

为什么我们需要 MCP？

在 MCP 出现之前，每个 AI 工具集都有自己的集成方式。你想让 Claude 读 GitHub？写一个插件。想让 GPT 查数据库？再写一个插件。这种 M×N 的集成复杂度 让整个行业苦不堪言。

MCP 的核心思想很简单：用一套统一的协议连接所有 AI 模型与外部工具，把 M×N 变成 M+N。就像 USB-C 统一了充电接口，MCP 统一了 AI 与外部世界的通信方式。

MCP 协议的核心架构

MCP 采用经典的 Client-Server 架构，通信层基于 JSON-RPC 2.0，传输方式支持 stdio 和 HTTP/SSE 两种模式。

通信流程

一个典型的 MCP 交互如下：

1. 初始化握手：Client 发送 initialize 请求，Server 返回能力声明
2. 能力发现：Client 调用 tools/list 获取可用工具列表
3. 工具调用：Client 发送 tools/call，Server 执行并返回结果
4. 资源访问：通过 resources/read 获取外部数据

动手实现：从零搭建一个 MCP Server

下面用 Python 实现一个天气查询 MCP Server，让 AI 能实时获取任意城市的天气信息。

第一步：安装依赖

pip install mcp httpx

第二步：编写 Server

import json
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 创建 MCP Server 实例
app = Server("weather-mcp-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """声明 Server 提供的所有工具"""
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的实时天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 '北京'、'Shanghai'"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_forecast",
            description="获取未来三天的天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    },
                    "days": {
                        "type": "integer",
                        "description": "预报天数（1-7）",
                        "default": 3
                    }
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """处理工具调用请求"""
    if name == "get_weather":
        city = arguments["city"]
        # 这里接入真实天气 API，示例返回模拟数据
        weather_data = await fetch_weather(city)
        return [TextContent(type="text", text=json.dumps(weather_data, ensure_ascii=False, indent=2))]
    elif name == "get_forecast":
        city = arguments["city"]
        days = arguments.get("days", 3)
        forecast = await fetch_forecast(city, days)
        return [TextContent(type="text", text=json.dumps(forecast, ensure_ascii=False, indent=2))]
    else:
        raise ValueError(f"未知工具: {name}")

async def fetch_weather(city: str) -> dict:
    """获取实时天气（示例使用 wttr.in）"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://wttr.in/{city}?format=j1&lang=zh")
        data = resp.json()
        current = data["current_condition"][0]
        return {
            "city": city,
            "temperature": current["temp_C"] + "°C",
            "feels_like": current["FeelsLikeC"] + "°C",
            "humidity": current["humidity"] + "%",
            "description": current["lang_zh"][0]["value"],
            "wind_speed": current["windspeedKmph"] + " km/h"
        }

async def fetch_forecast(city: str, days: int) -> dict:
    """获取天气预报"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://wttr.in/{city}?format=j1&lang=zh")
        data = resp.json()
        forecasts = []
        for day in data["weather"][:dates]:
            forecasts.append({
                "date": day["date"],
                "max_temp": day["maxtempC"] + "°C",
                "min_temp": day["mintempC"] + "°C",
                "description": day["hourly"][4]["lang_zh"][0]["value"]
            })
        return {"city": city, "forecast": forecasts}

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())

第三步：配置 Claude Desktop 接入

在 claude_desktop_config.json 中添加：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"],
      "transport": "stdio"
    }
  }
}

重启 Claude Desktop 后，你就可以直接问：“北京现在天气怎么样？”，Claude 会自动调用你的 MCP Server 获取实时数据。

MCP 与 Function Calling 的区别

很多人会问：OpenAI 早就有 Function Calling 了，为什么还需要 MCP？

一句话总结：Function Calling 是”让 AI 调函数”，MCP 是”让 AI 接入整个工具生态”。

生产环境最佳实践

1. 错误处理要完善：工具调用可能超时、参数不合法、外部 API 不可用，务必返回清晰的错误信息，让 AI 能据此做出合理决策
2. 工具描述要精准：description 字段直接影响 AI 的调用决策，模糊的描述会导致误调用或漏调用
3. 控制工具数量：单个 Server 建议不超过 15 个工具，太多会导致 AI 选择困难。可以通过分组或路由策略拆分
4. 安全第一：生产环境务必加入鉴权、速率限制和输入校验，避免工具被滥用
5. 善用 Resources：把频繁变化的数据（如配置、状态）暴露为 Resources，比每次都调用 Tool 更高效

展望：MCP 的未来

MCP 正在快速演进。目前的重点方向包括：

• 多模态支持：让工具返回图片、音频等富媒体内容
• Agent-to-Agent 通信：不同 AI Agent 之间通过 MCP 协作
• 远程 Server 托管：通过 HTTP/SSE 模式部署云端 MCP Server，支持多租户
• 安全与授权：OAuth 2.1 集成，细粒度的权限控制

可以预见，MCP 正在成为 AI Agent 时代的事实标准协议。越早掌握它，越能在下一波 AI 应用浪潮中抢占先机。

📝 本文基于 MCP SDK 最新版本编写，代码示例兼容 Python SDK 1.x。最后更新：2026 年 6 月