MCP 协议原理与自定义服务器开发实战：从零搭建 AI 工具调用网关

在一个内部工具平台项目中，我需要让 Claude 能直接查询公司内部的 API 和数据源，而不是每次都让用户复制粘贴结果。调研后我选了 MCP（Model Context Protocol）作为标准化方案，花了两天从读协议到跑通第一个自定义服务器。

问题是什么

让 AI 模型调用外部工具，传统上有两条路：一是 Function Calling，每个模型各自定义一套 JSON Schema；二是直接把工具逻辑塞进 System Prompt。前者绑定厂商，后者不可维护。我需要一个与模型无关、与传输层无关的标准协议层，让 AI 应用和后端服务之间能像 HTTP 一样松耦合。

我当时面对的是：一个内部运维系统，有 20+ 个 REST API 要暴露给 AI，涉及用户认证、敏感数据过滤、异步任务调度。直接用 OpenAI Function Calling，换模型就得重写集成层；用 LangChain 工具又太重。

解决思路

MCP 的核心设计可以理解为"AI 世界的 USB-C 接口"——它定义了客户端（AI 应用）和服务器（工具提供方）之间的标准通信协议。我对比了三种方案：

MCP 的核心概念只有 4 个：

• Resources（资源）：暴露给 AI 的数据，类似 REST 的 GET
• Tools（工具）：AI 可以调用的操作，类似 REST 的 POST
• Prompts（提示）：预定义的提示模板
• Transport（传输）：Stdio 或 SSE，通信的具体方式

我选了基于 Python 的 mcp 官方 SDK，用 Stdio 传输做本地开发测试。

操作步骤

步骤1：安装 SDK 并初始化项目

mkdir mcp-server-demo && cd mcp-server-demo
python -m venv .venv && source .venv/bin/activate
pip install mcp httpx

MCP 官方 SDK 目前支持 Python 和 TypeScript。Python 版更成熟，TypeScript 版更新更活跃。我用 Python 起步，调试工具更顺手。

创建项目结构：

mcp-server-demo/
├── server.py          # MCP 服务器入口
├── tools/             # 工具实现
│   └── weather.py
└── requirements.txt

步骤2：实现第一个 MCP 服务器

# server.py
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio

server = Server("weather-server")

@server.list_tools()
async def handle_list_tools():
    return [
        {
            "name": "get_weather",
            "description": "获取指定城市的实时天气",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 Beijing、Shanghai"
                    }
                },
                "required": ["city"]
            }
        }
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments["city"]
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"{city} 当前天气：多云，25°C，湿度 60%"
                }
            ]
        }

async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream,
            InitializationOptions(
                server_name="weather-server",
                server_version="0.1.0"
            )
        )

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

核心就是 list_tools 和 call_tool 两个装饰器——它们是 MCP 协议层与业务逻辑之间的桥梁。协议本身自动处理握手、错误码、序列化，我只管实现工具。

步骤3：用 Inspector 调试工具

MCP 官方提供了 Inspector 调试工具，可视化查看工具列表和测试调用：

npx @anthropic-ai/mcp-inspector server.py

浏览器打开 http://localhost:5173，界面分三栏：

• 左栏：服务器提供的所有工具列表
• 中栏：选中工具的输入 Schema
• 右栏：调用结果和时序数据

注意: Inspector 默认走 Stdio 传输，它会自动启动你的 server.py 子进程并通过 stdin/stdout 通信。如果服务器依赖环境变量，启动 Inspector 前先 export。

调试时我踩了个坑：Inspector 的 --transport 参数默认走 Stdio，但生产环境需要走 SSE（Server-Sent Events）。SSE 模式启动：

npx @anthropic-ai/mcp-inspector --transport sse --sse-port 8000 server.py

步骤4：添加认证与上下文传递

真实场景中，工具调用需要带上用户身份。MCP 通过 initializationOptions 传递元数据：

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    request_context = server.request_context
    user_id = request_context.meta.get("user_id", "anonymous")
    
    print(f"[AUDIT] 用户 {user_id} 调用了 {name} 工具", flush=True)
    
    if name == "get_weather":
        city = arguments["city"]
        data = await fetch_weather(city)
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"{data['city']} 天气：{data['weather']}"
                }
            ],
            "meta": {
                "source": "internal-api",
                "cached": False
            }
        }

注意: MCP 的上下文传递在不同传输模式下表现不同。Stdio 模式需要自己管理状态，SSE 模式可以通过 HTTP Header 传递认证信息。我建议开发用 Stdio，生产用 SSE + JWT 认证。

步骤5：对接 Claude Desktop 做端到端测试

最后把服务器跑起来，让 Claude Desktop 能发现并调用：

# 启动 MCP 服务器（SSE 模式）
python server.py --transport sse --port 8000

在 Claude Desktop 的配置文件中注册：

{
  "mcpServers": {
    "weather-demo": {
      "command": "python",
      "args": ["server.py", "--transport", "sse", "--port", "8000"]
    }
  }
}

重启 Claude Desktop，在对话中输入"北京今天天气怎么样"，Claude 会自动发现 get_weather 工具并调用。从用户视角看就是一句话的事，背后 MCP 完成了协议握手、工具发现、参数校验、数据返回的全流程。

结果与总结

两天内我跑通了完整链路：从读协议到 Claude Desktop 实际调用自定义工具。MCP 的简洁程度超出预期——核心就 4 个概念、两个装饰器，比 Function Calling 更规范，比 LangChain 更轻量。

几个关键踩坑记录：

1. Stdio 模式只适合本地开发。服务器进程必须保持前台运行，没法做负载均衡。上线必须切 SSE。
2. 协议版本兼容问题。MCP 还在快速迭代，mcp SDK 的 v0.x API 变动频繁。我在 0.5.x 上写的代码升级到 0.7.x 改了三个接口名。建议锁定版本号。
3. 错误处理要细粒度。工具内部崩溃时，MCP 会返回通用错误码。我后来加了 try/except 包装层，把业务异常映射到 MCP 的错误码体系（InvalidParams、InternalError 等），这样客户端能区分"参数不对"和"服务挂了"。
4. 长耗时工具要异步。默认工具调用是同步等待的，如果某个工具执行超过 30 秒，客户端可能超时。我用 asyncio.create_task + 轮询模式处理长时间任务。

延伸思考

MCP 最让我兴奋的不是它现在的样子，而是它可能长成的样子：

• MCP Gateway：在公司内部架设一个 MCP 网关，所有业务团队的工具服务器注册到网关，统一做认证、限流、审计。这比每个 AI 应用各自集成一套工具层干净得多。
• MCP + OpenAPI 自动转换：我后面试着写了个脚本，从 Swagger 文档自动生成 MCP 工具定义。20 个接口的微服务，标注下安全等级就直接暴露给 AI 了。这个思路能极大降低接入成本。
• MCP 联邦协议：你问 AI"查一下上海最近的天气和明天的会议安排"，它同时调用了天气服务器和日历服务器，两个服务器分属不同的团队、不同的部署单元。MCP 的标准化让这种联邦查询成为可能，而不用每个服务都去对接每个 AI 平台。

如果你也在搭 AI 工具集成层，现在入 MCP 正好——生态初具雏形但还没固化，踩坑的人还不多，你有机会成为团队里最早吃螃蟹的那个。