1
// MCP 工具定义的 JSON Schema 示例
2
{
3
  "name": "search_codebase",
4
  "description": "在代码库中搜索匹配特定模式的代码片段",
5
  "inputSchema": {
6
    "type": "object",
7
    "properties": {
8
      "pattern": {
9
        "type": "string",
10
        "description": "搜索的正则表达式或文本模式"
11
      },
12
      "fileGlob": {
13
        "type": "string",
14
        "description": "文件过滤模式，如 *.py 或 *.tsx",
15
        "default": "*"
16
      },
17
      "maxResults": {
18
        "type": "integer",
19
        "description": "最大返回结果数",
20
        "default": 20
21
      }
22
    },
23
    "required": ["pattern"]
24
  }
25
}

1
# 使用 claude-context MCP 服务器
2
from mcp import ClientSession, StdioServerParameters
3

4
# 启动 claude-context 的 MCP 服务器
5
server_params = StdioServerParameters(
6
    command="python",
7
    args=["-m", "claude_context.server"]
8
)
9

10
async with ClientSession(server_params) as session:
11
    # 列出可用工具
12
    tools = await session.list_tools()
13
    print("可用工具:", [t.name for t in tools])
14
    # 输出: ['store_memory', 'retrieve_memory', 'search_knowledge', 'index_project']
15

16
    # 存储关于项目架构的记忆
17
    await session.call_tool("store_memory", {
18
        "key": "project/arch/auth",
19
        "content": "认证模块使用 JWT + Redis 会话缓存，Token 有效期 2 小时",
20
        "tags": ["auth", "architecture", "jwt"]
21
    })
22

23
    # 下次会话时检索
24
    result = await session.call_tool("retrieve_memory", {
25
        "query": "认证模块是怎么设计的？",
26
        "max_results": 3
27
    })
28
    print(result.content[0].text)

1
# 一个简单的 MCP 服务器：获取当前系统信息
2
import psutil
3
import json
4
import sys
5
from mcp.server import Server, stdio_server
6

7
app = Server("system-info")
8

9
@app.tool("get_system_stats")
10
async def get_system_stats(include_disk: bool = False) -> str:
11
    """获取系统资源使用情况统计"""
12
    cpu_percent = psutil.cpu_percent(interval=1)
13
    memory = psutil.virtual_memory()
14

15
    result = {
16
        "cpu": f"{cpu_percent}%",
17
        "memory": {
18
            "total": f"{memory.total / 1024**3:.1f}GB",
19
            "used": f"{memory.used / 1024**3:.1f}GB",
20
            "percent": f"{memory.percent}%"
21
        },
22
        "processes": len(psutil.pids())
23
    }
24

25
    if include_disk:
26
        disk = psutil.disk_usage("/")
27
        result["disk"] = {
28
            "total": f"{disk.total / 1024**3:.1f}GB",
29
            "used": f"{disk.used / 1024**3:.1f}GB",
30
            "percent": f"{disk.percent}%"
31
        }
32

33
    return json.dumps(result, ensure_ascii=False)
34

35
if __name__ == "__main__":
36
    stdio_server.run(app)

1
# MCP 客户端 — 任何模型都可以用，不限于某个厂商
2
from mcp import ClientSession
3

4
async def extract_data_with_any_model(model, query: str):
5
    """
6
    让任何模型通过 MCP 工具获取外部数据
7
    model 参数可以是 OpenAI、Anthropic、GLM 等任意兼容的 API
8
    """
9
    async with ClientSession() as session:
10
        tools = await session.list_tools()
11
        result = await session.call_tool("query_database", {
12
            "sql": f"SELECT * FROM datasets WHERE description LIKE '%{query}%'"
13
        })
14
        response = await model.generate(
15
            prompt=f"基于以下数据回答用户问题：{result.content[0].text}",
16
            tools=tools  # 模型可以自主决定是否使用工具
17
        )
18
        return response

1
# 给内部监控系统添加 MCP 接口
2
from mcp.server import Server, stdio_server
3

4
app = Server("monitor")
5

6
@app.tool("check_service_health")
7
async def check_health(service_name: str) -> str:
8
    """检查指定服务的健康状态"""
9
    result = await monitor_client.query(f"SELECT * FROM health WHERE service='{service_name}'")
10
    return format_health_report(result)
11

12
@app.tool("list_alerts")
13
async def list_active_alerts(severity: str = "critical") -> str:
14
    """列出当前活跃的报警信息"""
15
    alerts = await monitor_client.query(f"SELECT * FROM alerts WHERE severity='{severity}' AND resolved=false")
16
    return format_alerts(alerts)