08-MCP 下篇：开发自己的 MCP Server —— 当现成的工具不够用-洪萨配资

MCP 下篇：开发自己的 MCP Server —— 当现成的工具不够用

上篇你学会了装别人的 MCP Server。但每个公司的 API、每个团队的内部工具、每个项目的数据源都不一样。总有一个时刻，现成的 MCP Server 都满足不了你。这篇就是为那个时刻准备的。

什么时候需要自定义 MCP Server

三条判断标准：

1. 内部 API / 私有系统 → 公司内部的 CI/CD、项目管理、数据平台 → 没有现成的 MCP Server，必须自己写 2. 定制数据源 → 内部的文档系统、设计规范库、代码规范 → 接入后 AI 能直接查内部标准 3. 重复的固定操作 → 每次发布都要走"查 Jira → 读设计稿 → 生成 SQL 迁移 → 创建 PR" → 与其每次口头描述，不如写成一个 MCP Server 封装

一个判断框架：

你手动帮 AI 做的操作中，有没有： ├─ 每次都要你手动查某个系统 → 写成 MCP Tool 让 AI 自己查 ├─ 每次都要你复制粘贴某些数据 → 写成 MCP Resource 让 AI 自己读 └─ 每次都要你重复某些上下文 → 写成 MCP Prompt 让 AI 自己加载

从零开发：Python 版 MCP Server

我们实际写一个 MCP Server，让它连接一个"内部任务管理系统"（模拟）。

项目结构

my-mcp-server/ ├── server.py # MCP Server 主文件 ├── mcp.json # Claude Code 配置（测试用） └── requirements.txt # 只依赖 mcp 一个包

Step 1：安装依赖

pipinstallmcp

Step 2：写 Server

# server.py —— 一个连接内部任务系统的 MCP Server"""MCP Server：内部任务管理。提供任务查询、创建、统计功能。"""importasynciofrommcp.serverimportServerfrommcp.server.stdioimportstdio_serverfrommcp.typesimportTool,TextContent# 模拟的内部任务数据库TASKS_DB=[{"id":"T-001","title":"用户登录重构","status":"in_progress","assignee":"张三","priority":"high"},{"id":"T-002","title":"支付接口性能优化","status":"todo","assignee":"李四","priority":"critical"},{"id":"T-003","title":"首页改版","status":"done","assignee":"王五","priority":"medium"},{"id":"T-004","title":"日志系统升级","status":"in_progress","assignee":"张三","priority":"low"},]# 创建 Server 实例server=Server("task-manager")@server.list_tools()asyncdeflist_tools()->list[Tool]:"""告诉 Claude Code 这个 Server 提供哪些 Tool。 这是 Tool Search 的关键——name 和 description 决定了 AI 在什么情况下会调用这个 Tool。 """return[Tool(name="search_tasks",description="按状态、负责人或优先级搜索内部任务。""适用于：了解谁在做什么、查某个模块是否有进行中的任务。""关键词：任务、task、todo、in_progress、指派、负责人",inputSchema={"type":"object","properties":{"status":{"type":"string","enum":["todo","in_progress","done",None],"description":"任务状态过滤。可选值：todo/in_progress/done。为 null 则不过滤。","default":None},"assignee":{"type":"string","description":"按负责人过滤。支持模糊匹配。可选，默认不过滤。","default":None},"priority":{"type":"string","enum":["low","medium","high","critical",None],"description":"按优先级过滤。可选。","default":None}},"required":[]# 所有参数都可选}),Tool(name="create_task",description="在内部任务系统中创建新任务。""适用于：需要记录待办事项、分配工作。",inputSchema={"type":"object","properties":{"title":{"type":"string","description":"任务标题。简洁明确，不超过 100 字。"},"assignee":{"type":"string","description":"任务负责人。可以是人名或团队名。","default":"待分配"},"priority":{"type":"string","enum":["low","medium","high","critical"],"description":"任务优先级。","default":"medium"}},"required":["title"]# title 必填}),Tool(name="task_stats",description="获取内部任务系统的统计数据：各状态任务数、各负责人任务数、各优先级任务数。""适用于：了解团队负载、生成进度报告。",inputSchema={"type":"object","properties":{},"required":[]}),]@server.call_tool()asyncdefcall_tool(name:str,arguments:dict)->list[TextContent]:"""执行 Tool 调用。name 是 Tool 名称，arguments 是 AI 传入的参数。"""ifname=="search_tasks":# 按条件过滤任务results=TASKS_DB[:]ifarguments.get("status"):results=[tfortinresultsift["status"]==arguments["status"]]ifarguments.get("assignee"):keyword=arguments["assignee"]results=[tfortinresultsifkeywordint["assignee"]]ifarguments.get("priority"):results=[tfortinresultsift["priority"]==arguments["priority"]]ifnotresults:return[TextContent(type="text",text="没有找到匹配的任务。")]lines=["| ID | 标题 | 状态 | 负责人 | 优先级 |","|----|------|------|--------|--------|"]fortinresults:status_map={"todo":"📋 待办","in_progress":"🔄 进行中","done":"✅ 完成"}lines.append(f"|{t['id']}|{t['title']}|{status_map[t['status']]}|{t['assignee']}|{t['priority']}|")return[TextContent(type="text",text="\n".join(lines))]elifname=="create_task":task_id=f"T-{len(TASKS_DB)+1:03d}"new_task={"id":task_id,"title":arguments["title"],"status":"todo","assignee":arguments.get("assignee","待分配"),"priority":arguments.get("priority","medium")}TASKS_DB.append(new_task)return[TextContent(type="text",text=f"✅ 任务创建成功。\n"f"- ID:{task_id}\n"f"- 标题:{arguments['title']}\n"f"- 负责人:{new_task['assignee']}\n"f"- 优先级:{new_task['priority']}")]elifname=="task_stats":# 统计各维度的任务分布status_counts={}assignee_counts={}fortinTASKS_DB:status_counts[t["status"]]=status_counts.get(t["status"],0)+1assignee_counts[t["assignee"]]=assignee_counts.get(t["assignee"],0)+1lines=["## 任务统计","","### 按状态",f"- 📋 待办:{status_counts.get('todo',0)}",f"- 🔄 进行中:{status_counts.get('in_progress',0)}",f"- ✅ 完成:{status_counts.get('done',0)}","","### 按负责人",]forname,countinassignee_counts.items():lines.append(f"-{name}:{count}个任务")return[TextContent(type="text",text="\n".join(lines))]asyncdefmain():"""启动 MCP Server（stdio 模式）。"""asyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream,server.create_initialization_options())if__name__=="__main__":asyncio.run(main())

