Skip to content

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 logs

MCP 有工具输出大小的限制和警告机制。对可能返回大量数据的工具,主动分页或截断。


八、本地开发与调试

8.1 用 stdio 本地测试

bash
# 直接跑 server,用 MCP inspector 或手动测试
python weather_server.py

8.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.json

HTTP server 需要处理认证(见 07.3)和部署。核心工具逻辑不变,只是换传输层。


十、一个完整的开发流程

text
1. 明确要暴露哪些能力 → 列出工具清单
2. 为每个工具写清 description(决定触发)
3. 实现工具逻辑,在 server 侧做校验(安全)
4. 加输出限制(成本)
5. stdio 本地测试,用 /mcp 确认连接
6. 让 Claude 实际调用,观察它是否在正确场景触发
7. 迭代 description 直到触发准确
8. 需要共享则部署成 HTTP + 认证
9. 用 project 作用域写进 .mcp.json 分发给团队

延伸阅读

基于 VitePress 构建 · 内容采用原作者授权