1
┌──────────────────────┐
2
│   AI 模型 / Agent     │  ← Claude、GPT、Gemini 等
3
│   (MCP Client)        │
4
└────────┬─────────────┘
5
         │ JSON-RPC over stdio/SSE/WebSocket
6
         ▼
7
┌──────────────────────┐
8
│   MCP Protocol       │  ← 标准化的协议层
9
│   (工具发现/调用/通知) │
10
└────────┬─────────────┘
11
         ▼
12
┌──────────────────────┐
13
│   MCP Server         │  ← 你的工具/API/数据库
14
│   (暴露能力给 AI)     │
15
└──────────────────────┘

1
// MCP Server 的最小实现
2
import { Server } from "@modelcontextprotocol/sdk";
3

4
const server = new Server({
5
  name: "my-database-server",
6
  version: "1.0.0"
7
}, {
8
  capabilities: {
9
    resources: {},  // 暴露数据资源
10
    tools: {},      // 暴露可调用工具
11
    prompts: {}     // 暴露模板提示词
12
  }
13
});
14

15
// 1. 列出可用工具
16
server.setRequestHandler("tools/list", async () => ({
17
  tools: [{
18
    name: "query_database",
19
    description: "对 PostgreSQL 数据库执行 SQL 查询",
20
    inputSchema: {
21
      type: "object",
22
      properties: {
23
        sql: { type: "string", description: "SQL 查询语句" },
24
        limit: { type: "number", default: 100 }
25
      }
26
    }
27
  }]
28
}));
29

30
// 2. 调用工具
31
server.setRequestHandler("tools/call", async (request) => {
32
  if (request.params.name === "query_database") {
33
    const { sql, limit } = request.params.arguments;
34
    const result = await db.query(sql, { limit });
35
    return {
36
      content: [{ type: "text", text: JSON.stringify(result) }]
37
    };
38
  }
39
});
40

41
// 3. 启动服务
42
server.connect(transport);

1
# code_analyzer_server.py — MCP Server for code analysis
2
import subprocess
3
import json
4
from mcp.server import Server, NotificationOptions
5
from mcp.server.models import InitializationOptions
6

7
server = Server("code-analyzer")
8

9
@server.list_tools()
10
async def handle_list_tools() -> list[dict]:
11
    return [
12
        {
13
            "name": "count_lines_of_code",
14
            "description": "统计指定目录下的代码行数",
15
            "inputSchema": {
16
                "type": "object",
17
                "properties": {
18
                    "path": {"type": "string", "description": "目录路径"},
19
                    "extensions": {
20
                        "type": "array",
21
                        "items": {"type": "string"},
22
                        "description": "文件扩展名过滤, 如 ['.py', '.js']"
23
                    }
24
                }
25
            }
26
        },
27
        {
28
            "name": "run_tests",
29
            "description": "运行测试并返回结果摘要",
30
            "inputSchema": {
31
                "type": "object",
32
                "properties": {
33
                    "path": {"type": "string", "description": "项目路径"},
34
                    "test_pattern": {"type": "string", "description": "测试匹配模式, 如 'test_*'"}
35
                }
36
            }
37
        }
38
    ]
39

40
@server.call_tool()
41
async def handle_call_tool(name: str, arguments: dict) -> list[dict]:
42
    if name == "count_lines_of_code":
43
        path = arguments["path"]
44
        exts = arguments.get("extensions", [".py"])
45
        result = subprocess.run(
46
            ["cloc", "--json"] + [f"--include-lang={e.lstrip('.')}" for e in exts] + [path],
47
            capture_output=True, text=True
48
        )
49
        data = json.loads(result.stdout)
50
        return [{"type": "text", "text": json.dumps(data, indent=2)}]
51

52
    elif name == "run_tests":
53
        path = arguments["path"]
54
        pattern = arguments.get("test_pattern", "test_*")
55
        result = subprocess.run(
56
            ["pytest", "-x", "--tb=short", f"-k={pattern}", path],
57
            capture_output=True, text=True
58
        )
59
        summary = {
60
            "passed": "passed" in result.stdout and "failed" not in result.stdout,
61
            "output": result.stdout[-2000:],
62
            "return_code": result.returncode
63
        }
64
        return [{"type": "text", "text": json.dumps(summary, indent=2)}]
65

66
# 启动
67
if __name__ == "__main__":
68
    from mcp.server.stdio import stdio_server
69
    async with stdio_server() as (read, write):
70
        await server.run(read, write, InitializationOptions(
71
            server_name="code-analyzer",
72
            server_version="1.0.0"
73
        ))

1
{
2
  "mcpServers": {
3
    "code-analyzer": {
4
      "command": "python",
5
      "args": ["code_analyzer_server.py"]
6
    }
7
  }
8
}