主题
07.4 手写一个 MCP Server
从零写一个能用的 MCP server,把你的内部系统变成 Claude 的工具。
读完你能做什么:用 Python 或 TypeScript 写出一个真实可用的 server,并接入 Claude Code。
说明:MCP SDK 会演进,以下代码演示核心概念与结构。具体 API 以官方 SDK(
https://modelcontextprotocol.io)的当前版本为准。
一、什么时候值得自己写
text
✅ 你有内部 API / 数据库 / 系统,没有现成的 MCP server
✅ 你想把一组固定操作封装成结构化工具(而不是让 Claude 用 curl 猜)
✅ 你需要在工具层做权限、审计、参数校验
❌ 已有官方 server(GitHub、Sentry 等)→ 直接用
❌ 只是想读文件 → 用内置工具
❌ 只是固定流程 → 用 Skill二、一个 server 的最小结构
无论什么语言,一个 MCP server 都要做三件事:
text
1. 声明它提供哪些工具(名字、描述、参数 schema)
2. 实现每个工具的逻辑
3. 通过某种传输(stdio/HTTP)与客户端通信三、Python 版:一个天气查询 server
3.1 安装 SDK
bash
pip install mcp --break-system-packages # 以官方包名为准3.2 完整代码
python
# weather_server.py
from mcp.server.fastmcp import FastMCP
import httpx
# 创建 server
mcp = FastMCP("weather")
@mcp.tool()
async def get_weather(city: str) -> str:
"""查询指定城市的当前天气。
Args:
city: 城市名称,如 "Beijing"、"Shanghai"
"""
# 这里用一个假设的天气 API(需替换为真实 API)
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.example-weather.com/current", # 需替换
params={"q": city},
timeout=10.0,
)
resp.raise_for_status()
data = resp.json()
return (
f"{city} 当前天气:{data['condition']},"
f"温度 {data['temp']}°C,湿度 {data['humidity']}%"
)
@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
"""查询未来几天的天气预报。
Args:
city: 城市名称
days: 预报天数,1-7,默认 3
"""
if not 1 <= days <= 7:
return "错误:days 必须在 1-7 之间"
# ... 实现预报逻辑
return f"{city} 未来 {days} 天预报:..."
if __name__ == "__main__":
# 通过 stdio 传输运行
mcp.run(transport="stdio")3.3 关键点
| 元素 | 作用 |
|---|---|
@mcp.tool() | 把函数声明为 MCP 工具 |
| 函数的 docstring | 成为工具的 description,Claude 据此决定何时调用 —— 写好它至关重要 |
| 类型注解 | 自动生成参数 schema |
| 参数校验 | 在函数里做,返回清晰的错误信息 |
3.4 接入 Claude Code
bash
claude mcp add --transport stdio weather -- python /path/to/weather_server.py(路径需替换为你的实际路径)
测试:
text
> /mcp # 确认 weather server 已连接
> 北京现在天气怎么样? # Claude 应该调用 get_weather四、TypeScript 版:一个内部工单 server
4.1 安装
bash
npm install @modelcontextprotocol/sdk # 以官方包名为准4.2 核心代码
typescript
// ticket-server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const server = new McpServer({
name: "internal-tickets",
version: "1.0.0",
});
// 工具 1:查询工单
server.tool(
"list_tickets",
"列出符合条件的工单。用于查看待处理任务、按状态筛选工单。",
{
status: z.enum(["open", "in_progress", "closed"]).optional()
.describe("按状态筛选,省略则返回全部"),
assignee: z.string().optional().describe("按负责人筛选"),
},
async ({ status, assignee }) => {
const tickets = await queryTickets({ status, assignee }); // 你的内部实现
return {
content: [{
type: "text",
text: JSON.stringify(tickets, null, 2),
}],
};
}
);
// 工具 2:创建工单(写操作,注意权限)
server.tool(
"create_ticket",
"创建一个新工单。仅在用户明确要求创建时使用。",
{
title: z.string().describe("工单标题"),
body: z.string().describe("详细描述"),
priority: z.enum(["low", "medium", "high"]).default("medium"),
},
async ({ title, body, priority }) => {
const ticket = await createTicket({ title, body, priority });
return {
content: [{ type: "text", text: `已创建工单 #${ticket.id}` }],
};
}
);
// 通过 stdio 运行
const transport = new StdioServerTransport();
await server.connect(transport);4.3 接入
bash
claude mcp add --transport stdio tickets -- node /path/to/ticket-server.js五、写好工具描述:决定成败的一步
Claude 通过工具的 description 决定何时调用它。 描述写不好,工具就形同虚设。
5.1 对比
python
# ❌ 太简略,Claude 不知道什么时候用
@mcp.tool()
async def query(sql: str) -> str:
"""执行查询"""
# ✅ 清晰说明用途、适用场景、参数含义
@mcp.tool()
async def query_analytics_db(sql: str) -> str:
"""在分析数据库上执行只读 SQL 查询。
用于回答关于用户行为、销售数据、运营指标的问题,
比如"上月新增用户数""某产品的转化率"。
只支持 SELECT 查询。表结构见 get_schema 工具。
Args:
sql: 一条 SELECT 语句。禁止 INSERT/UPDATE/DELETE。
"""5.2 描述的三要素
text
1. 这个工具做什么(一句话)
2. 什么场景该用它(给 Claude 触发信号)
3. 每个参数的含义和约束这和写 Skill 的 description 是同一个道理,见 08.3 编写高触发率的 description。
六、在 server 侧做安全
不要信任 Claude 传来的参数。 在 server 里做校验和限制:
python
@mcp.tool()
async def query_db(sql: str) -> str:
"""只读查询分析库。"""
# 强制只读
normalized = sql.strip().lower()
if not normalized.startswith("select"):
return "错误:只允许 SELECT 查询"
forbidden = ["insert", "update", "delete", "drop", "alter", "truncate"]
if any(kw in normalized for kw in forbidden):
return "错误:检测到写操作关键词,已拒绝"
# 限制返回行数,避免撑爆上下文
if "limit" not in normalized:
sql += " LIMIT 1000"
result = await run_query(sql)
return format_result(result)server 侧的校验是最后一道确定性防线 —— 它不依赖 Claude 的"自觉",无论 Claude 被怎样诱导,写操作在这里就被挡住了。
七、输出限制
工具返回的内容会进 Claude 的上下文。返回过大会撑爆窗口:
python
@mcp.tool()
async def read_logs(lines: int = 100) -> str:
"""读取应用日志的最近若干行。"""
lines = min(lines, 500) # 硬上限,防止一次拉几万行
logs = get_recent_logs(lines)
return logsMCP 有工具输出大小的限制和警告机制。对可能返回大量数据的工具,主动分页或截断。
八、本地开发与调试
8.1 用 stdio 本地测试
bash
# 直接跑 server,用 MCP inspector 或手动测试
python weather_server.py8.2 在 Claude Code 里调试
bash
claude --debug "mcp" "北京天气" # 看 MCP 相关的调试日志text
> /mcp # 查看 server 状态、工具列表、错误8.3 常见问题
| 症状 | 原因 | 对策 |
|---|---|---|
| server 连不上 | 命令/路径错误 | claude mcp get <name> 检查配置 |
| 工具不被调用 | description 不清晰 | 改进描述 |
| 调用报错 | 参数 schema 不匹配 | 检查类型注解 |
| 返回乱码 | 编码问题 | 确保返回 UTF-8 文本 |
| 上下文爆炸 | 返回太大 | 加输出限制 |
九、从 stdio 到 HTTP:让团队共享
本地 stdio server 只有你能用。要让团队共享,部署成 HTTP server:
text
stdio(本地进程)
→ 适合开发、个人工具
HTTP(部署到服务器)
→ 适合团队共享,配合 project 作用域的 .mcp.jsonHTTP server 需要处理认证(见 07.3)和部署。核心工具逻辑不变,只是换传输层。
十、一个完整的开发流程
text
1. 明确要暴露哪些能力 → 列出工具清单
2. 为每个工具写清 description(决定触发)
3. 实现工具逻辑,在 server 侧做校验(安全)
4. 加输出限制(成本)
5. stdio 本地测试,用 /mcp 确认连接
6. 让 Claude 实际调用,观察它是否在正确场景触发
7. 迭代 description 直到触发准确
8. 需要共享则部署成 HTTP + 认证
9. 用 project 作用域写进 .mcp.json 分发给团队