云计算百科
云计算领域专业知识百科平台

MCP Server 教程:从零构建一个自定义工具服务器(2026 最新)

请添加图片描述

TL;DR

MCP(Model Context Protocol)让 AI 模型能够安全调用外部工具。本文从零开始,用 Python 构建一个完整的 MCP Server,包含自定义工具注册、请求处理、错误处理,并接入 Claude Desktop 进行测试。全文代码可直接运行。

1. MCP 是什么?为什么要自己写 Server?

MCP(Model Context Protocol)是 Anthropic 提出的一种开放协议,定义了 AI 模型与外部工具之间的标准化接口。你可以把它理解成 AI 世界的 USB 协议——只要双方都遵守同一套规范,任何模型都可以无缝接入任何工具。简单说,MCP 让 LLM 不再只是"聊天的模型",而是可以调用真实世界的 API、数据库、文件系统来完成任务的 Agent。

MCP 的核心架构分为三层:MCP Server(工具提供方)、MCP Client(通常是 Claude Desktop 或其他 AI 应用)和 LLM Host。Server 通过 JSON-RPC 协议暴露工具列表和调用接口,Client 将这些工具注册给 LLM,LLM 在推理过程中自主决定何时调用哪个工具。整个过程对用户来说是透明的——你只看到模型突然说"我来查一下天气",然后它确实查了。

常见的 MCP Server 场景包括:

  • 查询数据库并返回结构化结果
  • 调用外部 API(天气、搜索、翻译)
  • 操作本地文件系统
  • 执行自定义计算逻辑
  • 发送消息到 Slack 或邮件

官方和社区已经提供了大量现成的 MCP Server——SQL 数据库连接器、文件系统工具、GitHub 集成等。但当你需要一个高度定制化的工具时,比如连接内部业务系统、封装专有算法或对接公司 API,就绕不开自己写 Server 这条路。而且理解 MCP 的内部机制,对日常调试"工具为什么不响应"以及优化工具触发策略也至关重要。

2. 理解 MCP Server 的通信模型

在动手写代码之前,有必要理解 MCP 的通信流程,这对后面的调试会有很大帮助。

