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

发布时间:2026/6/9 11:40:33

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安装依赖pipinstallmcpStep 2写 Server# server.py —— 一个连接内部任务系统的 MCP ServerMCP 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 实例serverServer(task-manager)server.list_tools()asyncdeflist_tools()-list[Tool]:告诉 Claude Code 这个 Server 提供哪些 Tool。 这是 Tool Search 的关键——name 和 description 决定了 AI 在什么情况下会调用这个 Tool。 return[Tool(namesearch_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(namecreate_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(nametask_stats,description获取内部任务系统的统计数据各状态任务数、各负责人任务数、各优先级任务数。适用于了解团队负载、生成进度报告。,inputSchema{type:object,properties:{},required:[]}),]server.call_tool()asyncdefcall_tool(name:str,arguments:dict)-list[TextContent]:执行 Tool 调用。name 是 Tool 名称arguments 是 AI 传入的参数。ifnamesearch_tasks:# 按条件过滤任务resultsTASKS_DB[:]ifarguments.get(status):results[tfortinresultsift[status]arguments[status]]ifarguments.get(assignee):keywordarguments[assignee]results[tfortinresultsifkeywordint[assignee]]ifarguments.get(priority):results[tfortinresultsift[priority]arguments[priority]]ifnotresults:return[TextContent(typetext,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(typetext,text\n.join(lines))]elifnamecreate_task:task_idfT-{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(typetext,textf✅ 任务创建成功。\nf- ID:{task_id}\nf- 标题:{arguments[title]}\nf- 负责人:{new_task[assignee]}\nf- 优先级:{new_task[priority]})]elifnametask_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(typetext,text\n.join(lines))]asyncdefmain():启动 MCP Serverstdio 模式。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张三, statusin_progress)] 返回 | ID | 标题 | 状态 | 负责人 | 优先级 | |----|------|------|--------|--------| | T-001 | 用户登录重构 | 进行中 | 张三 | high | | T-004 | 日志系统升级 | 进行中 | 张三 | low | → 张三目前有两个进行中的任务用户登录重构高优先级和日志系统升级低优先级。 你: 帮我创建一个任务修复登录页的 XSS 漏洞高优先级给张三 AI: [调用 task-manager: create_task(title修复登录页的 XSS 漏洞, assignee张三, priorityhigh)] → ✅ 任务 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}frommodelcontextprotocol/sdk/server/mcp.js;import{StdioServerTransport}frommodelcontextprotocol/sdk/server/stdio.js;import{z}fromzod;constservernewMcpServer({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:...}]};});consttransportnewStdioServerTransport();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 缺少默认值# ❌ 有 bugproperties:{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):resultrequests.get(https://slow-api.company.com/endpoint)returnresult# 这个请求可能耗时 30 秒# MCP 默认超时 60 秒。但如果 API 真的慢可能会超时。# ✅ 用异步 显式超时控制asyncdefcall_tool(name,arguments):try:asyncwithhttpx.AsyncClient(timeout30)asclient:resultawaitclient.get(https://slow-api.company.com/endpoint)returnresultexcepthttpx.TimeoutException:return{error:API 超时请稍后重试}坑四stdio vs SSE 的选型stdio标准输入输出 → MCP Server 作为本地子进程运行 → 适合本地工具、简单 API 封装 → Claude Code 会自动管理进程生命周期 → 配置type: stdio SSEServer-Sent Events → MCP Server 作为远程 HTTP 服务运行 → 适合公司内部的共享 MCP 服务、需要独立部署的场景 → 需要你自己管理进程 → 配置type: sse, url: https://mcp.company.com/sse90% 的场景用 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读完这两篇你应该能把现成的 MCP Server 配置到最优状态能判断什么时候该自己写 MCP Server能从零写一个可用的 MCP Server 并接入 Claude Code延伸阅读MCP 协议规范MCP Python SDKMCP Inspector

相关新闻