MCP Server 开发指南：AI Agent 工具扩展实战

凌晨两点，我正在调试一个自动化客服 Agent，突然收到前端的紧急告警：ConnectionError: timeout after 30000ms。用户反馈 AI 完全无法调用外部工具，订单查询、库存检查、物流追踪全部失效。

经过排查，发现问题出在我自定义的 MCP Server 没有正确实现资源订阅机制，导致长连接断开后工具调用请求堆积。这就是为什么我要写这篇教程——帮助国内开发者避免我踩过的坑，系统掌握 MCP Server 的开发方法。

一、MCP 协议核心原理

Model Context Protocol（MCP）是 Anthropic 推出的开放标准，用于连接 AI 模型与外部工具。与传统的 Function Calling 不同，MCP 采用服务器-客户端架构，支持双向通信和实时事件推送。

HolySheep AI 平台提供了兼容 MCP 协议的网关服务，国内开发者可以通过 立即注册 获得低于 50ms 的延迟表现，相比直接调用海外服务商的 200-400ms 延迟，调试效率提升数倍。

协议架构图

┌─────────────────────────────────────────────────────────┐
│                    MCP Host (如 Claude Desktop)          │
└─────────────────────────┬───────────────────────────────┘
                          │ JSON-RPC 2.0 over SSE
┌─────────────────────────▼───────────────────────────────┐
│              MCP Client (SDK 封装层)                      │
│  - 工具调用 (tools/call)                                 │
│  - 资源订阅 (resources/subscribe)                        │
│  - 采样请求 (sampling/createMessage)                     │
└─────────────────────────┬───────────────────────────────┘
                          │
┌─────────────────────────▼───────────────────────────────┐
│              MCP Server (你的自定义服务)                  │
│  - 暴露工具清单 (tools/list)                             │
│  - 处理调用请求 (tools/call)                             │
│  - 管理资源变更 (resources/list & 订阅)                   │
└─────────────────────────────────────────────────────────┘

二、开发环境准备

首先安装 MCP SDK，我推荐使用 Python 版本的 mcp 库：

pip install mcp --upgrade

验证安装
python -c "import mcp; print(mcp.__version__)"

项目结构建议如下：

mcp-order-service/
├── server.py              # 主服务入口
├── tools/
│   ├── __init__.py
│   ├── order.py           # 订单相关工具
│   └── inventory.py       # 库存相关工具
├── resources/
│   ├── __init__.py
│   └── product.py         # 产品资源
├── config.py              # 配置管理
└── requirements.txt

三、工具定义与注册

工具是 AI Agent 与外部系统交互的核心桥梁。每个工具需要定义清晰的 Schema，让 AI 理解何时以及如何调用。

# server.py
from mcp.server import Server
from mcp.types import Tool, CallToolResult
from mcp.server.stdio import stdio_server
import asyncio

app = Server("order-service")

定义工具列表
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_order_status",
            description="查询订单物流状态，支持批量查询",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单编号，格式：ORD-YYYYMMDD-XXXXX"
                    },
                    "include_history": {
                        "type": "boolean",
                        "description": "是否包含历史轨迹",
                        "default": False
                    }
                },
                "required": ["order_id"]
            }
        ),
        Tool(
            name="check_inventory",
            description="实时查询商品库存，支持多 SKU",
            inputSchema={
                "type": "object",
                "properties": {
                    "sku_list": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "SKU 列表，最多 50 个"
                    },
                    "warehouse_code": {
                        "type": "string",
                        "description": "仓库编码，不填则查全渠道"
                    }
                },
                "required": ["sku_list"]
            }
        ),
        Tool(
            name="calculate_shipping",
            description="计算最优配送方案及费用",
            inputSchema={
                "type": "object",
                "properties": {
                    "destination": {
                        "type": "object",
                        "properties": {
                            "province": {"type": "string"},
                            "city": {"type": "string"},
                            "district": {"type": "string"},
                            "address": {"type": "string"}
                        },
                        "required": ["province", "city"]
                    },
                    "weight_kg": {"type": "number", "minimum": 0.1, "maximum": 100},
                    "sku_list": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["destination", "weight_kg"]
            }
        )
    ]

处理工具调用
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    if name == "get_order_status":
        return await handle_order_status(**arguments)
    elif name == "check_inventory":
        return await handle_inventory_check(**arguments)
    elif name == "calculate_shipping":
        return await handle_shipping_calculation(**arguments)
    else:
        raise ValueError(f"Unknown tool: {name}")

async def handle_order_status(order_id: str, include_history: bool = False):
    # 这里连接你的订单系统
    # 使用 HolySheep AI API 进行日志分析
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.holysheep.ai/v1/mcp/tools/order_status",
            params={"order_id": order_id, "history": include_history},
            headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}
        )
    return CallToolResult(content=[{"type": "text", "text": response.text}])

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__":
    asyncio.run(main())