MCP 使用 JSON-RPC 2.0 协议进行通信。一个典型的工具调用生命周期如下:

  • 初始化阶段:Client 发送 initialize 请求,Server 回复协议版本和能力声明
  • 工具发现:Client 调用 tools/list,Server 返回注册的所有工具及其参数 Schema
  • 工具调用:LLM 决定使用某个工具后,Client 发送 tools/call 请求,携带工具名和参数
  • 结果返回:Server 执行完毕,返回 TextContent 或 ImageContent 等结果
  • 保持连接:Server 进入事件循环,等待下一个请求
  • 整个通信走的是双工流——Server 和 Client 可以同时发送和接收消息。在 stdio 模式下,这些 JSON-RPC 消息通过标准输入输出传递,非常轻量。

    3. 环境准备

    首先安装 MCP Python SDK:

    pip install mcp

    SDK 版本要求 >= 1.0.0。建议使用 Python 3.10+。

    4. 构建第一个 MCP Server

    下面是一个完整的 MCP Server,它提供了三个工具:一个加法计算器、一个天气查询模拟、和一个 Markdown 格式化输出。

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

    # 1. 创建 Server 实例
    server = Server("my-custom-tools")

    # 2. 注册工具列表
    @server.list_tools()
    async def handle_list_tools() > list[types.Tool]:
    return [
    types.Tool(
    name="add",
    description="计算两个数字的和",
    inputSchema={
    "type": "object",
    "properties": {
    "a": {"type": "number", "description": "第一个加数"},
    "b": {"type": "number", "description": "第二个加数"},
    },
    "required": ["a", "b"],
    },
    ),
    types.Tool(
    name="get_weather",
    description="查询指定城市的模拟天气",
    inputSchema={
    "type": "object",
    "properties": {
    "city": {"type": "string", "description": "城市名称"},
    },
    "required": ["city"],
    },
    ),
    types.Tool(
    name="format_markdown_table",
    description="将 JSON 数据格式化为 Markdown 表格",
    inputSchema={
    "type": "object",
    "properties": {
    "data": {
    "type": "array",
    "items": {"type": "object"},
    "description": "要格式化的数据数组",
    },
    "headers": {
    "type": "array",
    "items": {"type": "string"},
    "description": "表头字段名列表",
    },
    },
    "required": ["data", "headers"],
    },
    ),
    ]

    # 3. 实现工具调用逻辑
    @server.call_tool()
    async def handle_call_tool(
    name: str, arguments: dict
    ) > list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name == "add":
    result = arguments["a"] + arguments["b"]
    return [types.TextContent(type="text", text=str(result))]

    elif name == "get_weather":
    city = arguments["city"]
    # 模拟天气数据
    weather_data = {
    "北京": {"温度": 28, "天气": "晴", "湿度": "45%"},
    "上海": {"温度": 32, "天气": "多云", "湿度": "65%"},
    "深圳": {"温度": 30, "天气": "阵雨", "湿度": "78%"},
    "杭州": {"温度": 26, "天气": "阴", "湿度": "70%"},
    }
    info = weather_data.get(city, {"温度": "N/A", "天气": "未知", "湿度": "N/A"})
    text = f"📍 {city}\\n🌡 温度: {info['温度']}°C\\n☁ 天气: {info['天气']}\\n💧 湿度: {info['湿度']}"
    return [types.TextContent(type="text", text=text)]

    elif name == "format_markdown_table":
    data = arguments["data"]
    headers = arguments["headers"]
    if not data or not headers:
    return [types.TextContent(type="text", text="(空数据)")]

    # 生成表头
    header_row = "| " + " | ".join(headers) + " |"
    separator = "| " + " | ".join(["—"] * len(headers)) + " |"
    rows = []
    for item in data:
    row = "| " + " | ".join(str(item.get(h, "")) for h in headers) + " |"
    rows.append(row)

    table = header_row + "\\n" + separator + "\\n" + "\\n".join(rows)
    return [types.TextContent(type="text", text=table)]

    else:
    raise ValueError(f"未知工具: {name}")

    # 4. 启动 Server
    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="my-custom-tools",
    server_version="0.1.0",
    capabilities=server.get_capabilities(
    notification_options=NotificationOptions(),
    experimental_capabilities={},
    ),
    ),
    )

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

    5. 测试与接入

    启动 Server:

    python server.py

    你会看到进程在 stdio 模式下等待 MCP 客户端的连接。MCP 支持两种传输方式:stdio(标准输入输出)和 SSE(Server-Sent Events)。开发阶段推荐用 stdio,生产环境考虑 SSE。

    接入 Claude Desktop

    要把它接入 Claude Desktop,编辑你的 claude_desktop_config.json(通常位于 ~/Library/Application Support/Claude/):

    {
    "mcpServers": {
    "my-custom-tools": {
    "command": "python",
    "args": ["/path/to/your/server.py"]
    }
    }
    }

    重启 Claude Desktop,点击输入框旁的锤子图标,你就能看到注册的三个工具。在对话中试试这样问:

    • “用 my-custom-tools 帮我计算 1234 + 5678”
    • “北京今天天气怎么样?”
    • “把这段数据做成表格:{…}”

    Claude 会自动识别意图并调用对应的工具。如果 Claude 没有主动调用,你可以在 prompt 里明确要求:“请使用 my-custom-tools 中的工具来完成这个任务”。

    用 Python 客户端测试(不依赖 Claude Desktop)

    如果你想在纯代码环境验证 Server,可以写一个简单的测试客户端:

    # test_client.py
    import asyncio
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client

    async def test():
    server_params = StdioServerParameters(
    command="python",
    args=["server.py"],
    )
    async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
    await session.initialize()
    tools = await session.list_tools()
    print("可用工具:", [t.name for t in tools.tools])

    result = await session.call_tool("add", {"a": 1234, "b": 5678})
    print("add 结果:", result.content[0].text)

    result = await session.call_tool("get_weather", {"city": "北京"})
    print("天气结果:", result.content[0].text)

    asyncio.run(test())

    运行 python test_client.py,你会看到工具列表和调用结果直接打印出来。这种方式特别适合 CI 测试和调试。

    6. 踩坑与最佳实践

    踩坑 1:inputSchema 格式要严格

    MCP 使用 JSON Schema 来描述参数,字段名必须是驼峰式(inputSchema 而不是 input_schema)。参数类型用 number(而非 int),因为 JSON 不区分整型和浮点。

    踩坑 2:错误处理要显式

    工具调用中如果发生异常,一定要 raise ValueError(或捕获后返回错误信息文本),否则 MCP 客户端会收到一个无法解析的错误,导致模型认为工具不可用。

    踩坑 3:工具名不要用中文

    MCP 协议层面工具名必须是小写字母+下划线。中文名可能在部分客户端中无法正常调用。

    踩坑 4:返回格式统一用 TextContent

    call_tool 的返回值必须是 list[types.TextContent | types.ImageContent | types.EmbeddedResource]。最简单的做法是始终返回 [types.TextContent(type="text", text=str(result))]。

    最佳实践清单

    • 每个工具给出清晰、完整的中文 description,LLM 靠 description 决定是否调用
    • inputSchema 中每个字段都加 description,尤其是枚举值和格式约束
    • 工具数量控制在 5-8 个以内,太多会让模型选择困难
    • 耗时操作(网络请求)加上超时和重试逻辑
    • 敏感操作(写文件、执行命令)在工具内加确认机制

    7. 扩展方向

    上面的例子只是起点。你可以进一步:

    • 接入真实 API(如天气 API、搜索 API)
    • 连接 SQLite/PostgreSQL 数据库,让模型直接查询
    • 实现文件读写工具,让模型能管理本地文件
    • 组合多个 MCP Server,形成工具网络

    MCP 的价值在于它把"让模型用工具"这件事标准化了。一旦你掌握了写 Server 的方法,就等于给你的 AI Agent 开了无限可能。

    参考资料

    • MCP 官方文档
    • MCP Python SDK GitHub
    • MCP 快速入门
    赞(0)
    未经允许不得转载:网硕互联帮助中心 » MCP Server 教程:从零构建一个自定义工具服务器(2026 最新)
    分享到: 更多 (0)

    评论 抢沙发

    评论前必须登录!