【Claude】MCP 协议深度解析与自定义服务器开发 — 已解决

张

张建站

2026/7/2 11:31:14

10分钟阅读

【Claude】MCP 协议深度解析与自定义服务器开发 — 已解决适用版本Claude Code v1.0.x 及以上、MCP Protocol v2024-11-05受影响场景自定义工具集成、外部数据源接入、企业内部 API 桥接阅读时长约 30 分钟目录问题现象原理深挖MCP 协议架构根因分析MCP 开发中的常见问题多方案解决从开发到部署验证回归MCP 服务器验证避坑最佳实践附录MCP 协议速查表1. 问题现象1.1 典型问题表现问题一MCP 服务器连接失败使用数据库工具查询数据 Error: MCP server db-tools not connected # 配置了 MCP 服务器但无法连接问题二自定义 MCP 工具不出现// .claude/mcp-servers.json { my-tools: { command: node, args: [./mcp-server.js] } } // 但 Claude Code 中看不到自定义工具问题三MCP 工具调用超时查询数据库 [Claude 调用 MCP 工具: query_db] [等待...] Error: Tool execution timeout (30s) # MCP 工具执行时间过长问题四MCP 服务器开发不知道从何入手开发者想要: - 自定义企业内部 API 的 MCP 工具 - 集成 Jira/Confluence 的 MCP 服务器 - 自定义数据库查询工具但不知道 MCP 协议格式和开发方式问题五MCP 服务器在 CI 中不稳定# CI 环境中 MCP 服务器频繁断开 claude -p 使用 MCP 工具 # Error: MCP connection lost2. 原理深挖MCP 协议架构2.1 MCP 协议概述Model Context Protocol (MCP) 是 Anthropic 提出的开放协议允许外部工具服务器与 Claude及兼容客户端通信提供自定义工具能力。┌─────────────────────────────────────────────────┐ │ MCP 架构 │ ├─────────────────────────────────────────────────┤ │ │ │ Claude Code (MCP Client) │ │ │ │ │ │ JSON-RPC 2.0 over stdio/SSE │ │ │ │ │ ├──→ MCP Server A (文件系统工具) │ │ ├──→ MCP Server B (数据库工具) │ │ ├──→ MCP Server C (企业 API 工具) │ │ └──→ MCP Server D (自定义工具) │ │ │ │ 每个服务器: │ │ - 独立进程 (stdio) 或 HTTP 服务 (SSE) │ │ - 通过 JSON-RPC 通信 │ │ - 提供工具 (tools) 和资源 (resources) │ │ │ └─────────────────────────────────────────────────┘2.2 通信协议传输方式 1: stdio (推荐) Claude Code ←stdin/stdout→ MCP Server (子进程) - 零网络开销 - 生命周期由 Claude Code 管理 - 适合本地工具传输方式 2: SSE (HTTP) Claude Code ←HTTP/SSE→ MCP Server (HTTP 服务) - 支持远程服务器 - 需要网络配置 - 适合共享/远程工具消息格式: JSON-RPC 2.0 请求: {jsonrpc: 2.0, id: 1, method: tools/list, params: {}} 响应: {jsonrpc: 2.0, id: 1, result: {tools: [...]}} 通知: {jsonrpc: 2.0, method: notification, params: {...}}2.3 MCP 生命周期1. 初始化 Client → Server: initialize({protocolVersion, capabilities}) Server → Client: {protocolVersion, capabilities, serverInfo} 2. 工具发现 Client → Server: tools/list() Server → Client: {tools: [{name, description, inputSchema}]} 3. 工具调用 Client → Server: tools/call({name, arguments}) Server → Client: {content: [{type, text}], isError} 4. 资源访问 (可选) Client → Server: resources/list() Server → Client: {resources: [{uri, name, description}]} 5. 关闭 Client 关闭 stdin → Server 检测到 EOF → 退出2.4 工具定义格式{ name: query_database, description: 执行 SQL 查询并返回结果, inputSchema: { type: object, properties: { sql: { type: string, description: SQL 查询语句 }, database: { type: string, description: 数据库名称, enum: [prod, staging, dev] }, limit: { type: integer, description: 返回行数限制, default: 100 } }, required: [sql, database] } }2.5 Claude Code MCP 配置// .claude/settings.json 或 ~/.claude/settings.json { mcpServers: { filesystem: { command: npx, args: [-y, anthropic-ai/mcp-filesystem, /path/to/allowed/dir] }, database: { command: node, args: [./mcp-servers/db-server.js], env: { DB_HOST: localhost, DB_PORT: 5432 } }, remote-api: { url: https://internal-company.com/mcp-server, headers: { Authorization: Bearer ${MCP_AUTH_TOKEN} } } } }3. 根因分析MCP 开发中的常见问题3.1 根因一协议理解不足开发者不了解 MCP 的 JSON-RPC 消息格式直接用 HTTP REST 风格开发导致通信失败。3.2 根因二stdio 通信错误stdio 模式下MCP 服务器意外向 stdout 输出调试信息干扰 JSON-RPC 消息。3.3 根因三工具 Schema 错误inputSchema 不符合 JSON Schema 规范Claude 无法正确调用工具。3.4 根因四超时处理缺失长时间运行的工具没有超时处理Claude Code 等待 30 秒后强制断开。3.5 根因五环境变量不传递MCP 服务器需要的 API Key、数据库密码等环境变量未在配置中传递。3.6 根因六错误处理不完善工具执行失败时返回格式不正确Claude 无法理解错误原因。4. 多方案解决从开发到部署4.1 方案一Python MCP 服务器开发#!/usr/bin/env python3 自定义 MCP 服务器 — 企业数据库查询工具使用 Anthropic 官方 MCP SDK pip install mcp import json import sys import os import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( Tool, TextContent, ImageContent, LoggingLevel ) import logging # 配置日志 (输出到 stderr不干扰 stdout JSON-RPC) logging.basicConfig( streamsys.stderr, levellogging.INFO, format[MCP] %(asctime)s %(levelname)s %(message)s ) logger logging.getLogger(db-mcp-server) # 创建 MCP 服务器 server Server(db-tools) # 数据库连接 (示例) DB_CONFIG { host: os.environ.get(DB_HOST, localhost), port: int(os.environ.get(DB_PORT, 5432)), user: os.environ.get(DB_USER, admin), password: os.environ.get(DB_PASSWORD, ), } # 工具定义 server.list_tools() async def list_tools() - list[Tool]: 返回可用工具列表 return [ Tool( namequery_database, description执行 SQL 查询并返回结果。仅支持 SELECT 语句。, inputSchema{ type: object, properties: { sql: { type: string, description: SQL SELECT 查询语句 }, database: { type: string, description: 数据库名称, enum: [main, analytics, logs], default: main }, limit: { type: integer, description: 返回行数限制, default: 100, minimum: 1, maximum: 1000 } }, required: [sql] } ), Tool( namelist_tables, description列出指定数据库中的所有表, inputSchema{ type: object, properties: { database: { type: string, enum: [main, analytics, logs], default: main } } } ), Tool( namedescribe_table, description显示表结构列名、类型、注释, inputSchema{ type: object, properties: { table: { type: string, description: 表名 }, database: { type: string, enum: [main, analytics, logs], default: main } }, required: [table] } ) ] # 工具调用处理 server.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: 处理工具调用 logger.info(fTool called: {name} with {arguments}) try: if name query_database: return await handle_query(arguments) elif name list_tables: return await handle_list_tables(arguments) elif name describe_table: return await handle_describe_table(arguments) else: return [TextContent( typetext, textf错误: 未知工具 {name} )] except Exception as e: logger.error(fTool error: {e}) return [TextContent( typetext, textf工具执行错误: {str(e)} )] async def handle_query(args): 执行数据库查询 sql args.get(sql, ) database args.get(database, main) limit args.get(limit, 100) # 安全检查: 只允许 SELECT if not sql.strip().upper().startswith(SELECT): return [TextContent( typetext, text错误: 仅支持 SELECT 查询 )] # 模拟查询 (实际使用 psycopg2/pymysql 等) try: # 模拟数据 results [ {id: 1, name: Alice, email: aliceexample.com}, {id: 2, name: Bob, email: bobexample.com}, ] # 格式化为表格 if results: headers list(results[0].keys()) table | | .join(headers) |\n table | ---| * len(headers) \n for row in results[:limit]: table | | .join(str(row.get(h, )) for h in headers) |\n else: table (无结果) return [TextContent(typetext, textf查询结果 ({database}):\n\n{table})] except Exception as e: return [TextContent(typetext, textf查询失败: {e})] async def handle_list_tables(args): 列出表 database args.get(database, main) tables [users, orders, products, inventory] return [TextContent( typetext, textf数据库 {database} 的表:\n \n.join(f - {t} for t in tables) )] async def handle_describe_table(args): 描述表结构 table args.get(table, ) # 模拟表结构 columns [ {name: id, type: integer, comment: 主键}, {name: name, type: varchar(255), comment: 名称}, {name: email, type: varchar(255), comment: 邮箱}, ![配图](https://i-blog.csdnimg.cn/img_convert/7cb4cf5d3c4ded629375540e8b02f102.png) {name: created_at, type: timestamp, comment: 创建时间}, ] desc f表 {table} 结构:\n\n desc | 列名 | 类型 | 说明 |\n|---|---|---|\n for col in columns: desc f| {col[name]} | {col[type]} | {col[comment]} |\n return [TextContent(typetext, textdesc)] # 主函数 async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ __main__: asyncio.run(main())4.2 方案二TypeScript MCP 服务器开发/** * 自定义 MCP 服务器 — Jira 集成工具 * * npm install anthropic-ai/mcp-sdk */ import { Server } from anthropic-ai/mcp-sdk; import { StdioServerTransport } from anthropic-ai/mcp-sdk; import { Tool, TextContent } from anthropic-ai/mcp-sdk; // 创建服务器 const server new Server( { name: jira-tools, version: 1.0.0 }, { capabilities: { tools: {} } } ); // Jira API 配置 const JIRA_BASE_URL process.env.JIRA_BASE_URL || https://company.atlassian.net; const JIRA_TOKEN process.env.JIRA_TOKEN || ; // 工具列表 server.setRequestHandler(tools/list, async () ({ tools: [ { name: search_issues, description: 搜索 Jira 问题JQL 查询, inputSchema: { type: object, properties: { jql: { type: string, description: JQL 查询语句如 project PROJ AND status Open }, maxResults: { type: integer, description: 最大返回数, default: 50 } }, required: [jql] } }, { name: get_issue, description: 获取 Jira 问题详情, inputSchema: { type: object, properties: { issueKey: { type: string, description: 问题键如 PROJ-123 } }, required: [issueKey] } }, { name: create_issue, description: 创建 Jira 问题, inputSchema: { type: object, properties: { project: { type: string, description: 项目键 }, summary: { type: string, description: 标题 }, description: { type: string, description: 描述 }, issueType: { type: string, enum: [Bug, Task, Story, Epic], description: 问题类型 }, priority: { type: string, enum: [Highest, High, Medium, Low, Lowest] } }, required: [project, summary, issueType] } } ] })); // 工具调用处理 server.setRequestHandler(tools/call, async (request) { const { name, arguments: args } request.params; try { switch (name) { case search_issues: return await searchIssues(args.jql, args.maxResults || 50); case get_issue: return await getIssue(args.issueKey); case create_issue: return await createIssue(args); default: return { content: [{ type: text, text: 未知工具: ${name} }], isError: true }; } } catch (error) { return { content: [{ type: text, text: 错误: ${error.message} }], isError: true }; } }); // Jira API 调用 async function searchIssues(jql: string, maxResults: number) { const response await fetch( ${JIRA_BASE_URL}/rest/api/2/search?jql${encodeURIComponent(jql)}maxResults${maxResults}, { headers: { Authorization: Bearer ${JIRA_TOKEN}, Accept: application/json } } ); if (!response.ok) { throw new Error(Jira API: ${response.status} ${response.statusText}); } const data await response.json(); const issues data.issues.map((issue: any) ({ key: issue.key, summary: issue.fields.summary, status: issue.fields.status.name, priority: issue.fields.priority.name, assignee: issue.fields.assignee?.displayName || 未分配 })); const text 找到 ${data.total} 个问题:\n\n issues.map(i ${i.key}: ${i.summary} [${i.status}] (${i.priority})).join(\n); return { content: [{ type: text, text }] }; } async function getIssue(issueKey: string) { const response await fetch( ${JIRA_BASE_URL}/rest/api/2/issue/${issueKey}, { headers: { Authorization: Bearer ${JIRA_TOKEN}, Accept: application/json } } ); if (!response.ok) { throw new Error(Jira API: ${response.status}); } const issue await response.json(); const text 问题: ${issue.key} 标题: ${issue.fields.summary} 状态: ${issue.fields.status.name} 类型: ${issue.fields.issuetype.name} 优先级: ${issue.fields.priority?.name || 无} 报告人: ${issue.fields.reporter?.displayName || 未知} 经办人: ${issue.fields.assignee?.displayName || 未分配} 创建时间: ${issue.fields.created} 描述: ${issue.fields.description || (无描述)}; return { content: [{ type: text, text }] }; } async function createIssue(args: any) { const body { fields: { project: { key: args.project }, summary: args.summary, description: args.description || , issuetype: { name: args.issueType }, priority: args.priority ? { name: args.priority } : undefined } }; const response await fetch( ${JIRA_BASE_URL}/rest/api/2/issue, { method: POST, headers: { Authorization: Bearer ${JIRA_TOKEN}, Content-Type: application/json }, body: JSON.stringify(body) } ); if (!response.ok) { const error await response.text(); throw new Error(创建失败: ${error}); } const data await response.json(); return { content: [{ type: text, text: ✓ 问题已创建: ${data.key}\nURL: ${JIRA_BASE_URL}/browse/${data.key} }] }; } // 启动服务器 const transport new StdioServerTransport(); server.connect(transport); console.error([MCP] Jira tools server started); // stderr, 不干扰 stdout4.3 方案三Claude Code 中配置 MCP 服务器// .claude/settings.json — MCP 服务器配置 { mcpServers: { // 本地 stdio 服务器 db-tools: { command: python3, args: [./mcp-servers/db-server.py], env: { DB_HOST: localhost, DB_PORT: 5432, DB_USER: admin, DB_PASSWORD: ${DB_PASSWORD} // 从环境变量读取 } }, // Node.js 服务器 jira-tools: { command: node, args: [./mcp-servers/jira-server.js], env: { JIRA_BASE_URL: https://company.atlassian.net, JIRA_TOKEN: ${JIRA_TOKEN} } }, // 远程 SSE 服务器 remote-api: { url: https://internal.company.com/mcp, headers: { Authorization: Bearer ${MCP_AUTH_TOKEN} } }, // 使用 npx 运行的官方服务器 filesystem: { command: npx, args: [-y, anthropic-ai/mcp-filesystem, /Users/zhubo/projects] } } }4.4 方案四MCP 服务器调试#!/bin/bash # debug-mcp-server.sh — 调试 MCP 服务器 # 1. 直接测试 MCP 服务器 (模拟 Claude Code 的请求) echo {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:test,version:1.0}}} | \ python3 ./mcp-servers/db-server.py 2/dev/null # 预期响应: {jsonrpc:2.0,id:1,result:{...}} # 2. 列出工具 echo {jsonrpc:2.0,id:2,method:tools/list,params:{}} | \ python3 ./mcp-servers/db-server.py 2/dev/null # 3. 调用工具 echo {jsonrpc:2.0,id:3,method:tools/call,params:{name:list_tables,arguments:{database:main}}} | \ python3 ./mcp-servers/db-server.py 2/dev/null # 4. 查看 stderr 日志 echo {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{}}} | \ python3 ./mcp-servers/db-server.py 21 /dev/null # stderr 输出: [MCP] 2024-01-15 INFO ... # 5. 在 Claude Code 中检查 MCP 连接状态 # 交互模式中输入: # /mcp # 应显示所有已连接的 MCP 服务器和可用工具4.5 方案五错误处理与超时 MCP 服务器的健壮错误处理和超时管理 import asyncio import signal import sys from functools import wraps def with_timeout(seconds30): 工具调用超时装饰器 def decorator(func): wraps(func) async def wrapper(*args, **kwargs): try: return await asyncio.wait_for( func(*args, **kwargs), timeoutseconds ) except asyncio.TimeoutError: return [TextContent( typetext, textf错误: 工具执行超时 ({seconds}秒) )] return wrapper return decorator def with_error_handling(func): 错误处理装饰器 wraps(func) async def wrapper(*args, **kwargs): try: return await func(*args, **kwargs) except ConnectionError as e: return [TextContent(typetext, textf连接错误: {e})] except ValueError as e: return [TextContent(typetext, textf参数错误: {e})] except PermissionError as e: return [TextContent(typetext, textf权限错误: {e})] except Exception as e: logger.error(fUnexpected error: {e}, exc_infoTrue) return [TextContent(typetext, textf内部错误: {e})] return wrapper # 使用装饰器 server.call_tool() with_error_handling async def call_tool(name: str, arguments: dict): if name query_database: return await with_timeout(30)(handle_query)(arguments) # ... # 优雅关闭 def setup_graceful_shutdown(): 设置优雅关闭 def signal_handler(signum, frame): logger.info(收到关闭信号正在清理...) # 清理数据库连接等资源 sys.exit(0) signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler)4.6 方案六MCP 服务器模板#!/usr/bin/env python3 通用 MCP 服务器模板复制此文件并修改工具定义即可快速创建新的 MCP 服务器 import asyncio import sys import logging from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent # 配置 stderr 日志 logging.basicConfig( streamsys.stderr, levellogging.INFO, format[MCP:%(name)s] %(levelname)s %(message)s ) logger logging.getLogger(template) server Server(template-server) # # 工具定义 (修改此处) # TOOLS [ Tool( nameexample_tool, description示例工具描述, inputSchema{ type: object, properties: { param1: { type: string, description: 参数1说明 } }, required: [param1] } ), ] # # 工具处理 (修改此处) # async def handle_example_tool(args: dict) - list[TextContent]: param1 args.get(param1, ) result f处理结果: {param1} return [TextContent(typetext, textresult)] # 工具路由 TOOL_HANDLERS { example_tool: handle_example_tool, } # # MCP 协议实现 (通常不需要修改) # server.list_tools() async def list_tools() - list[Tool]: return TOOLS server.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: logger.info(f调用: {name}({arguments})) handler TOOL_HANDLERS.get(name) if handler: try: return await handler(arguments) except Exception as e: logger.error(f工具错误: {e}, exc_infoTrue) return [TextContent(typetext, textf错误: {e})] else: return [TextContent(typetext, textf未知工具: {name})] async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ __main__: asyncio.run(main())5. 验证回归MCP 服务器验证5.1 MCP 服务器测试脚本#!/bin/bash # test-mcp-server.sh — 测试 MCP 服务器 SERVER_CMD$1 if [ -z $SERVER_CMD ]; then echo 用法: $0 python3 ./mcp-servers/db-server.py exit 1 fi echo MCP 服务器测试 # 测试 1: 初始化 echo -n 测试初始化... INIT_RESPONSE$(echo {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:test,version:1.0}}} | $SERVER_CMD 2/dev/null) if echo $INIT_RESPONSE | python3 -c import sys,json; djson.load(sys.stdin); assert result in d 2/dev/null; then echo ✓ else echo ✗ 初始化失败 echo 响应: $INIT_RESPONSE exit 1 fi # 测试 2: 工具列表 echo -n 测试工具列表... LIST_RESPONSE$(echo {jsonrpc:2.0,id:2,method:tools/list,params:{}} | $SERVER_CMD 2/dev/null) TOOL_COUNT$(echo $LIST_RESPONSE | python3 -c import sys,json djson.load(sys.stdin) print(len(d.get(result,{}).get(tools,[]))) 2/dev/null) if [ $TOOL_COUNT -gt 0 ]; then echo ✓ ($TOOL_COUNT 个工具) else echo ✗ 无工具 exit 1 fi # 测试 3: 工具调用 echo -n 测试工具调用... CALL_RESPONSE$(echo {jsonrpc:2.0,id:3,method:tools/call,params:{name:list_tables,arguments:{database:main}}} | $SERVER_CMD 2/dev/null) if echo $CALL_RESPONSE | python3 -c import sys,json; djson.load(sys.stdin); assert result in d 2/dev/null; then echo ✓ else echo ✗ 工具调用失败 exit 1 fi echo echo 所有测试通过 5.2 验证清单#验证项预期方法1服务器启动无错误直接运行2初始化响应有 resultJSON-RPC 测试3工具列表tools 0tools/list4工具调用有结果tools/call5错误处理返回错误文本无效参数6stdout 纯净仅 JSON-RPC检查无其他输出7stderr 日志有日志21 /dev/null8Claude Code 集成工具可用/mcp 命令6. 避坑最佳实践6.1 MCP 开发原则原则 1: stdout 纯净 — 只输出 JSON-RPC日志到 stderr 原则 2: Schema 规范 — 严格遵循 JSON Schema 原则 3: 超时处理 — 工具执行设超时原则 4: 错误友好 — 返回可读的错误文本原则 5: 环境变量 — 密钥通过 env 传递原则 6: 优雅关闭 — 处理 SIGTERM/SIGINT 原则 7: 异步处理 — 使用 async/await 原则 8: 最小权限 — 工具只暴露必要能力6.2 常见陷阱#陷阱后果解决1stdout 输出日志JSON-RPC 解析失败日志到 stderr2Schema 不规范Claude 调用失败遵循 JSON Schema3无超时Claude 等待 30s 后断开工具设超时4密钥硬编码安全风险用环境变量5同步阻塞响应慢用 async/await6不处理 EOF服务器不退出检测 stdin 关闭7无错误处理Claude 收到异常返回 TextContent8env 不传递连接失败settings.json 配置 env7. 附录MCP 协议速查表7.1 JSON-RPC 方法方法方向说明initializeClient→Server初始化握手tools/listClient→Server列出工具tools/callClient→Server调用工具resources/listClient→Server列出资源resources/readClient→Server读取资源notifications/initializedClient→Server初始化完成通知7.2 工具定义模板{ name: tool_name, description: 工具描述, inputSchema: { type: object, properties: { param: { type: string|integer|boolean|array|object, description: 参数说明, enum: [option1, option2], default: option1 } }, required: [param] } }7.3 配置位置配置文件范围用途~/.claude/settings.json全局全局 MCP 服务器.claude/settings.json项目项目 MCP 服务器.claude/settings.local.json个人个人 MCP 配置结语MCP 协议是 Claude Code 扩展能力的核心机制。通过自定义 MCP 服务器可以集成企业内部 API、数据库、外部服务使 Claude Code 获得超越内置工具的能力。核心要点回顾stdio 通信MCP 服务器通过 stdin/stdout 的 JSON-RPC 通信stdout 必须纯净工具定义严格遵循 JSON Schema 规范定义 inputSchema错误处理工具失败时返回可读的 TextContent 错误文本超时管理长操作设超时避免 Claude Code 30s 后强制断开环境变量密钥通过 settings.json 的 env 配置传递调试方法直接用 JSON-RPC 消息测试服务器stderr 查看日志模板复用使用通用模板快速创建新 MCP 服务器Claude Code 集成在 settings.json 的 mcpServers 中配置用 /mcp 检查状态