四、资源订阅与实时通知

这是最容易出错的部分。当 AI Agent 需要监听数据变化时（如库存扣减、新订单通知），需要正确实现资源订阅机制。

# resources/product.py
from mcp.types import Resource, ResourceTemplate
from typing import AsyncIterator

静态资源：产品目录
PRODUCT_CATALOG = Resource(
    uri="order://products/catalog",
    name="Product Catalog",
    description="完整的产品目录信息",
    mimeType="application/json"
)

动态资源模板：单个产品
PRODUCT_RESOURCE_TEMPLATE = ResourceTemplate(
    uriTemplate="order://products/{sku}",
    name="Product by SKU",
    description="根据 SKU 查询产品详情，支持实时库存更新"
)

资源列表
@app.list_resources()
async def list_resources() -> list[Resource]:
    return [PRODUCT_CATALOG]

资源模板订阅
@app.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
    return [PRODUCT_RESOURCE_TEMPLATE]

读取资源内容
@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "order://products/catalog":
        return await fetch_full_catalog()
    elif uri.startswith("order://products/"):
        sku = uri.split("/")[-1]
        return await fetch_product_by_sku(sku)
    raise ValueError(f"Unknown resource: {uri}")

关键：实现订阅机制，这是报错 "ConnectionError: timeout" 的根源
@app.subscribe_resource()
async def subscribe_resource(uri: str) -> AsyncIterator[str]:
    """
    实时推送资源变更
    这里必须实现真正的长连接，否则 30 秒后必然超时
    """
    sku = uri.split("/")[-1]
    queue = asyncio.Queue()
    
    # 注册变更回调
    def on_change(event):
        asyncio.create_task(queue.put(event))
    
    inventory_service.subscribe(sku, on_change)
    
    try:
        while True:
            # 使用 wait_for 设置合理的超时时间
            event = await asyncio.wait_for(queue.get(), timeout=25.0)
            yield f"data: {json.dumps(event)}\n\n"
    except asyncio.TimeoutError:
        # 超时前发送心跳，保持连接活跃
        yield "data: {\"type\": \"heartbeat\"}\n\n"
    finally:
        inventory_service.unsubscribe(sku, on_change)

五、集成 HolySheep AI 网关

将你的 MCP Server 接入 HolySheep AI 平台，享受国内专属的高速通道。我实测的延迟数据：

• 北京节点 → HolySheep API：28ms
• 上海节点 → HolySheep API：35ms
• 广州节点 → HolySheep API：42ms

# mcp_client_example.py
import asyncio
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client

async def connect_to_order_service():
    # 启动你的 MCP Server 作为子进程
    process = await asyncio.create_subprocess_exec(
        "python", "server.py",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE
    )
    
    # 通过 stdio 连接到 MCP Server
    async with stdio_client(process) 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]}")
            
            # 调用工具
            result = await session.call_tool(
                "get_order_status",
                {"order_id": "ORD-20260305-00001", "include_history": True}
            )
            print(result.content[0].text)
            
            # 订阅资源变更
            async for update in session.subscribe_resource("order://products/SKU-12345"):
                print(f"库存更新: {update}")

asyncio.run(connect_to_order_service())

六、生产环境部署

实际生产中，建议使用进程管理器和反向代理：

# docker-compose.yml
version: '3.8'
services:
  mcp-order-service:
    build: .
    restart: always
    environment:
      - HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
      - LOG_LEVEL=INFO
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    
  nginx:
    image: nginx:alpine
    ports:
      - "8080:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - mcp-order-service

nginx.conf 配置 WebSocket 转发
events {
    worker_connections 1024;
}

http {
    upstream mcp_backend {
        server mcp-order-service:8000;
    }
    
    server {
        listen 80;
        
        # 标准 HTTP 转发
        location / {
            proxy_pass http://mcp_backend;
            proxy_set_header Host $host;
        }
        
        # MCP SSE 长连接支持
        location /mcp/stream {
            proxy_pass http://mcp_backend;
            proxy_set_header Host $host;
            proxy_buffering off;
            proxy_cache off;
            proxy_read_timeout 86400s;
            proxy_send_timeout 86400s;
            chunked_transfer_encoding on;
        }
    }
}

常见报错排查

错误 1：ConnectionError: timeout after 30000ms

原因：资源订阅未正确实现心跳机制，长连接在 30 秒后被代理或客户端断开。

# ❌ 错误示例：没有心跳的订阅
async def subscribe_resource_bad(uri: str):
    while True:
        event = await queue.get()  # 永远阻塞，30秒后超时
        yield f"data: {event}\n\n"

