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

适用版本：Claude Code v1.0.x 及以上、MCP Protocol v2024-11-05
受影响场景：自定义工具集成、外部数据源接入、企业内部 API 桥接
阅读时长：约 30 分钟

目录

  1. 问题现象
  2. 原理深挖：MCP 协议架构
  3. 根因分析：MCP 开发中的常见问题
  4. 多方案解决：从开发到部署
  5. 验证回归：MCP 服务器验证
  6. 避坑最佳实践
  7. 附录：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 lost

2. 原理深挖：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(
    stream=sys.stderr,
    level=logging.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(
            name="query_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(
            name="list_tables",
            description="列出指定数据库中的所有表",
            inputSchema={
                "type": "object",
                "properties": {
                    "database": {
                        "type": "string",
                        "enum": ["main", "analytics", "logs"],
                        "default": "main"
                    }
                }
            }
        ),
        Tool(
            name="describe_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(f"Tool 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(
                type="text",
                text=f"错误: 未知工具 '{name}'"
            )]
    except Exception as e:
        logger.error(f"Tool error: {e}")
        return [TextContent(
            type="text",
            text=f"工具执行错误: {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(
            type="text",
            text="错误: 仅支持 SELECT 查询"
        )]
    
    # 模拟查询 (实际使用 psycopg2/pymysql 等)
    try:
        # 模拟数据
        results = [
            {"id": 1, "name": "Alice", "email": "alice@example.com"},
            {"id": 2, "name": "Bob", "email": "bob@example.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(type="text", text=f"查询结果 ({database}):\n\n{table}")]
    
    except Exception as e:
        return [TextContent(type="text", text=f"查询失败: {e}")]

async def handle_list_tables(args):
    """列出表"""
    database = args.get("database", "main")
    tables = ["users", "orders", "products", "inventory"]
    return [TextContent(
        type="text",
        text=f"数据库 {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(type="text", text=desc)]

# 主函数
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, 不干扰 stdout

4.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 2>&1 >/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(seconds=30):
    """工具调用超时装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(
                    func(*args, **kwargs),
                    timeout=seconds
                )
            except asyncio.TimeoutError:
                return [TextContent(
                    type="text",
                    text=f"错误: 工具执行超时 ({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(type="text", text=f"连接错误: {e}")]
        except ValueError as e:
            return [TextContent(type="text", text=f"参数错误: {e}")]
        except PermissionError as e:
            return [TextContent(type="text", text=f"权限错误: {e}")]
        except Exception as e:
            logger.error(f"Unexpected error: {e}", exc_info=True)
            return [TextContent(type="text", text=f"内部错误: {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(
    stream=sys.stderr,
    level=logging.INFO,
    format='[MCP:%(name)s] %(levelname)s %(message)s'
)
logger = logging.getLogger("template")

server = Server("template-server")

# ============================================
# 工具定义 (修改此处)
# ============================================

TOOLS = [
    Tool(
        name="example_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(type="text", text=result)]

# 工具路由
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_info=True)
            return [TextContent(type="text", text=f"错误: {e}")]
    else:
        return [TextContent(type="text", text=f"未知工具: {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; d=json.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
d=json.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; d=json.load(sys.stdin); assert 'result' in d" 2>/dev/null; then
    echo "✓"
else
    echo "✗ 工具调用失败"
    exit 1
fi

echo ""
echo "=== 所有测试通过 ==="

5.2 验证清单

# 验证项 预期 方法
1 服务器启动 无错误 直接运行
2 初始化响应 有 result JSON-RPC 测试
3 工具列表 tools > 0 tools/list
4 工具调用 有结果 tools/call
5 错误处理 返回错误文本 无效参数
6 stdout 纯净 仅 JSON-RPC 检查无其他输出
7 stderr 日志 有日志 2>&1 >/dev/null
8 Claude 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 常见陷阱

# 陷阱 后果 解决
1 stdout 输出日志 JSON-RPC 解析失败 日志到 stderr
2 Schema 不规范 Claude 调用失败 遵循 JSON Schema
3 无超时 Claude 等待 30s 后断开 工具设超时
4 密钥硬编码 安全风险 用环境变量
5 同步阻塞 响应慢 用 async/await
6 不处理 EOF 服务器不退出 检测 stdin 关闭
7 无错误处理 Claude 收到异常 返回 TextContent
8 env 不传递 连接失败 settings.json 配置 env

7. 附录：MCP 协议速查表

7.1 JSON-RPC 方法

方法 方向 说明
initialize Client→Server 初始化握手
tools/list Client→Server 列出工具
tools/call Client→Server 调用工具
resources/list Client→Server 列出资源
resources/read Client→Server 读取资源
notifications/initialized Client→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 获得超越内置工具的能力。

核心要点回顾：

  1. stdio 通信：MCP 服务器通过 stdin/stdout 的 JSON-RPC 通信，stdout 必须纯净
  2. 工具定义：严格遵循 JSON Schema 规范定义 inputSchema
  3. 错误处理：工具失败时返回可读的 TextContent 错误文本
  4. 超时管理：长操作设超时，避免 Claude Code 30s 后强制断开
  5. 环境变量：密钥通过 settings.json 的 env 配置传递
  6. 调试方法：直接用 JSON-RPC 消息测试服务器，stderr 查看日志
  7. 模板复用：使用通用模板快速创建新 MCP 服务器
  8. Claude Code 集成：在 settings.json 的 mcpServers 中配置，用 /mcp 检查状态
# 开发工具
Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

Harness Engineering 实战：用 Spring AI Alibaba 构建可控的 AI 智能体

文章摘要 本文介绍了如何基于Spring AI Alibaba框架和Harness Engineering理念构建可控的AI智能体。Harness Engineering是一种新兴的AI工程化范式，强调为AI智能体构建可靠、可控的运行环境（"马具"），而非仅优化模型本身。文章详细阐述了Harness Engineering的关键组件（规则、技能、门禁等），并提供了Windows环境下的实践指南，包

avatar MCP技术社区

20_项目实战六_企业文档智能检索平台_LangSmith监控_权限管控_数据看板

前面我们已经做过个人知识库问答助手、客服机器人、SQL 查询助手、代码审查 Agent 和多 Agent 日报系统。这些项目解决的是单点能力。多个部门都要上传文档。不同用户只能看自己有权限的文档。支持 PDF、Word、Markdown、网页、内部 Wiki。文档更新后要增量入库。用户提问时要返回答案和引用来源。管理员要看到 QPS、延迟、Token 消耗、用户活跃度。出问题时要能追踪每一次检索、

avatar MCP技术社区

AI Agent 跑完任务怎么通知你？我写了个推送服务讲解一

将 Excel 转换为 TXT 不仅是格式的简单转换，更是实现数据自动化流程的重要环节。数据清洗与预处理： 将复杂格式的 Excel 数据转换为 TXT，作为其他数据分析工具（如Spark、Hadoop）的输入源，进行更深层次的处理。配置管理： 将 Excel 中维护的系统配置、参数表等导出为 TXT 格式，便于脚本直接读取，同时更适合使用 Git 等工具进行版本管理。自动化报告生成： 从 Exc

avatar MCP技术社区