从零构建MCP Server:Model Context Protocol开发完全指南

引言

随着大语言模型(LLM)能力的飞速发展,如何让模型安全、高效地连接外部世界——比如数据库、API、文件系统——成为 AI 应用落地的关键瓶颈。Anthropic 推出的 Model Context Protocol (MCP) 正是为了解决这一难题而生的开放标准。它将 LLM 与外部工具、数据源的集成规范化,让开发者可以用一种统一的方式“教会”模型使用各种外部能力。

本文将带你从零开始,掌握 MCP Server 的开发流程:从核心概念梳理,到编写第一个完整可运行的服务器,再到配置客户端进行真实交互,最后分享生产环境中的常见问题与最佳实践。读完这篇文章,你将具备独立开发并部署 MCP Server 的能力。

MCP 核心概念

在动手编写代码之前,我们需要理解 MCP 的架构与关键原语。

1. 客户端-服务器架构

MCP 采用经典的 C/S 模型:

  • MCP Host:发起连接的一方,通常是 Claude Desktop、IDE 插件或你自己的 AI 应用;
  • MCP Client:运行在 Host 内部,负责与 Server 维持一对一连接;
  • MCP Server:你将要开发的部分,提供具体的工具、资源和提示等能力。

一个典型的交互流程如下:

  1. Host 启动内嵌的 Client,Client 与 Server 建立连接(如 stdio 或 HTTP);
  2. Client 请求 Server 的能力列表(工具、资源……);
  3. 用户通过 Host 与模型对话,模型根据需求调用 Server 提供的工具;
  4. Server 执行工具并返回结果,模型汇总后回应用户。

2. 三大核心原语

MCP 定义了三种服务器可暴露的能力:

  • 工具(Tools):类似 API 端点,模型可以调用它们来执行操作,例如查询天气、发送邮件、读写文件。每个工具由名称、描述和输入模式(JSON Schema)构成;
  • 资源(Resources):暴露只读数据,如文件内容或数据库记录。资源用 URI 标识,支持订阅和内容变更通知;
  • 提示(Prompts):预定义的对话模板,可帮助用户快速开启特定任务。

本教程将重点放在工具的构建上,因为它是绝大多数开发场景的核心。

3. 传输机制

MCP 支持两种主流传输方式:

  • stdio:通过标准输入输出通信,简单直接,特别适合本地进程间调用(如 Claude Desktop 集成);
  • Server-Sent Events (SSE):基于 HTTP 的流式传输,允许远程连接,适用于将 Server 部署为服务供多个客户端使用。

我们将首先使用 stdio 进行演示,它无需任何网络配置,极易上手。

实战示例:构建一个天气查询 MCP Server

下面我们使用 Python 和最新的 mcp 库,一步步实现一个能够查询天气的工具服务器,并配置到 Claude Desktop 中测试。

1. 环境准备

确保你已安装 Python 3.10+,然后创建项目目录并安装依赖:

mkdir mcp-weather-server && cd mcp-weather-server
python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate
pip install mcp

mcp 库由 Anthropic 官方维护,提供了编写 Server 和 Client 的全套基础设施。

2. 编写服务器代码

创建 weather_server.py,填入以下完整代码(包含详细注释):

#!/usr/bin/env python3
"""
一个简单的 MCP Server:提供天气查询工具。
使用 stdio 传输,可直接接入 Claude Desktop。
"""
import asyncio
import json
from typing import Any

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

# 1. 创建 Server 实例,名称用于标识
app = Server("weather-server")

# 2. 声明工具列表 —— 告诉客户端我们能做什么
@app.list_tools()
async def list_tools() -> list[Tool]:
    """返回当前服务器支持的所有工具定义"""
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的当前天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,例如 Beijing"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位,默认为摄氏度"
                    }
                },
                "required": ["city"]
            }
        )
    ]

# 3. 处理工具调用 —— 真正执行逻辑的地方
@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """根据工具名称和参数执行操作,返回结果"""
    if name == "get_weather":
        city = arguments.get("city", "Unknown")
        unit = arguments.get("unit", "celsius")

        # 模拟天气数据(实际应用中可替换为真实 API 调用)
        weather = {
            "Beijing": {"temp": 22, "condition": "晴"},
            "Shanghai": {"temp": 28, "condition": "多云"},
            "Guangzhou": {"temp": 30, "condition": "雷阵雨"},
        }
        info = weather.get(city, {"temp": 20, "condition": "未知"})
        temp = info["temp"]
        condition = info["condition"]

        if unit == "fahrenheit":
            temp = temp * 9 / 5 + 32

        result_text = f"{city}当前天气:{condition},温度{temp:.1f}{'℉' if unit == 'fahrenheit' else '℃'}"
        return [TextContent(type="text", text=result_text)]

    # 未知工具应抛出明确错误
    raise ValueError(f"未知工具: {name}")

