🧩 MCP生态

MCP协议实战指南:从架构解析到生产级Server开发

发布时间:2026-05-14 分类: MCP生态
摘要:MCP协议实战指南:从架构解析到生产级Server开发想给你的AI Agent装上“万能插头”?还在为每个工具单独写集成代码?MCP(Model Context Protocol)就是答案。 这个由Anthropic提出、正在成为行业标准的协议,正在彻底改变AI Agent与外部工具的交互方式。今天,我们不谈虚的,直接上手拆解MCP的四层架构,手把手带你开发一个能直接用的MCP Server...

封面

MCP协议实战指南:从架构解析到生产级Server开发

想给你的AI Agent装上“万能插头”?还在为每个工具单独写集成代码?

MCP(Model Context Protocol)就是答案。 这个由Anthropic提出、正在成为行业标准的协议,正在彻底改变AI Agent与外部工具的交互方式。今天,我们不谈虚的,直接上手拆解MCP的四层架构,手把手带你开发一个能直接用的MCP Server。

一、MCP四层架构:为什么它比Function Calling强?

传统AI工具集成就像“一对一适配器”——每接一个新工具,就得写一套专属代码。MCP把这套逻辑标准化了,核心是四层架构:

1. 传输层(Transport)
负责底层通信。支持两种模式:

  • stdio:适合本地进程通信,延迟最低
  • HTTP + SSE(Server-Sent Events):适合远程服务,支持流式响应

2. 协议层(Protocol)
基于 JSON-RPC 2.0,所有消息都是标准JSON格式。关键点:

  • 请求必须有 jsonrpc: "2.0"idmethodparams
  • 响应必须有 jsonrpc: "2.0"idresulterror
  • 通知(notification)没有 id,不需要回复

3. 能力层(Capabilities)
Server声明自己能做什么:

{
  "capabilities": {
    "tools": {"listChanged": true},
    "resources": {"subscribe": true},
    "prompts": {"listChanged": true}
  }
}

4. 应用层(Application)
具体实现:工具函数、资源读取、提示词模板等。

对比Function Calling:MCP是标准化协议,一次开发,所有支持MCP的AI模型(Claude、GPT、本地模型)都能调用你的工具。而Function Calling是模型私有实现,换模型就得重写。

二、通信机制:JSON-RPC 2.0实战拆解

MCP所有交互都基于JSON-RPC 2.0,我们看一个完整流程:

1. 初始化握手
Client发送:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {},
    "clientInfo": {"name": "test-client", "version": "1.0"}
  }
}

Server响应:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {"listChanged": true}
    },
    "serverInfo": {"name": "my-server", "version": "1.0"}
  }
}

2. 工具发现
Client获取可用工具列表:

{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}

Server返回工具定义:

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [{
      "name": "get_weather",
      "description": "获取指定城市天气",
      "inputSchema": {
        "type": "object",
        "properties": {
          "city": {"type": "string", "description": "城市名称"}
        },
        "required": ["city"]
      }
    }]
  }
}

3. 工具调用
Client执行工具:

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {"city": "北京"}
  }
}

关键细节

  • 每个请求有唯一 id,响应回复相同 id
  • 通知(如 notifications/initialized)没有 id,不需要回复
  • 错误响应包含 error.codeerror.message

三、生产级MCP Server开发:天气查询服务

我们用Python开发一个真实的天气查询MCP Server,支持stdio和HTTP两种传输方式。

项目结构

weather-mcp-server/
├── server.py          # 主服务
├── weather_api.py     # 天气API封装
├── requirements.txt
└── README.md

1. 安装依赖

pip install mcp fastapi uvicorn httpx

2. 核心实现(server.py)

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

# 初始化MCP Server
server = Server("weather-server")

