MCP Server 开发实战：从零构建一个生产级 MCP 服务

2026-06-10MCP开发实战协议

快速上手：5 分钟写一个 MCP Server

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

server = Server("weather-server")

@server.tool()
async def get_weather(city: str, unit: str = "celsius") -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称，如 "北京"、"上海"
        unit: 温度单位，celsius 或 fahrenheit
    """
    # 实际调用天气API...
    return f"{city}: 晴，28°C，湿度 45%"

@server.tool()
async def get_forecast(city: str, days: int = 3) -> dict:
    """获取未来几天天气预报"""
    return {"city": city, "forecast": [...]}

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write)

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

在 Claude Desktop 中配置

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["path/to/server.py"],
      "env": { "API_KEY": "xxx" }
    }
  }
}

MCP Server 的生产级考量

1. 错误处理与重试

外部 API 不可靠是常态。每个工具调用都应有超时、重试、降级策略：

@server.tool()
async def get_weather(city: str):
    for attempt in range(3):
        try:
            async with httpx.AsyncClient(timeout=5) as client:
                resp = await client.get(f"https://api.weather.com/{city}")
                return resp.json()
        except httpx.TimeoutException:
            if attempt == 2:
                return {"error": "天气服务暂时不可用，请稍后重试"}
            await asyncio.sleep(2 ** attempt)

2. 认证与权限

生产环境的 MCP Server 需要处理认证：

  • API Key 注入：通过环境变量传入，不在代码中硬编码
  • 工具级权限：危险操作（写操作）可设置 require_confirmation
  • 用户隔离：多租户场景下按 user_id 隔离数据和配额

3. 资源与工具的正确设计

类型何时用示例
Tool有副作用的操作、需要参数的计算发送邮件、创建订单、查询数据库
Resource只读数据、可以被 LLM 直接消费的内容文件内容、知识库文档、配置项
Prompt可复用的提示模板代码审查模板、SQL优化模板

4. 日志与可观测性

stdio 传输模式下，stderr 是唯一的日志通道（stdout 被协议占用）：

import logging
logging.basicConfig(
    level=logging.INFO,
    stream=sys.stderr,  # 注意：输出到 stderr
    format="%(asctime)s [%(levelname)s] %(message)s"
)

MCP Server 的 SSE 模式

stdio 适合本地，SSE 适合远程服务和多客户端：

# SSE Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

transport = SseServerTransport("/messages")

async def handle_sse(request):
    async with transport.connect_sse(request.scope, request.receive, request._send) as streams:
        await server.run(streams[0], streams[1])

app = Starlette(routes=[
    Route("/sse", endpoint=handle_sse),
    Route("/messages", endpoint=transport.handle_post_message, methods=["POST"]),
])

MCP 与其他协议的对比

协议定位适用场景
MCPLLM ↔ 工具/数据标准化Agent 工具集成
A2A (Agent-to-Agent)Agent ↔ Agent 通信多Agent协作
OpenAPI/SwaggerHTTP REST API 描述传统微服务
gRPC高性能 RPC内部服务通信

常见坑与对策

  • 坑：stdio 缓冲区阻塞 — 大量输出时 stdout 缓冲区满导致 Server 卡死。对策：及时 flush；分批返回大数据
  • 坑：SSE 连接断开无感知 — SSE 是单向的，Server 无法主动知道 Client 离线。对策：加心跳机制
  • 坑：工具定义 Schema 过于宽松 — LLM 容易"钻空子"传入意料外的参数。对策：尽量用 enum、pattern 等约束
  • 坑：Server 进程泄露 — 子进程 MCP Server 未正确清理。对策：Host 退出时给子进程发 SIGTERM，设置进程组

← 返回首页

未标记