# 4. 主函数,启动 stdio 传输并运行服务器
async def main():
    """使用 stdio 协议启动服务器"""
    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())

3. 运行与测试

直接用 Python 运行即可启动一个 stdio MCP Server:

python weather_server.py

此时服务器会阻塞等待标准输入,这符合 stdio 传输的特性。为了验证工具是否工作正常,我们可以使用 MCP 自带的开发者工具 mcp dev(需额外安装 mcp[cli]):

pip install 'mcp[cli]'
mcp dev weather_server.py

该命令会启动一个本地 web 界面,让你可视化地列出工具、发起调用并查看结果。推荐在开发阶段使用。

4. 接入 Claude Desktop

最实用的场景是将 MCP Server 配置到 Claude Desktop 中,让模型直接调用你的工具。

  1. 打开 Claude Desktop 的配置文件:
    - macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
    - Windows: %APPDATA%\Claude\claude_desktop_config.json

  2. 编辑文件,添加 mcpServers 段:

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/你的绝对路径/mcp-weather-server/weather_server.py"]
    }
  }
}
  1. 保存并重启 Claude Desktop,在对话中输入:“帮我查一下北京今天天气怎么样?”。如果一切正常,Claude 会自动调用 get_weather 工具并展示返回结果。

如果日志发现连接错误,请检查 Python 路径和文件权限。

进阶:添加资源与错误处理

暴露静态资源

除了工具,你也可以公开只读资源。例如,为上面的天气服务器添加一个“城市列表”资源:

from mcp.types import Resource, ResourceContents

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="weather://cities.json",
            name="支持的城市列表",
            description="当前系统支持查询天气的城市",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> ResourceContents:
    if uri == "weather://cities.json":
        cities = ["Beijing", "Shanghai", "Guangzhou"]
        return ResourceContents(
            uri=uri,
            mimeType="application/json",
            text=json.dumps(cities, ensure_ascii=False)
        )
    raise ValueError("资源不存在")

这样客户端就可以通过读取资源获取城市列表,用于下拉选择等 UI 组件。

健壮的错误处理

生产环境中,务必捕获异常并返回结构化错误,而非让服务器崩溃:

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        if name == "get_weather":
            city = arguments.get("city")
            if not city:
                raise ValueError("缺少参数 'city'")
            # ... 业务逻辑
    except Exception as e:
        # 返回错误消息,模型会据此理解并做出相应反馈
        return [TextContent(type="text", text=f"工具调用失败:{str(e)}")]

模型接收到错误字符串后,通常会向用户解释问题或调整参数重试。

常见问题与注意事项

1. stdio vs SSE,如何选择?

  • stdio:最简单,无需网络配置,适合本地工具和 Claude Desktop 集成。缺点是只能被一个客户端连接;
  • SSE:支持远程访问和多个客户端,需要实现一个 HTTP 服务器并处理跨域。当你的服务需要被多个用户或远程 AI 应用调用时,使用 SSE 是更好的选择。

从 stdio 切换到 SSE,只需修改主函数:

import uvicorn
from mcp.server.sse import sse_server

if __name__ == "__main__":
    uvicorn.run(sse_server(app), host="0.0.0.0", port=8000)

然后客户端连接 http://your-server:8000/sse

2. 安全考量

  • 输入校验:永远不要信任模型传来的参数,必须进行类型检查和范围限制,防止注入攻击;
  • 权限控制:文件系统工具应限制可读写的目录范围;网络请求应做域名白名单;
  • 敏感信息:不要在工具描述或返回值中硬编码 API 密钥等敏感数据,应从环境变量读取。

3. 性能与资源管理

当工具涉及耗时操作(如调用外部 API)时,务必使用异步 I/O(async/await),避免阻塞事件循环。对于计算密集型任务,考虑放在线程池或子进程中执行。

4. 版本兼容与更新

MCP 协议仍在快速演进,你需要留意 mcp 库的更新日志。升级前在测试环境验证,避免破坏现有集成。

总结

本文从 MCP 的基本架构讲起,通过一个完整的天气查询服务器实例,演示了如何声明工具、处理调用,并将服务接入 Claude Desktop。我们进一步探讨了资源暴露、错误处理、传输选择以及安全实践。

MCP 的价值在于标准化——它让 LLM 与外部世界的交互不再是千篇一律的硬编码插件,而是可复用、可组合的模块。掌握了 MCP Server 的开发,你就拥有了一把钥匙,可以安全、优雅地为 AI 应用打开通往无限数据与功能的大门。

未来,随着生态的完善,MCP 有望成为 AI 工具集成的事实标准。现在开始动手实践,你将站在这股浪潮的最前沿。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