Step 3：配置 Claude Code 使用这个 Server

// 项目的 .mcp.json{"mcpServers":{"task-manager":{"type":"stdio","command":"python","args":["server.py"],"cwd":"my-mcp-server"// 工作目录}}}

Step 4：在 Claude Code 中测试

你: "张三现在手上有哪些进行中的任务？" AI: [调用 task-manager: search_tasks(assignee="张三", status="in_progress")] 返回： | ID | 标题 | 状态 | 负责人 | 优先级 | |----|------|------|--------|--------| | T-001 | 用户登录重构 | 🔄 进行中 | 张三 | high | | T-004 | 日志系统升级 | 🔄 进行中 | 张三 | low | → "张三目前有两个进行中的任务：用户登录重构（高优先级）和日志系统升级（低优先级）。" 你: "帮我创建一个任务：修复登录页的 XSS 漏洞，高优先级，给张三" AI: [调用 task-manager: create_task(title="修复登录页的 XSS 漏洞", assignee="张三", priority="high")] → "✅ 任务 T-005 创建成功，已分配给张三。" 你: "团队现在负载怎么样？" AI: [调用 task-manager: task_stats()] 返回： ## 任务统计 ### 按状态 - 📋 待办: 2 - 🔄 进行中: 2 - ✅ 完成: 1 ### 按负责人 - 张三: 3 个任务 - 李四: 1 个任务 - 王五: 1 个任务 → "张三负载最重，有 3 个任务。"

一个 120 行的 Python 文件，你的 AI 就多了一个"任务管理助手"能力。

Node.js 版本速览

如果你更习惯 Node.js，用@modelcontextprotocol/sdk：

// server.jsimport{McpServer}from"@modelcontextprotocol/sdk/server/mcp.js";import{StdioServerTransport}from"@modelcontextprotocol/sdk/server/stdio.js";import{z}from"zod";constserver=newMcpServer({name:"task-manager",version:"1.0.0"});// 用 Zod schema 定义 Tool 参数，比手写 JSON Schema 更简洁server.registerTool("search_tasks",{description:"按状态、负责人或优先级搜索内部任务",inputSchema:{status:z.enum(["todo","in_progress","done"]).optional(),assignee:z.string().optional(),priority:z.enum(["low","medium","high","critical"]).optional(),}},async({status,assignee,priority})=>{// 实现逻辑同上return{content:[{type:"text",text:"..."}]};});consttransport=newStdioServerTransport();awaitserver.connect(transport);

Tool、Resource、Prompt 怎么选

MCP 协议提供了三种暴露给 AI 的方式，很多人分不清：

Tool（工具） → AI 主动调用，执行一个操作并等待结果 → 例：search_tasks、create_task、deploy、run_tests → 关键：AI 决定何时调用，返回实时计算结果 Resource（资源） → AI 按需读取的数据，像文件一样 → 例：设计规范文件、API 文档、项目架构图 → 关键：更像"读文件"，数据相对静态，可被缓存 Prompt（提示模板） → 预定义的场景化提示词 → 例："code-review" 场景的完整提示词 → 关键：结合了上下文和指令，但本质上是文本

