freebird2913

1
                ┌─────────────────────────┐
2
                │      AI Application      │
3
                │    (Claude Desktop /     │
4
                │     VS Code / 自建Agent)  │
5
                └──────────┬──────────────┘
6
                           │ MCP Protocol
7
                ┌──────────▼──────────────┐
8
                │      MCP Server          │
9
                │                          │
10
                │  ┌──────┐ ┌──────┐ ┌───┐ │
11
                │  │Tools │ │Resrcs│ │Pr │ │
12
                │  │(动作)│ │(数据)│ │(话术)│ │
13
                │  └──────┘ └──────┘ └───┘ │
14
                └──────────────────────────┘

1
# 之前的做法：一次性注册所有工具
2
@mcp.tool()
3
def search_database(query: str): ...
4
@mcp.tool()
5
def send_email(to: str, content: str): ...
6
@mcp.tool()
7
def generate_report(template: str): ...
8
# ... 还有47个工具
9

10
# 新方案：使用 ToolSearch 技术
11
# 客户端不再是"全部加载"，而是先加载一个"工具发现工具"
12
# 当 AI 需要搜索数据库时，才去查询 search_database 的具体信息

1
❌ 旧方式：
2
   AI: 调用 search_user(id=1) → 返回结果
3
   AI: 调用 get_orders(user_id=1) → 返回结果
4
   AI: 调用 calculate_total(orders) → 返回结果
5
   AI: 总结...
6

7
✅ 新方式：
8
   AI: 直接生成Python脚本
9
       user = search_user(1)
10
       orders = get_orders(user['id'])
11
       return calculate_total(orders)

1
@mcp.tool()
2
def read_file(path: str): ...
3
@mcp.tool()
4
def write_file(path: str, content: str): ...
5
@mcp.tool()
6
def delete_file(path: str): ...
7
@mcp.tool()
8
def list_files(dir: str): ...
9
@mcp.tool()
10
def copy_file(src: str, dst: str): ...
11
# ... 没完没了

1
@mcp.tool()
2
def execute_file_operations(operations_json: str) -> str:
3
    """
4
    批量执行文件操作。
5
    参数是一个JSON数组，支持 read/write/delete/copy/move/list 等操作。
6
    示例：[{"op": "read", "path": "/tmp/test.txt"}]
7
    """
8
    operations = json.loads(operations_json)
9
    results = []
10
    for op in operations:
11
        # 在沙箱环境中执行
12
        results.append(process_operation(op))
13
    return json.dumps(results)

1
❌ 没有 Gateway：
2
   Agent ──→ MCP Server A (直接暴露)
3
         ──→ MCP Server B (直接暴露)
4
         ──→ MCP Server C (直接暴露)
5
   → 没有权限控制、没有审计日志、没有速率限制
6

7
✅ 有 Gateway：
8
   Agent ──→ MCP Gateway ──→ MCP Server A
9
                      ├──→ MCP Server B
10
                      └──→ MCP Server C
11
   → 统一认证、授权、限流、审计、脱敏

1
# mcp_gateway.py — 轻量级 MCP 网关
2
from fastapi import FastAPI, HTTPException, Request
3
import httpx
4
import json
5
import time
6

7
app = FastAPI()
8

9
# MCP 服务器注册表
10
servers_registry = {
11
    "weather": {
12
        "url": "http://localhost:8001/mcp",
13
        "description": "天气查询服务",
14
        "rate_limit": 100,  # 每分钟100次
15
    },
16
    "database": {
17
        "url": "http://localhost:8002/mcp",
18
        "description": "数据库查询服务",
19
        "rate_limit": 50,
20
    },
21
    "filesystem": {
22
        "url": "http://localhost:8003/mcp",
23
        "description": "文件系统服务",
24
        "rate_limit": 200,
25
    }
26
}
27

28
# 简单的速率限制器
29
rate_limits = {}
30

31
def check_rate_limit(server_name: str) -> bool:
32
    now = time.time()
33
    if server_name not in rate_limits:
34
        rate_limits[server_name] = []
35

36
    # 清理过期记录
37
    rate_limits[server_name] = [
38
        t for t in rate_limits[server_name] if now - t < 60
39
    ]
40

41
    limit = servers_registry[server_name]["rate_limit"]
42
    if len(rate_limits[server_name]) >= limit:
43
        return False
44

45
    rate_limits[server_name].append(now)
46
    return True
47

48

49
@app.post("/mcp/{server_name}")
50
async def route_mcp_request(server_name: str, request: Request):
51
    if server_name not in servers_registry:
52
        raise HTTPException(status_code=404, detail="MCP server not found")
53

54
    # 速率限制检查
55
    if not check_rate_limit(server_name):
56
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
57

58
    # 转发请求到目标 MCP 服务器
59
    body = await request.body()
60
    target = servers_registry[server_name]
61

62
    async with httpx.AsyncClient() as client:
63
        response = await client.post(
64
            target["url"],
65
            content=body,
66
            headers={"Content-Type": "application/json"},
67
            timeout=30
68
        )
69

70
    return response.json()
71

72

73
@app.get("/discovery")
74
async def discover_servers():
75
    """返回所有可用的 MCP 服务器列表"""
76
    return {
77
        name: {"description": info["description"]}
78
        for name, info in servers_registry.items()
79
    }

1
# 查询天气 → 路由到 weather 服务器
2
curl -X POST http://localhost:8000/mcp/weather \
3
  -H "Content-Type: application/json" \
4
  -d '{"jsonrpc": "2.0", "method": "tools/call", ...}'
5

