# 4.4 多服务器协同 **第4周 | 第4课 | 多 MCP Server 协同工作 | 预计时长：30分钟** --- ## 学习目标 完成本课后，你将能够： - 同时运行多个 MCP Server 并让一个 Agent 使用所有工具 - 设计合理的 Server 拆分策略 - 构建"交易大厅"综合场景 - 理解多 Server 场景下的安全考量 --- ## 1. 为什么需要多个 Server？ 在实际项目中，不同工具往往有独立的生命周期、部署环境和权限要求： ``` 不合理的做法（一个巨型 Server）： ┌──────────────────────────────────────────┐ │ mega-server.py │ │ │ │ 数据库连接 + API调用 + 文件操作 + 邮件 │ │ + 支付处理 + 消息队列 + ... │ │ │ │ 问题： │ │ ❌ 任何一个依赖更新都要重启整个 Server │ │ ❌ 无法单独控制权限（要么全给，要么全不给）│ │ ❌ 无法独立扩展和部署 │ │ ❌ 代码臃肿，难以维护 │ └──────────────────────────────────────────┘ 合理的做法（多个独立 Server）： ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ db-server │ │ api-server │ │ file-server │ │ (数据库) │ │ (外部API) │ │ (文件操作) │ │ │ │ │ │ │ │ list_users │ │ get_weather │ │ read_file │ │ get_order │ │ get_news │ │ write_file │ │ search_prod │ │ translate │ │ list_files │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ └─────────────────┼─────────────────┘ │ ┌─────────┴─────────┐ │ Agent Client │ │ (统一调用所有工具) │ └───────────────────┘ ``` **拆分原则**： | 考虑因素 | 拆分为多个 Server | 合并到一个 Server | |---------|-------------------|-------------------| | 独立性 | 高（独立部署/升级） | 低（一起部署/升级） | | 权限控制 | 细粒度（每个 Server 独立配置） | 粗粒度（全给或全不给） | | 复杂度 | 高（需要管理多个进程） | 低（一个进程） | | 适用场景 | 生产环境、多团队 | 个人项目、原型验证 | --- ## 2. 多 Server 运行方式 ### 2.1 Claude Desktop 配置 在 Claude Desktop 中，可以在配置文件里注册多个 Server： ```json { "mcpServers": { "database": { "command": "python", "args": ["db_server.py"], "env": {} }, "weather-api": { "command": "python", "args": ["api_server.py"], "env": {} }, "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/you/documents"], "env": {} } } } ``` Claude Desktop 启动时会同时启动所有配置的 Server，在对话中自动聚合所有工具供 LLM 选择。 ### 2.2 代码方式：聚合多个 Server 在自定义 Agent 中，你需要自己管理多个连接。以下是聚合模式的核心代码： ```python # multi_server_agent.py """ 多 Server Agent：同时连接多个 MCP Server，统一调用所有工具 """ import asyncio from dataclasses import dataclass from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client @dataclass class ServerConfig: """Server 配置""" name: str command: str args: list[str] # 定义所有要连接的 Server SERVERS = [ ServerConfig("database", "python", ["db_server.py"]), ServerConfig("weather-api", "python", ["api_server.py"]), ] class MultiServerAgent: """多 Server Agent：管理多个 MCP 连接""" def __init__(self): self.sessions: dict[str, ClientSession] = {} self._cleanup_funcs = [] async def connect_all(self): """连接所有 Server""" for server in SERVERS: print(f"🔌 连接 {server.name}...") # 创建 stdio 客户端 params = StdioServerParameters( command=server.command, args=server.args, ) # 建立连接并初始化 client = stdio_client(params) read_write = await client.__aenter__() self._cleanup_funcs.append(lambda: client.__aexit__(None, None, None)) session = ClientSession(read_write[0], read_write[1]) await session.__aenter__() self._cleanup_funcs.append(lambda: session.__aexit__(None, None, None)) await session.initialize() self.sessions[server.name] = session print(f" ✅ {server.name} 连接成功") async def list_all_tools(self) -> list[dict]: """列出所有 Server 的所有工具""" all_tools = [] for server_name, session in self.sessions.items(): tools_result = await session.list_tools() for tool in tools_result.tools: all_tools.append({ "server": server_name, "name": tool.name, "description": tool.description, "parameters": tool.inputSchema, }) return all_tools async def call_tool(self, tool_name: str, arguments: dict) -> str: """ 调用工具（自动查找在哪个 Server 上）。 实际应用中，工具名应该是全局唯一的， 或者通过工具名到 server 的映射来路由。 """ # 遍历所有 Server 查找工具 for server_name, session in self.sessions.items(): try: result = await session.call_tool(tool_name, arguments) return "\n".join(c.text for c in result.content) except Exception: # 这个 Server 上没有这个工具，继续找下一个 continue return f"错误：找不到工具 '{tool_name}'" async def cleanup(self): """关闭所有连接""" for cleanup in reversed(self._cleanup_funcs): try: await cleanup() except Exception: pass async def main(): agent = MultiServerAgent() try: # 1. 连接所有 Server await agent.connect_all() print() # 2. 列出所有可用工具 print("=" * 60) print("📋 所有可用工具：") print("=" * 60) tools = await agent.list_all_tools() for t in tools: print(f" [{t['server']}] {t['name']} — {t['description']}") print() # 3. 跨 Server 调用 print("=" * 60) print("🔄 跨 Server 调用演示：") print("=" * 60) # 从数据库 Server 查询用户 print("\n步骤 1：从数据库查询张三的信息") result = await agent.call_tool("get_user", {"user_id": 1}) print(result) # 从 API Server 查询天气 print("\n步骤 2：查询北京的实时天气") result = await agent.call_tool("get_realtime_weather", {"city": "北京"}) print(result) print("\n✅ 多 Server 协同调用完成！") finally: await agent.cleanup() if __name__ == "__main__": asyncio.run(main()) ``` --- ## 3. 实战项目：交易大厅 我们构建一个模拟的"交易大厅"场景，包含三个独立 Server： | Server | 职责 | 工具 | |--------|------|------| | 账户 Server | 用户账户管理 | 查余额、查持仓 | | 行情 Server | 市场价格 | 查股价、查汇率 | | 交易 Server | 交易执行 | 买入、卖出 | ### 3.1 账户 Server ```python # account_server.py """ 账户 Server：管理用户资金和持仓 """ from mcp.server.fastmcp import FastMCP mcp = FastMCP("account-server") # 模拟账户数据 _ACCOUNTS = { "user001": {"name": "张三", "balance": 100000.00, "currency": "CNY"}, "user002": {"name": "李四", "balance": 50000.00, "currency": "CNY"}, "user003": {"name": "王五", "balance": 200000.00, "currency": "CNY"}, } # 模拟持仓数据 _POSITIONS = { "user001": [ {"symbol": "600519", "name": "贵州茅台", "shares": 100, "cost": 1680.00}, {"symbol": "000858", "name": "五粮液", "shares": 500, "cost": 145.00}, ], "user002": [ {"symbol": "000001", "name": "平安银行", "shares": 2000, "cost": 12.50}, ], "user003": [], } @mcp.tool() def check_balance(user_id: str) -> str: """ 查询账户余额。 Args: user_id: 用户 ID（如 user001） Returns: 账户余额信息 """ if user_id not in _ACCOUNTS: return f"错误：用户 '{user_id}' 不存在" acc = _ACCOUNTS[user_id] return ( f"账户信息：\n" f" 用户：{acc['name']}\n" f" 账户：{user_id}\n" f" 可用余额：¥{acc['balance']:,.2f}\n" f" 货币：{acc['currency']}" ) @mcp.tool() def check_positions(user_id: str) -> str: """ 查询持仓情况。 Args: user_id: 用户 ID Returns: 持仓列表 """ if user_id not in _ACCOUNTS: return f"错误：用户 '{user_id}' 不存在" positions = _POSITIONS.get(user_id, []) acc = _ACCOUNTS[user_id] if not positions: return f"{acc['name']}（{user_id}）当前无持仓" lines = [f"{acc['name']} 的持仓："] lines.append("-" * 60) for pos in positions: lines.append( f" {pos['symbol']} {pos['name']} | " f"{pos['shares']} 股 | 成本 ¥{pos['cost']:.2f}" ) return "\n".join(lines) @mcp.tool() def get_account_summary(user_id: str) -> str: """ 获取账户总览（余额 + 持仓）。 Args: user_id: 用户 ID Returns: 账户总览信息 """ balance_info = check_balance(user_id) positions_info = check_positions(user_id) return f"{balance_info}\n\n{positions_info}" if __name__ == "__main__": mcp.run() ``` ### 3.2 行情 Server ```python # market_server.py """ 行情 Server：提供股票价格和汇率查询 """ from mcp.server.fastmcp import FastMCP mcp = FastMCP("market-server") # 模拟股票行情数据（实际项目中应从 API 获取） _STOCK_PRICES = { "600519": {"name": "贵州茅台", "price": 1685.00, "change": "+1.2%"}, "000858": {"name": "五粮液", "price": 148.50, "change": "-0.5%"}, "000001": {"name": "平安银行", "price": 12.85, "change": "+0.8%"}, "601318": {"name": "中国平安", "price": 48.20, "change": "+2.1%"}, "000333": {"name": "美的集团", "price": 68.90, "change": "-1.3%"}, "600036": {"name": "招商银行", "price": 35.60, "change": "+0.3%"}, } # 模拟汇率 _EXCHANGE_RATES = { "USD_CNY": 7.24, "EUR_CNY": 7.85, "JPY_CNY": 0.048, "HKD_CNY": 0.93, "GBP_CNY": 9.15, } @mcp.tool() def get_stock_price(symbol: str) -> str: """ 查询股票实时价格。 Args: symbol: 股票代码（如 600519） Returns: 股票价格信息 """ if symbol not in _STOCK_PRICES: available = ", ".join(_STOCK_PRICES.keys()) return f"错误：不支持的股票代码 '{symbol}'\n支持的代码：{available}" stock = _STOCK_PRICES[symbol] return ( f"{stock['name']}（{symbol}）：\n" f" 最新价：¥{stock['price']:.2f}\n" f" 涨跌幅：{stock['change']}" ) @mcp.tool() def get_exchange_rate(from_currency: str, to_currency: str) -> str: """ 查询汇率。 Args: from_currency: 源货币代码 to_currency: 目标货币代码 Returns: 汇率信息 """ key = f"{from_currency.upper()}_{to_currency.upper()}" if key not in _EXCHANGE_RATES: return f"错误：不支持 {from_currency} → {to_currency}" rate = _EXCHANGE_RATES[key] return f"{from_currency}/{to_currency} = {rate}" @mcp.tool() def calculate_trade_value(symbol: str, shares: int) -> str: """ 计算交易所需金额。 Args: symbol: 股票代码 shares: 买入股数 Returns: 预估交易金额 """ if symbol not in _STOCK_PRICES: return f"错误：不支持的股票代码 '{symbol}'" price = _STOCK_PRICES[symbol]["price"] total = price * shares commission = max(total * 0.0003, 5.00) # 最低 5 元手续费 stamp_tax = total * 0.001 # 印花税（卖出时收取） return ( f"交易预估（买入 {shares} 股 {symbol}）：\n" f" 股票金额：¥{total:,.2f}\n" f" 手续费： ¥{commission:,.2f}\n" f" 合计需： ¥{total + commission:,.2f}" ) if __name__ == "__main__": mcp.run() ``` ### 3.3 交易 Server ```python # trade_server.py """ 交易 Server：执行买入/卖出操作 """ from mcp.server.fastmcp import FastMCP from datetime import datetime mcp = FastMCP("trade-server") # 模拟交易记录 _TRADE_LOG = [] @mcp.tool() def buy_stock(user_id: str, symbol: str, shares: int, price: float) -> str: """ 执行买入操作。 Args: user_id: 用户 ID symbol: 股票代码 shares: 买入股数 price: 买入价格 Returns: 交易确认信息 """ total = shares * price commission = max(total * 0.0003, 5.00) total_cost = total + commission # 记录交易 trade = { "id": len(_TRADE_LOG) + 1, "user_id": user_id, "action": "买入", "symbol": symbol, "shares": shares, "price": price, "total": total_cost, "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "status": "已确认", } _TRADE_LOG.append(trade) return ( f"✅ 买入确认 #{trade['id']}：\n" f" 用户：{user_id}\n" f" 代码：{symbol}\n" f" 数量：{shares} 股\n" f" 价格：¥{price:.2f}\n" f" 金额：¥{total:,.2f}\n" f" 手续费：¥{commission:,.2f}\n" f" 合计：¥{total_cost:,.2f}\n" f" 时间：{trade['time']}\n" f" 状态：{trade['status']}" ) @mcp.tool() def sell_stock(user_id: str, symbol: str, shares: int, price: float) -> str: """ 执行卖出操作。 Args: user_id: 用户 ID symbol: 股票代码 shares: 卖出股数 price: 卖出价格 Returns: 交易确认信息 """ total = shares * price commission = max(total * 0.0003, 5.00) stamp_tax = total * 0.001 # 卖出收取印花税 net_proceeds = total - commission - stamp_tax trade = { "id": len(_TRADE_LOG) + 1, "user_id": user_id, "action": "卖出", "symbol": symbol, "shares": shares, "price": price, "total": net_proceeds, "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "status": "已确认", } _TRADE_LOG.append(trade) return ( f"✅ 卖出确认 #{trade['id']}：\n" f" 用户：{user_id}\n" f" 代码：{symbol}\n" f" 数量：{shares} 股\n" f" 价格：¥{price:.2f}\n" f" 金额：¥{total:,.2f}\n" f" 手续费：¥{commission:,.2f}\n" f" 印花税：¥{stamp_tax:,.2f}\n" f" 净收入：¥{net_proceeds:,.2f}\n" f" 时间：{trade['time']}\n" f" 状态：{trade['status']}" ) @mcp.tool() def get_trade_history(user_id: str) -> str: """ 查询用户交易历史。 Args: user_id: 用户 ID Returns: 交易记录列表 """ user_trades = [t for t in _TRADE_LOG if t["user_id"] == user_id] if not user_trades: return f"用户 {user_id} 暂无交易记录" lines = [f"{user_id} 交易历史（共 {len(user_trades)} 笔）："] lines.append("-" * 70) for t in user_trades: action_icon = "🟢" if t["action"] == "买入" else "🔴" lines.append( f" {action_icon} #{t['id']} {t['action']} {t['symbol']} | " f"{t['shares']}股×¥{t['price']:.2f} | " f"¥{t['total']:,.2f} | {t['time']}" ) return "\n".join(lines) if __name__ == "__main__": mcp.run() ``` ### 3.4 交易大厅 Agent ```python # trading_floor.py """ 交易大厅 Agent：模拟 Agent 同时使用三个 Server 完成交易 """ import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class TradingFloorAgent: """交易大厅 Agent""" def __init__(self): self.sessions = {} async def connect(self, name, command, args): """连接一个 Server""" params = StdioServerParameters(command=command, args=args) client = stdio_client(params) rw = await client.__aenter__() session = ClientSession(rw[0], rw[1]) await session.__aenter__() await session.initialize() self.sessions[name] = session print(f" ✅ {name} 已连接") return session async def call(self, session_name, tool, args): """调用工具""" session = self.sessions[session_name] result = await session.call_tool(tool, args) return "\n".join(c.text for c in result.content) async def close_all(self): """关闭所有连接""" for s in self.sessions.values(): try: await s.__aexit__(None, None, None) except Exception: pass async def main(): agent = TradingFloorAgent() try: # === 连接三个 Server === print("🔌 连接交易大厅...") await agent.connect("account", "python", ["account_server.py"]) await agent.connect("market", "python", ["market_server.py"]) await agent.connect("trade", "python", ["trade_server.py"]) print() # === 模拟一个完整的交易流程 === user_id = "user001" symbol = "601318" # 中国平安 print("=" * 60) print("🏦 交易大厅 — Agent 自主交易流程") print("=" * 60) # 步骤 1：检查账户 print("\n📋 步骤 1：检查账户余额") print("-" * 40) result = await agent.call("account", "check_balance", {"user_id": user_id}) print(result) # 步骤 2：查看持仓 print("\n📋 步骤 2：查看当前持仓") print("-" * 40) result = await agent.call("account", "check_positions", {"user_id": user_id}) print(result) # 步骤 3：查询目标股票价格 print("\n📈 步骤 3：查询目标股票价格") print("-" * 40) result = await agent.call("market", "get_stock_price", {"symbol": symbol}) print(result) # 步骤 4：计算交易金额 print("\n📊 步骤 4：计算预估交易金额（买入 200 股）") print("-" * 40) result = await agent.call("market", "calculate_trade_value", { "symbol": symbol, "shares": 200 }) print(result) # 步骤 5：执行买入 print("\n💰 步骤 5：执行买入操作") print("-" * 40) result = await agent.call("trade", "buy_stock", { "user_id": user_id, "symbol": symbol, "shares": 200, "price": 48.20 }) print(result) # 步骤 6：查询交易历史 print("\n📜 步骤 6：确认交易记录") print("-" * 40) result = await agent.call("trade", "get_trade_history", {"user_id": user_id}) print(result) print("\n" + "=" * 60) print("✅ 交易完成！Agent 成功调用 3 个 Server 的 6 个工具") print("=" * 60) finally: await agent.close_all() if __name__ == "__main__": asyncio.run(main()) ``` 运行： ```bash python trading_floor.py ``` --- ## 4. Server 拆分策略 ### 4.1 什么时候拆分？ - **不同团队维护** → 各自独立的 Server - **不同的数据源 / API** → 按数据源拆分 - **权限隔离**（读 vs 写）→ 按权限拆分 - **独立部署需求** → 各自独立 ### 4.2 什么时候合并？ - **同一个数据源**（如都是数据库操作）→ 合并 - **高度耦合**（工具 A 必须调用工具 B 的结果）→ 合并 - **小型项目**（2-3 个工具）→ 合并更简单 ### 4.3 命名规范 多 Server 场景下，工具命名必须有规范： ``` ✅ 好的命名： db_list_users — 清楚来自数据库 Server api_get_weather — 清楚来自 API Server file_read — 清楚来自文件 Server ❌ 差的命名： list — 什么列表？ get — 获取什么？ query — 查询什么？ ``` **最佳实践**：在工具描述中注明数据来源和 Server 名称。 ```python @mcp.tool() def db_list_users(page: int = 1) -> str: """ [db-server] 列出所有用户（分页）。 数据来源：trade_db 数据库 users 表。 """ ... ``` --- ## 5. 安全考量 多 Server 环境下，安全是首要关注点： ### 5.1 工具权限 ``` 高风险工具（谨慎暴露）： ❌ delete_user — 删除用户 ❌ execute_sql — 执行任意 SQL ❌ send_email — 发送邮件 ❌ transfer_money — 转账 中风险工具（加确认）： ⚠️ buy_stock — 买入（需要金额确认） ⚠️ update_user — 修改用户信息 ⚠️ delete_record — 删除记录 低风险工具（可暴露）： ✅ list_users — 列表查询 ✅ get_weather — 天气查询 ✅ get_stock_price — 价格查询 ``` ### 5.2 输入验证 Server 端的工具必须对所有输入做严格验证： ```python @mcp.tool() def safe_query(user_id: str, max_rows: int = 100) -> str: """安全查询：带参数验证""" # 验证用户 ID 格式 if not user_id.startswith("user") or not user_id[4:].isdigit(): return "错误：无效的用户 ID 格式" # 限制返回行数（防止数据泄露） max_rows = min(max_rows, 100) # 上限 100 # 使用参数化查询（防止 SQL 注入） conn = _get_db() cursor = conn.cursor() cursor.execute( "SELECT id, name, email FROM users WHERE id = ? LIMIT ?", (user_id, max_rows), ) # ... ``` ### 5.3 审计日志 ```python import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger("mcp-audit") @mcp.tool() def sensitive_operation(user_id: str, action: str) -> str: """敏感操作：记录审计日志""" logger.info(f"AUDIT: user={user_id}, action={action}") # 执行操作... ``` --- ## 动手练习 ### 练习 1：完成交易大厅 在 `trading_floor.py` 的基础上，添加一个卖出流程： 1. 查询 user001 持有的茅台股票 2. 查询茅台当前价格 3. 计算卖出 50 股的预估收入 4. 执行卖出操作 5. 查询交易历史确认 ### 练习 2：添加风控工具 在交易 Server 中添加风控工具： ```python @mcp.tool() def risk_check(user_id: str, symbol: str, shares: int, price: float) -> str: """ 交易前风控检查。 检查项： 1. 交易金额是否超过账户余额的 30% 2. 单只股票持仓是否超过总资产的 50% 3. 交易金额是否超过单笔限额（5万元） 返回检查结果和是否允许交易。 """ ... ``` 在买入流程中，先调用风控检查，通过后才执行买入。 ### 练习 3：多 Server 错误处理 模拟以下场景并处理： 1. 行情 Server 挂了（修改 `market_server.py` 让它启动时报错） 2. Agent 如何优雅降级？（跳过行情查询，使用缓存价格） 3. 交易 Server 返回"余额不足"，Agent 如何反馈给用户？ --- ## 本课总结 - **多 Server 协同**：一个 Agent 可以同时连接多个 MCP Server，聚合所有工具 - **拆分策略**：按数据源、团队、权限拆分；高度耦合的工具合并 - **交易大厅案例**：三个 Server（账户、行情、交易）协同完成完整交易流程 - **安全原则**： - 工具分级：高/中/低风险 - 输入验证：格式、范围、SQL 注入 - 审计日志：记录所有敏感操作 - **命名规范**：工具名和描述要清楚标识来源 Server --- ## 本周总结 本周我们系统学习了 MCP 协议： | 课程 | 内容 | 核心技能 | |------|------|---------| | 4.1 | MCP 是什么 | 理解 MCP 架构和三种核心操作 | | 4.2 | 搭建 MCP Server | 用 Python SDK 编写工具和 Client | | 4.3 | 接入数据库和 API | 连接 SQLite 和 HTTP API | | 4.4 | 多 Server 协同 | 多个 Server 同时工作、安全考量 | **本周核心收获**：MCP 让"工具"成为可插拔的模块。写好一个 MCP Server，任何支持 MCP 的 Agent 都能使用你的工具——这就是标准化带来的复利效应。