@server.list_tools()
async def list_tools():
    """声明可用工具"""
    return [
        Tool(
            name="get_weather",
            description="获取中国城市实时天气(使用和风天气API)",
            inputSchema={
                "type": "object",

![配图](https://yitb.com/usr/uploads/covers/cover_mcp_20260513_201151.jpg)

                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,如:北京、上海"
                    }
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    """执行工具调用"""
    if name == "get_weather":
        city = arguments.get("city")
        # 调用真实天气API
        weather_data = await fetch_weather(city)
        return [TextContent(type="text", text=weather_data)]
    
    raise ValueError(f"未知工具: {name}")

async def fetch_weather(city: str) -> str:
    """调用和风天气API(示例)"""
    # 实际开发中替换为真实API密钥
    api_key = "your_api_key_here"
    base_url = "https://devapi.qweather.com/v7/weather/now"
    
    # 城市名转LocationID(简化处理)
    location_map = {"北京": "101010100", "上海": "101020100"}
    location = location_map.get(city, "101010100")
    
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{base_url}?location={location}&key={api_key}"
        )
        data = response.json()
        
        if data.get("code") == "200":
            now = data["now"]
            return f"""城市: {city}
天气: {now['text']}
温度: {now['temp']}°C
体感温度: {now['feelsLike']}°C
湿度: {now['humidity']}%
风向: {now['windDir']}
更新时间: {now['obsTime']}"""
        else:
            return f"获取天气失败: {data.get('code')}"

# 启动stdio模式
if __name__ == "__main__":
    import asyncio
    asyncio.run(stdio_server(server))

3. 支持HTTP+SSE传输

# 添加HTTP支持(在server.py中)
from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import json

app = FastAPI()

@app.post("/mcp")
async def handle_mcp_request(request: dict):
    """处理MCP over HTTP请求"""
    method = request.get("method")
    params = request.get("params", {})
    
    if method == "initialize":
        return {
            "jsonrpc": "2.0",
            "id": request["id"],
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {"listChanged": True}},
                "serverInfo": {"name": "weather-server", "version": "1.0"}
            }
        }
    # ... 其他方法处理
    
@app.get("/mcp/sse")
async def sse_endpoint():
    """SSE流式端点"""
    async def event_generator():
        # 发送服务器事件
        yield {"data": json.dumps({"type": "connected"})}
    
    return EventSourceResponse(event_generator())

4. 部署与测试

# 1. 启动stdio模式(供Claude Desktop等客户端)
python server.py

# 2. 启动HTTP模式
uvicorn server:app --host 0.0.0.0 --port 8000

# 3. 测试工具调用
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "get_weather",
      "arguments": {"city": "北京"}
    }
  }'

四、MCP在AI Agent生态中的真实价值

1. 工具市场标准化
想象一个“AI工具商店”:开发者上传MCP Server,用户像安装浏览器插件一样安装工具。龙虾官网(yitb.com)的Agent生态正在朝这个方向发展。

2. 自动化赚钱案例
某跨境电商团队用MCP搭建了自动化流水线:

  • MCP Server 1:监控1688新品(每10分钟扫描)
  • MCP Server 2:自动生成多语言商品描述
  • MCP Server 3:同步到Shopify店铺
  • 结果:人力成本降低70%,上新速度提升5倍,月均新增SKU 3000+

3. 开发效率提升
传统集成一个新工具需要:阅读文档→写适配代码→测试→部署(平均2-4小时)。
用MCP:找到现成MCP Server→配置连接→直接使用(平均5分钟)。

五、下一步行动:三步上手MCP

1. 本地测试(5分钟)

  • 安装Claude Desktop
  • 在配置文件中添加我们的天气Server
  • 直接问Claude:“北京今天天气怎么样?”

2. 开发你的第一个MCP Server(1小时)

  • 克隆我们的示例代码
  • 修改fetch_weather函数,接入你常用的API(股票、新闻、内部系统)
  • 用stdio模式测试通过

3. 加入生态(长期)

  • 将你的MCP Server开源到GitHub
  • 提交到MCP工具目录(如龙虾官网的Agent市场)
  • 关注A2A协议发展,实现Server间互操作

关键资源

MCP不是银弹,但它是目前最接近“AI工具USB标准”的协议。现在开始开发,你的工具就能被所有支持MCP的AI模型调用——这才是真正的杠杆效应。

返回首页