6
# 发现可用服务
7
curl http://localhost:8000/discovery
8
# → {"weather": {"description": "天气查询服务"}, ...}

1
# 安全启动MCP服务器的模板
2
from mcp.server.fastmcp import FastMCP
3
import os
4

5
mcp = FastMCP("SecureServer")
6

7
# 1. 永远不要暴露在公网
8
# 只在 127.0.0.1 监听
9
if __name__ == "__main__":
10
    mcp.run(
11
        transport='streamable-http',
12
        host='127.0.0.1',       # 👈 只本地监听
13
        port=8000,
14
        path='/mcp'
15
    )

1
# orchestrator.py — 多MCP Server协调器
2
import asyncio
3
from mcp import ClientSession, StdioServerParameters
4
from mcp.client.stdio import stdio_client
5
import json
6

7

8
class MCPOrchestrator:
9
    """多MCP Server协调器"""
10

11
    def __init__(self):
12
        self.servers = {}
13
        self.sessions = {}
14
        self.tool_registry = {}
15

16
    async def connect_server(self, name: str, command: str, args: list):
17
        """连接一个MCP Server"""
18
        server_params = StdioServerParameters(
19
            command=command,
20
            args=args
21
        )
22

23
        stdio_transport = await stdio_client(server_params)
24
        read, write = stdio_transport
25
        session = await ClientSession(read, write).__aenter__()
26
        await session.initialize()
27

28
        self.servers[name] = session
29
        print(f"✅ 已连接 MCP Server: {name}")
30

31
        # 获取该服务器的所有工具并注册
32
        tools_response = await session.list_tools()
33
        for tool in tools_response.tools:
34
            self.tool_registry[tool.name] = {
35
                "server": name,
36
                "description": tool.description,
37
                "input_schema": tool.inputSchema
38
            }
39
            print(f"  📌 注册工具: {tool.name} ({tool.description})")
40

41
    async def call_tool(self, tool_name: str, arguments: dict):
42
        """调用指定工具（自动路由到对应的Server）"""
43
        if tool_name not in self.tool_registry:
44
            return f"错误: 未找到工具 {tool_name}"
45

46
        server_name = self.tool_registry[tool_name]["server"]
47
        session = self.servers[server_name]
48

49
        result = await session.call_tool(tool_name, arguments)
50
        return result.content
51

52
    def list_all_tools(self) -> str:
53
        """列出所有可用工具"""
54
        if not self.tool_registry:
55
            return "没有已注册的工具"
56

57
        result = "📋 可用工具清单：\n"
58
        for name, info in self.tool_registry.items():
59
            result += f"\n  🔧 {name}\n"
60
            result += f"     服务器: {info['server']}\n"
61
            result += f"     描述: {info['description']}\n"
62
        return result
63

64
    async def close_all(self):
65
        """关闭所有连接"""
66
        for name, session in self.servers.items():
67
            await session.__aexit__(None, None, None)
68
            print(f"🔌 已断开: {name}")
69

70

71
async def main():
72
    orchestrator = MCPOrchestrator()
73

74
    # 连接三个MCP Server
75
    await orchestrator.connect_server("Weather", "python", ["weather_server.py"])
76
    await orchestrator.connect_server("File", "python", ["file_server.py"])
77
    await orchestrator.connect_server("Calc", "python", ["calc_server.py"])
78

79
    print("\n" + "="*50)
80
    print(orchestrator.list_all_tools())
81
    print("="*50)
82

83
    # 演示：依次调用不同服务器的工具
84
    print("\n🟢 用户: 北京今天天气怎么样？")
85
    result = await orchestrator.call_tool("get_weather", {"city": "北京"})
86
    print(f"🤖 AI: {result}")
87

88
    print("\n🟢 用户: 帮我算一下 sin(30) + cos(60) 等于多少？")
89
    result = await orchestrator.call_tool("calculate", {"expression": "sin(30) + cos(60)"})
90
    print(f"🤖 AI: {result}")
91

92
    print("\n🟢 用户: 列出我所有的笔记")
93
    result = await orchestrator.call_tool("list_notes", {})
94
    print(f"🤖 AI: {result}")
95

96
    await orchestrator.close_all()
97

98

99
if __name__ == "__main__":
100
    asyncio.run(main())

1
✅ 已连接 MCP Server: Weather
2
  📌 注册工具: get_weather (获取城市天气)
3
✅ 已连接 MCP Server: File
4
  📌 注册工具: read_note (读取工作笔记文件)
5
  📌 注册工具: list_notes (列出所有工作笔记)
6
✅ 已连接 MCP Server: Calc
7
  📌 注册工具: calculate (执行数学计算)
8

9
📋 可用工具清单：
10
  🔧 get_weather
11
     服务器: Weather
12
     描述: 获取城市天气
13
  🔧 read_note
14
     服务器: File
15
     描述: 读取工作笔记文件
16
  🔧 list_notes
17
     服务器: File
18
     描述: 列出所有工作笔记
19
  🔧 calculate
20
     服务器: Calc
21
     描述: 执行数学计算
22

23
🟢 用户: 北京今天天气怎么样？
24
🤖 AI: ☀️ 25°C 晴
25

26
🟢 用户: 帮我算一下 sin(30) + cos(60) 等于多少？
27
🤖 AI: sin(30) + cos(60) = 1.0
28

29
🟢 用户: 列出我所有的笔记
30
🤖 AI: ['todo.md', 'ideas.md']

1
# 访问某个网站时，Agent 会自动检查
2
GET https://example.com/.well-known/mcp.json
3
# → {"servers": [{"name": "docs", "url": "..."}, ...]}