✅ 正确实现：带心跳的订阅
async def subscribe_resource_correct(uri: str) -> AsyncIterator[str]:
    queue = asyncio.Queue()
    last_heartbeat = time.time()
    
    while True:
        try:
            # 最多等待 25 秒，避免触发客户端超时
            event = await asyncio.wait_for(queue.get(), timeout=25.0)
            last_heartbeat = time.time()
            yield f"data: {json.dumps(event)}\n\n"
        except asyncio.TimeoutError:
            # 发送心跳，保持连接活跃
            yield "data: {\"type\": \"heartbeat\", \"timestamp\": " + str(int(time.time())) + "}\n\n"
            
            # 兜底：如果超过 5 分钟没有数据，断开连接
            if time.time() - last_heartbeat > 300:
                break

错误 2：401 Unauthorized

原因：请求头中缺少或使用了错误的 API Key。HolySheep AI 使用 Bearer Token 认证。

# ❌ 错误：使用错误的认证方式
response = await client.get(
    url,
    headers={"X-API-Key": "YOUR_KEY"}  # HolySheep 不支持这种格式
)

✅ 正确：Bearer Token 格式
response = await client.get(
    url,
    headers={"Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY')}"}
)

或使用 SDK（推荐）
from holysheep import HolySheepClient

client = HolySheepClient(api_key=os.environ.get('HOLYSHEEP_API_KEY'))
result = client.mcp.call_tool("get_order_status", order_id="ORD-123")

错误 3：tool not found

原因：工具注册时机错误或在 AI 模型未同步最新工具列表。

# ❌ 错误：在请求处理中动态注册工具
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name not in registered_tools:  # 此时 AI 可能已经缓存了工具列表
        await refresh_tools()  # 太晚了！
    ...

✅ 正确：在服务启动时预注册，并支持动态刷新
registered_tools = set()

@app.list_tools()
async def list_tools() -> list[Tool]:
    # 启动时加载
    return [TOOL_DEFINITIONS[name] for name in registered_tools]

async def refresh_tools():
    """提供独立的刷新接口，供外部调用"""
    global registered_tools
    new_tools = await fetch_tool_definitions()
    registered_tools = {t.name for t in new_tools}
    # 通知连接的客户端刷新

错误 4：JSON-RPC 解析失败

原因：响应格式不符合 JSON-RPC 2.0 规范。

# ❌ 错误：直接返回文本
return CallToolResult(content=[{"type": "text", "text": "操作成功"}])

✅ 正确：使用结构化结果
from mcp.types import TextContent, ImageContent

return CallToolResult(
    content=[
        TextContent(
            type="text",
            text="操作成功",
            annotations={"status": "success", "order_id": "ORD-123"}
        )
    ],
    isError=False
)

返回错误时
return CallToolResult(
    content=[TextContent(type="text", text="库存不足")],
    isError=True
)

七、性能优化建议

在我司的实际生产环境中，单个 MCP Server 支撑了日均 50 万次工具调用。以下是关键优化点：

• 连接池复用：使用 httpx.AsyncClient 而非每次请求创建新连接，QPS 提升 3 倍
• 结果缓存：对于不要求实时性的查询（如产品目录），增加 5-30 秒本地缓存
• 批量接口：将多个 SKU 查询合并为一次 API 调用，减少网络往返
• 优雅降级：外部服务不可用时返回缓存数据或友好提示，而非直接报错

八、价格与成本对比

使用 HolySheep AI 接入 MCP Server 的成本优势非常明显：

服务商	汇率	GPT-4.1 Input	DeepSeek V3.2 Output
官方 OpenAI	¥7.3/$1	$2.50/MTok	-
官方 Anthropic	¥7.3/$1	$3.00/MTok	-
HolySheep AI	¥1=$1	$2.50/MTok	$0.42/MTok

以我司月均 1000 万 Token 消耗计算，使用 HolySheep AI 相比官方渠道节省超过 85% 费用。充值支持微信/支付宝，实时到账。

总结

MCP Server 开发的核心在于三件事：清晰的工具 Schema 定义、可靠的资源订阅机制、规范的 JSON-RPC 响应格式。大多数报错都可以通过检查这三点解决。

我强烈建议国内开发者选择 HolySheep AI 作为 MCP 网关：低于 50ms 的响应延迟、¥1=$1 的无损汇率、以及完善的充值体系，能让你的 AI Agent 开发体验提升一个档次。

下一步你可以尝试将 MCP Server 部署到私有化环境，并接入 HolySheep AI 的监控面板，实时观察工具调用成功率和延迟分布。

👉 免费注册 HolySheep AI，获取首月赠额度