选择决策树

你暴露的东西是... │ ├─ 一个能执行的操作（查询、创建、修改、删除） │ → 用 Tool │ 例：查任务、创建 Issue、部署服务 │ ├─ 一份 AI 需要参考的数据（文档、规范、配置） │ → 用 Resource │ 例：设计规范、API 文档、代码风格指南 │ └─ 一个场景化的提示词模板 → 用 Prompt 但坦率说：这个功能现在几乎被 Skills 替代了。 如果你在用 Claude Code，用 Skills 比用 MCP Prompt 更方便。

实践中 Tool 是最常用的。80% 的自定义 MCP Server 只需要 Tool。

调试流程：从写完到能用

第一步：用 MCP Inspector 测试

MCP Inspector 是一个官方出的调试工具，让你在不启动 Claude Code 的情况下测试 Tool：

# 安装npx @modelcontextprotocol/inspector# 在 Inspector 中连接你的 Server# 手动传入参数，看 Tool 返回什么

这能快速暴露 80% 的问题——参数解析错误、返回格式不对、异常没捕获。

第二步：在 Claude Code 中集成

# 启动 Claude Code，查看 MCP 状态claude mcp list|greptask-manager# 如果没显示，检查日志claude mcp logs task-manager

常见启动错误：

ModuleNotFoundError: Python 依赖没装
Connection refused: Server 进程启动失败
Permission denied: 文件执行权限问题

第三步：验证 AI 是否会调用

用不同方式描述同一个需求，看 Tool 是否被触发：

测试 1: "张三有哪些任务？" → 应该触发 search_tasks 测试 2: "帮我查下 team 的工作进展" → 应该触发 task_stats 测试 3: "创建一个新 task" → 应该触发 create_task

如果 3 次都没触发 → Tool 的description写得太模糊，回去改。

四个常见坑

坑一：Tool description 太模糊

❌ "处理任务" → AI 看不懂这个 Tool 是干什么的，永远不会调用 ✅ "按状态（todo/in_progress/done）、负责人或优先级搜索内部任务系统。 返回匹配的任务列表，包含 ID、标题、状态、负责人、优先级。 适用于：查询某人手上有多少任务、查某个模块是否有正在进行的任务。" → AI 一看就知道什么时候该用

核心原则：Tool 描述是 AI 判断是否调用的唯一依据。描述要包含三个要素：功能（做什么）、条件（什么时候用）、产出（返回什么）。

坑二：inputSchema 缺少默认值

# ❌ 有 bug"properties":{"status":{"type":"string","enum":["todo","in_progress","done"]# 没写 default}}# 如果 AI 不传 status 参数 → Server 收到 None 或 undefined# → 你的代码没有处理 None 的情况 → 崩溃# ✅ 正确"properties":{"status":{"type":"string","enum":["todo","in_progress","done",None],"description":"任务状态过滤。为 null 则不过滤。","default":None# 明确默认值}}

坑三：异步调用超时

# ❌ 同步阻塞调用defcall_tool(name,arguments):result=requests.get("https://slow-api.company.com/endpoint")returnresult# 这个请求可能耗时 30+ 秒# MCP 默认超时 60 秒。但如果 API 真的慢，可能会超时。# ✅ 用异步 + 显式超时控制asyncdefcall_tool(name,arguments):try:asyncwithhttpx.AsyncClient(timeout=30)asclient:result=awaitclient.get("https://slow-api.company.com/endpoint")returnresultexcepthttpx.TimeoutException:return{"error":"API 超时，请稍后重试"}

坑四：stdio vs SSE 的选型

stdio（标准输入输出） → MCP Server 作为本地子进程运行 → 适合：本地工具、简单 API 封装 → Claude Code 会自动管理进程生命周期 → 配置："type": "stdio" SSE（Server-Sent Events） → MCP Server 作为远程 HTTP 服务运行 → 适合：公司内部的共享 MCP 服务、需要独立部署的场景 → 需要你自己管理进程 → 配置："type": "sse", "url": "https://mcp.company.com/sse"

90% 的场景用 stdio 就够了。SSE 只在"需要多个 AI 实例共享同一个 MCP Server"时才有优势。

MCP 上篇 + 下篇总结

上篇（第 7 篇）：用别人的 MCP → 5 个必装 MCP Server 实测 + 配置策略 → 三级配置层级 + 安全清单 → Tool Search：将 300+ Tool 的开销降低 85% 下篇（本篇）：写自己的 MCP → 从零开发 Python MCP Server（完整代码） → Tool / Resource / Prompt 选择框架 → MCP Inspector 调试流程 → 四个常见坑（描述模糊、缺少默认值、异步超时、stdio vs SSE）

读完这两篇，你应该：