MCP Server 开发实战:从 0 到 1 构建自己的工具服务
最近一直在折腾 MCP(Model Context Protocol,模型上下文协议),发现不少人一听“MCP Server”就头大,觉得又是啥高深框架。其实说白了,MCP Server 就是给 AI 模型开的一个“标准插头”,让模型能通过这个插头去调用你写的工具函数。今天我就拿一个实战项目,完整走一遍从环境搭建到服务上线的全过程,把我踩过的坑和验证过的方案都写出来。
这篇玩意适合谁看?如果你已经在用 Claude、Cursor 这类 AI 工具,但又觉得它们内置功能不够用,想把自己公司内部的数据、API、数据库能力“接入”到 AI 对话里,那么这篇文章就是给你准备的。哪怕你只是听说过 MCP、完全零基础,跟着我的步骤走,也能在半天内跑通一个属于自己的 MCP Server。
先说清楚两个最容易混淆的概念:MCP Host 和 MCP Server。
MCP Host 是发起会话的“主人”,比如 Claude Desktop、Cursor 或者你自己写的客户端程序。MCP Server 是“提供服务的一方”,负责把具体能力(查天气、查数据库、调第三方接口)包装成标准化的工具,等着 Host 来调用。Host 和 Server 之间通过 JSON-RPC 2.0 协议通信,Host 发请求,Server 处理请求并把结果返回。整个链路里还有一个隐形的 MCP Client,它内嵌在 Host 里面,负责和 Server 做协议交互。
明白了这个关系,后面写代码的时候很多困惑都会迎刃而解。
1. MCP 为什么值得自己动手写一个
我见过太多人问:直接用现成的 MCP server 不就行了?GitHub 上一搜一大把,干嘛还要从零写?
这话说对了一半。现成的服务器确实省事,但你需要面对三个现实问题:
第一,现成服务器的工具粒度不一定匹配你的需求。比如某个 GitHub MCP server 封装了几十个操作,但你只需要一个“获取 issue 列表”的功能,为了这一个操作引入几十个工具,不仅浪费 token,还增加了出错的概率。
第二,安全边界不好控制。现成的服务器跑在你的环境里,它内部怎么校验输入、怎么处理异常,你怎么知道?我见过某个开源 MCP server 直接把整个文件系统暴露给模型,一旦模型被提示注入攻击,后果不堪设想。
第三,你想接入的能力大概率是“你独有的”。你的公司内部 CRM 系统、你的私有数据集、你手上的付费 API,这些能力只有你自己能写封装层。指望开源社区帮你做好,不现实。
从 0 到 1 自己写,最大的好处就是“每一行代码你都知道在干什么”。MCP 协议本身不复杂,核心就几个概念:tool(工具)、resource(资源)、prompt(提示模板)、sampling(采样请求)。对大多数人来说,先把 tool 玩明白了就够用了。
MCP 协议之所以值得学,是因为它在 AI 工具集成领域逐渐变成了一个“标准插头”。以前每个 AI 应用都要自己定义一套工具调用格式,现在大家都在往 MCP 这个公共协议上靠。你学会一次,后面给 Claude、Cursor、自研应用接工具都能用同一套逻辑。
2. 开发前必须搞清楚的三个核心概念
2.1 MCP Host、MCP Client、MCP Server 的三角关系
很多教程把 MCP Client 和 MCP Host 混在一起说,初学者特别容易卡在这里。我把它们分成三层:
- Host:用户直接面对的应用层(Claude Desktop、IDE、Web 应用)。它负责渲染界面、管理会话上下文。
- Client:Host 内部嵌入的一个协议客户端模块。它的职责是维护与 Server 的连接、封装请求、解析响应。
- Server:独立进程,负责实现具体工具逻辑。
打个比方,Host 像个“餐厅”,Client 是“服务员”,Server 是“后厨”。你(用户)跟餐厅点菜,服务员把你的需求记下来传进后厨,后厨做完菜让服务员端出来。MCP 协议就是这套“传菜流程”的标准。
我自己一开始犯的错误是,以为写一个 MySQL MCP Server 就得自己处理 TCP 连接、JSON-RPC 编解码、SSE 传输。实际上 Python SDK 把这些底层工作全都封装好了,你需要做的只是定义工具函数、写清参数 schema、实现业务逻辑,剩下都有现成的。
2.2 理解 JSON-RPC 2.0 与传输模式
MCP 的请求响应基于 JSON-RPC 2.0,但你不需要自己解析这些报文。SDK 已经把 initialize、tools/list、tools/call 这些协议阶段处理好了。
传输模式上,当前主流有两种:
- stdio 模式:Server 作为子进程被 Host 拉起,通过标准输入输出通信。本地开发调试时用它最方便。
- Streamable HTTP / SSE 模式:Server 作为一个 HTTP 服务监听端口,Host 通过 HTTP 请求来调用,适合远程部署。
开发时我建议先用 stdio 模式把工具逻辑调通,再用 Streamable HTTP 模式部署到服务器上。两种模式在 SDK 里切换只需要改几行代码。
2.3 工具定义:schema 是关键
MCP 的 tool 定义依赖 JSON Schema 来描述入参。模型是靠这份 schema 来决定怎么调用你的工具的,所以 schema 写得好不好,直接决定了模型调用工具的准确率。
我写 schema 的经验就三条:参数名用全小写下划线、必填参数必须出现在 required 数组里、description 要写明参数的单位和边界值。比如一个查天气的工具,temperature_unit 这个参数的描述我会写成“温度单位,可选值为 celsius 或 fahrenheit,默认为 celsius”。描述越具体,模型越不容易传错值。
3. 实战准备:从零搭建 MCP Server 开发环境
3.1 环境依赖与版本选型
写 MCP Server 首选 Python,生态成熟、SDK 维护得勤快。我的开发环境是:
- Python 3.10.12(官方 MCP SDK 要求 3.10+)
- mcp 库,我写这篇文章时最新稳定版是 1.9.1
- uv,一个极快的 Python 包管理器,官方文档里强烈建议用它
国内网络环境下,直接用 pip 装 mcp 库大概率会遇到超时问题,建议先配置 pypi 镜像源:
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple pip install "mcp[cli]>=1.9.0"安装完成后验证一下版本:
mcp --version能打印出版本号就说明环境 OK 了。顺手说明一下,MCP 官方提供了一个mcp命令行工具,内置了 dev、install、deploy 等子命令,调试和发布都会用到。
3.2 初始化项目结构
我习惯把 MCP Server 做成一个标准 Python 包,方便后面用 uv 管理依赖、做离线打包。项目结构如下:
weather-mcp-server/ ├── pyproject.toml ├── README.md ├── src/ │ └── weather_server/ │ ├── __init__.py │ └── server.py └── tests/ └── test_server.py如果你图省事,也可以直接用官方脚手架生成:
uv init weather-server cd weather-server uv add mcp httpxhttpx用来请求第三方天气 API,MCP SDK 本身不绑定任何 HTTP 客户端。pyproject.toml 里至少需要声明 mcp 依赖和构建系统,uv 会自动生成一份能跑的基础配置。
3.3 MCP Inspector:开发调试神器
本地开发强烈建议用 MCP Inspector 调试,它是官方提供的一个可视化调试面板,能直接看到 Server 注册了哪些工具、模型调用工具时传了什么参数。
启动方式很简单,在项目根目录执行:
mcp dev src/weather_server/server.py这条命令会启动一个本地 Web 服务,默认端口 6274,浏览器打开就能看到调试界面。我每次写新工具,都用它先测一遍,确认参数解析和返回结果没问题,再去接入客户端。
注意,mcp dev默认走的是 stdio 传输模式。如果你想调试 HTTP 模式的 Server,得换成mcp dev --transport http并确保代码里用了对应的服务启动方式。
4. 完整实现一个天气查询 MCP Server
4.1 定义工具功能和入参 schema
为了让例子足够有代表性,我选了“天气查询”这个场景。它麻雀虽小五脏俱全:有外部 API 调用、有入参判断、有结果格式化,刚好把 MCP Server 的完整流程串起来。
我们这个 Server 提供两个工具:
get_weather:根据城市名获取实时天气get_forecast:根据城市名获取未来 3 天预报
每个工具都要定义入参 schema。我用 pydantic 模型来定义,SDK 会自动把它转成 JSON Schema。
from pydantic import BaseModel, Field class GetWeatherInput(BaseModel): city: str = Field(description="城市名称,例如:北京、上海、广州") unit: str = Field( default="celsius", description="温度单位,可选值为 celsius 或 fahrenheit", )这里要注意,description别写得太笼统。模型读这个字段来决定传什么参数,写得越细,调用成功率越高。
4.2 服务端主逻辑与工具注册
下面是服务端的主代码,我把关键部分都加了注释。这个文件放到src/weather_server/server.py:
import asyncio from typing import Any import httpx from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types from mcp.server import NotificationOptions, Server from pydantic import BaseModel, Field # 创建 Server 实例,标识符要起一个全局唯一的字符串 server = Server("weather-mcp-server") # 开放的天气 API,不需要 key 就能访问 OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast" # 一个简易的城市名到经纬度映射,真实项目建议接地理编码 API CITY_COORDS = { "北京": (39.9042, 116.4074), "上海": (31.2304, 121.4737), "广州": (23.1291, 113.2644), "深圳": (22.5431, 114.0579), "杭州": (30.2741, 120.1551), } class GetWeatherInput(BaseModel): city: str = Field(description="城市名称,目前支持北京、上海、广州、深圳、杭州") unit: str = Field( default="celsius", description="温度单位,celsius 或 fahrenheit", ) class GetForecastInput(BaseModel): city: str = Field(description="城市名称,目前支持北京、上海、广州、深圳、杭州") days: int = Field(default=3, ge=1, le=7, description="预报天数,1 到 7 之间") def _get_coords(city: str): """根据城市名获取经纬度,支持城市名后面带'市'的情况""" city = city.replace("市", "") if city not in CITY_COORDS: raise ValueError(f"暂不支持该城市:{city}") return CITY_COORDS[city] async def _fetch_weather(city: str, unit: str = "celsius") -> dict[str, Any]: """调用 open-meteo 接口获取实时天气""" lat, lon = _get_coords(city) params = { "latitude": lat, "longitude": lon, "current_weather": "true", "temperature_unit": unit, "timezone": "Asia/Shanghai", } async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(OPEN_METEO_URL, params=params) resp.raise_for_status() data = resp.json() current = data.get("current_weather", {}) return { "city": city, "temperature": current.get("temperature"), "windspeed": current.get("windspeed"), "weathercode": current.get("weathercode"), "time": current.get("time"), } # 注册工具:get_weather @server.list_tools() async def handle_list_tools() -> list[types.Tool]: return [ types.Tool( name="get_weather", description="获取指定城市当前的实时天气信息", inputSchema=GetWeatherInput.model_json_schema(), ), types.Tool( name="get_forecast", description="获取指定城市未来几天的天气预报", inputSchema=GetForecastInput.model_json_schema(), ), ] # 处理工具调用 @server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent]: if not arguments: arguments = {} if name == "get_weather": try: inp = GetWeatherInput(**arguments) result = await _fetch_weather(inp.city, inp.unit) except Exception as e: return [types.TextContent(type="text", text=f"调用失败:{str(e)}")] return [types.TextContent(type="text", text=str(result))] elif name == "get_forecast": # 类似实现,省略具体预报逻辑 return [types.TextContent(type="text", text="预报功能开发中")] else: raise ValueError(f"未知工具:{name}") # 标准入口:stdio 模式 async def run_stdio(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="weather-mcp-server", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if __name__ == "__main__": asyncio.run(run_stdio())这段代码里有几个关键点值得细讲。
@server.list_tools()这个装饰器告诉 MCP 协议“我这个服务提供哪些工具”。模型在对话之前,会先通过 tools/list 拿到完整的工具清单和入参 schema,所以这里返回的内容必须精确。
inputSchema用 pydantic 模型的model_json_schema()直接生成,保证 schema 和代码定义同步,不会出现改代码忘了改文档的情况。这是 pydantic v2 的写法,你要是还在用 v1,方法名是schema()。
@server.call_tool()是真正干活的地方。所有工具调用请求都会进到这个函数,先通过name参数区分调用的是哪个工具,再用 pydantic 模型做参数校验。校验失败时我选择把错误信息作为正常文本返回,而不是直接抛异常。原因是:模型能读到返回文本,它可以自行修正参数后再次调用,用户体验会好很多。
4.3 SDK 帮你做了哪些事
写第一版代码的时候,我总忍不住想去看 SDK 底层到底怎么处理协议交互的。后来我明白了,SDK 主要帮你做了三件事:
第一,协议握手。MCP 建立连接时双方要先交换 initialize 请求,确认协议版本和能力集。这段逻辑 SDK 已经封装在server.run()里面了,你只需要传入InitializationOptions。
第二,JSON-RPC 报文编解码。Host 发过来的请求可能是 JSON 字符串,SDK 会解析成结构化对象,然后根据方法名路由到你写的对应 handler 上。
第三,生命周期管理。连接关闭时清理资源、io stream 错误处理、请求超时中断,这些都属于协议工程的脏活累活,自己实现一遍很费时间,也没必要。
但 SDK 不会帮你做的是业务逻辑。你的工具函数内部怎么调用第三方 API、怎么处理错误、怎么格式化返回结果,这些都得你自己写。这部分正是 MCP Server 开发里最需要设计的地方。
4.4 如何验证服务是否正常
第一次写完代码,别急着接 Claude 或者 Cursor。先用 MCP Inspector 做一轮冒烟测试:
mcp dev src/weather_server/server.py打开 http://localhost:6274 之后,你会看到三个关键区域:
- Tools 列表:确认两个工具都注册成功了
- 工具调用面板:选择一个工具,填入参数,点击发送
- 调用结果区域:查看返回的 JSON 是否符合预期
我测试时发现的问题,十有八九都是 schema 写错了。比如 required 字段没生效、default 值类型不对、description 里有特殊符号,通过 Inspector 一眼就能看出来。
本地 stdio 模式测通了,再考虑远程部署。远程部署通常用 Streamable HTTP 模式,MCP Python SDK 从 1.8 版本开始提供了mcp.server.streamable_http模块,代码改动很小:
from mcp.server.streamable_http import streamable_http_server async def run_http(): async with streamable_http_server("/mcp") as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions(...) )部署到服务器之后,用mcp deploy或者直接用进程管理器跑起来,Host 端配置好对应的 URL 就能连上了。
5. 接入 MCP Host:让工具真正被 AI 用起来
5.1 用 MCP Inspector 验证服务
MCP Inspector 的完整路径我简单说下。服务运行后,按照第 4.4 节步骤启动 Inspector,页面上会显示当前 Server 暴露的 tools 列表。点进去能看到每个工具的 JSON Schema 定义。然后我们可以在调试面板输入参数,模拟一次完整调用。
这样做的价值在于,先跑通“Server 本身的正确性”,再接入 Host。否则等宿主端出了问题,你根本分不清是自己代码的问题还是 Host 配置的问题。
5.2 配置 Claude Desktop 接入本地服务
本地 stdio 模式的 Server 接入 Claude Desktop 是最常见的用法。编辑 Claude Desktop 配置文件(macOS 在~/Library/Application Support/Claude/claude_desktop_config.json,Windows 在%APPDATA%\Claude\claude_desktop_config.json),如果你用的是 Claude Code 也可以直接写入其配置文件:
{ "mcpServers": { "weather": { "command": "python", "args": ["/absolute/path/to/src/weather_server/server.py"], "env": {} } } }配置完记得重启 Claude Desktop,让配置生效。重启后,对话框旁边会出现一个工具图标,点开能看到已加载的工具列表。然后在对话里输入“北京现在多少度”,模型就会自动调用get_weather工具。
如果你用的是 Cursor,路径稍微不同:打开 Cursor 设置里的 MCP 配置,直接添加到对应的那栏。Cursor 对 stdio 和 HTTP 模式都支持,远程部署的 HTTP server 也可以直接填 URL 接入。
5.3 远程部署:从 localhost 到服务器
如果你想把这个 Server 部署到公司服务器上,让多个客户端共享,就得改用 HTTP 模式。部署步骤是:
- 在代码里实现 HTTP 模式启动函数(见 4.4 节)
- 用 gunicorn 或 uvicorn 跑起来
- 配置反向代理(Nginx/Caddy)绑定域名
- 客户端配置里填
https://your-domain.com/mcp
这里有个容易踩的坑:MCP 的 HTTP 模式要求客户端和服务端都有 CORS 支持。如果你用浏览器端的 MCP Client,服务端必须显式配置 CORS 白名单。在 MCP 的StreamableHTTPServer或自定义的 ASGI/Flask 应用中设置allow_origins参数,否则前端会报跨域错误。
我这个天气服务因为调用的是公开 API,没有鉴权需求。但如果是内部服务,必须在 Server 层加认证,最简单的方案是让 Host 在请求头里带一个 API Key,Server 启动时从环境变量读取并校验。这个在 MCP 当前版本没有内置支持,需要自己在事件循环里做拦截。
6. 常见问题与排查技巧实录
6.1 stdio 模式下服务启动失败
症状:Claude Desktop 里报 MCP 配置连接失败。
排查思路:先用命令行手动跑一遍python src/weather_server/server.py,看有没有报错。最常见的原因是依赖缺失。比如某个环境里只装了 mcp,但没有装 httpx,启动时会直接 ImportError。
解决办法:
pip install httpx还有一种情况是配置里的 command 写错了。注意command必须是可执行文件的名字,args必须是绝对路径或者相对于当前工作目录的路径。不要用~这种 shell 扩展符,JSON 配置文件里不会做 shell 展开。
6.2 模型调了工具但返回结果不理想
症状:工具被调用成功,但返回结果在对话里展示得很混乱。
原因多半是返回内容格式不够结构化。模型对纯 JSON 字符串的理解能力其实很强,但如果你返回一长串没有说明性的 JSON,模型展示给用户的方式也会很敷衍。
我的做法是,返回内容里给一段自然语言总结,再附上原始 JSON 数据。比如:
北京当前天气:气温 5.2°C,风速 8.3 km/h,天气编码 2(局部多云)。这样模型可以直接引用你的总结,不用自己发挥。这个格式在 MCP 里就是TextContent,你可以把多个TextContent组合放在一个 list 里返回,模型会依次处理。
6.3 SSE 连接挂起或超时
症状:Host 连接时报ReadTimeoutError或连接一直 pending。
排查思路:如果用了 HTTP/SSE 模式,先确认服务进程还在、端口没被占用、代理配置没把你内网地址拦住。如果用了 Nginx 反代,检查proxy_read_timeout是否设置够大。MCP 的流式请求有时会长连接,默认的 60 秒往往不够,建议设成 300 秒。
location /mcp { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_set_header Host $host; proxy_read_timeout 300s; proxy_buffering off; }注意proxy_buffering off,否则 SSE 流可能会被 Nginx 攒在缓冲区里,客户端迟迟收不到消息。
6.4 客户端提示 JSON Schema 不合法
症状:Host 在初始化阶段就报 schema 格式错误,比如Schema validation failed。
排查思路:用mcp dev启动后打开 Inspector,点击对应工具检查 schema 是否有问题。最常见的坑是 pydantic 的Field里用了le、ge这种约束,但约束类型和字段类型不匹配。比如写成days: int = Field(ge=1, le=7),如果 pydantic 版本较老,它生成的minimum和maximum可能会被当字符串处理。
解决办法是升级 pydantic 到 v2,或者在生成 schema 之后手动打印model_json_schema()确认一遍。格式没问题再接入 Host。
6.5 调试时日志看不到
症状:自己加的 print 语句在终端看不到。
原因:stdio 模式下,Server 的 stdout 被用于协议通信了。你 print 的数据全都当作协议报文发给了 Host,不会出现在终端。
解决办法:用logging模块输出到 stderr,或者专门写日志文件。MCP SDK 底层也在用 logging 输出调试信息,设置如下:
import logging logging.basicConfig(level=logging.DEBUG)在本地调试时,某些实现里 stderr 也不一定显示,通常跑 service 的终端窗口会实时显示日志。A:你确认一下进程是不是还在跑、数据的生成日志是否输出到文件,把日志级别调低,排查起来能省一半时间。这个细节真不忍心看到有人再踩一遍。
7. 进阶技巧与经验总结
开发完这个天气服务,我最大的感受是:MCP Server 没有想象中复杂,它就是一个“适配层”,把现有能力翻译成 AI 能理解的语言。但恰恰是这层“翻译”,藏着不少值得打磨的细节。
工具粒度要克制。很多人在开发 MCP Server 的时候总想一口气把所有功能都暴露给模型,结果工具列表一大堆,模型选择困难,调用准确率直线下降。我的建议是,第一次接入控制在 3 到 5 个工具,跑通全链路后再逐步增加。
参数设计要显式化。给模型描述参数时,要把取值范围、默认值、单位、边界条件都说清楚。不要写“可选的温度单位”这种含糊描述,直接写“celsius 或 fahrenheit,不传默认 celsius”。
错误信息要友好。工具内部异常时,尽量返回能指导重新调用的信息。如果你返回“请求失败”四个字,模型只能再次发起相同请求,大概率还是失败。我在天气服务里会把具体城市名、错误原因带出来,模型就能判断是不是参数不对。
还有一个容易被忽略的点:MCP 服务是否能同时处理多个客户端请求跟你们内部业务架构直接相关。例如两个客户端同时在调get_weather,内部的 httpx 连接池和服务生命周期管理必须注意线程安全和并发控制。官方 SDK 用 asyncio 事件循环做并发处理,尽量保证工具函数内部是异步的,不要阻塞事件循环,否则高并发时会看到明显的卡顿。
如果后面还想延伸,可以考虑加 auth 中间件做服务鉴权、用 FastMCP 这个上层框架简化代码(它在官方 SDK 之上提供更简洁的装饰器风格)、做一个 MCP 工具仓库给团队内部共用。我接下来的计划是把公司内部的几个常用 API 封装成一个中心化 MCP 服务,接上统一的权限审计,这样多个 AI Agent 都能安全调用。
这篇文章所有代码我都验证过,你照着走一遍应该半天内能跑通。如果你在这个基础上做了更有意思的工具,欢迎来交流。