# 2-1 Client-Server 架構詳解 回到白皮書首頁:[MCP 全方位技術白皮書](/@thc1006/mcp-whitepaper-home) --- ## MCP 的三層架構:不只是 Client-Server,而是智慧化生態系統 當我們談到 MCP 的 Client-Server 架構時,很多人第一個想法可能是「又是一個傳統的客戶端-伺服器模式」。但實際上,MCP 創建了一個**三層智慧化協作架構**,徹底重新定義了 AI 與外部世界的互動方式。 ## 核心架構:三個角色的協同演出 ### MCP Host:AI 的大腦中樞 **MCP Host** 是整個生態系統的「指揮官」,通常是 AI 應用程式、IDE 或桌面客戶端。 **主要職責:** - **決策制定**:根據用戶需求決定需要什麼工具 - **資源協調**:管理多個 MCP Client 的連接 - **上下文維護**:保持對話狀態和記憶 - **結果整合**:將來自不同 MCP Server 的結果統一處理 **實際例子:Claude Desktop** ``` 用戶:「幫我分析這個專案的 GitHub 活動,並更新專案文件」 Claude Desktop (MCP Host): 1. 理解需求:需要 GitHub 數據 + 文件操作 2. 決定策略:先獲取 GitHub 數據,再進行文件處理 3. 協調 Client:指示不同的 MCP Client 執行任務 4. 整合結果:將分析結果寫入文件 ``` ### MCP Client:協議翻譯官 **MCP Client** 是協議的具體實現者,負責維持與 MCP Server 的一對一連接。 **核心功能:** - **協議轉換**:將 Host 的高層次指令轉為 MCP 協議調用 - **連接管理**:維護與 Server 的穩定連接 - **錯誤處理**:處理網路異常、超時等問題 - **安全管控**:執行認證、授權和資料驗證 **技術實現:** ```python class MCPClient: def __init__(self, server_uri: str): self.server_uri = server_uri self.transport = None self.session = None async def connect(self): """建立與 MCP Server 的連接""" self.transport = await self._create_transport() self.session = await self._handshake() async def list_tools(self) -> List[Tool]: """發現 Server 提供的工具""" response = await self.session.request("tools/list") return [Tool.from_dict(tool) for tool in response.tools] async def call_tool(self, name: str, args: dict) -> dict: """調用特定工具""" response = await self.session.request("tools/call", { "name": name, "arguments": args }) return response.content ``` ### MCP Server:專業工具提供者 **MCP Server** 是實際執行工作的「專家」,每個 Server 通常專精於特定領域。 **核心職責:** - **能力暴露**:向 Client 描述自己提供的工具和資源 - **任務執行**:執行 Client 請求的具體操作 - **資料轉換**:將外部系統的資料轉換為 MCP 標準格式 - **安全控制**:確保操作在授權範圍內 **GitHub MCP Server 例子:** ```python class GitHubMCPServer: def __init__(self, github_token: str): self.github = GitHub(token=github_token) def list_tools(self) -> List[Tool]: return [ Tool( name="get_repository", description="獲取 GitHub 倉庫資訊", parameters={ "owner": {"type": "string", "required": True}, "repo": {"type": "string", "required": True} } ), Tool( name="list_issues", description="列出倉庫的 Issues", parameters={ "owner": {"type": "string", "required": True}, "repo": {"type": "string", "required": True}, "state": {"type": "string", "enum": ["open", "closed", "all"]} } ) ] async def handle_tool_call(self, tool_name: str, arguments: dict): if tool_name == "get_repository": return await self._get_repository(arguments) elif tool_name == "list_issues": return await self._list_issues(arguments) else: raise ToolNotFoundError(f"Unknown tool: {tool_name}") ``` ## 資料流向與控制流程 ### 完整的請求-響應週期 ``` 用戶輸入 ↓ MCP Host (AI 決策) ↓ MCP Client (協議轉換) ↓ MCP Server (工具執行) ↓ 外部系統 (實際操作) ↓ 結果回傳 (逆向流動) ``` ### 實際案例:分析專案活動 **步驟一:需求理解** ``` 用戶:「分析我們專案最近一週的開發活動」 MCP Host: - 解析需求:需要 GitHub 資料分析 - 識別所需工具:GitHub API、數據分析工具 - 制定執行計畫:獲取 commits → 分析趨勢 → 生成報告 ``` **步驟二:工具發現** ``` MCP Host → MCP Client:「列出所有可用工具」 MCP Client → GitHub MCP Server:GET /tools/list GitHub MCP Server → MCP Client: { "tools": [ {"name": "get_commits", "description": "獲取提交記錄"}, {"name": "get_pull_requests", "description": "獲取 PR 資訊"}, {"name": "get_contributors", "description": "獲取貢獻者資訊"} ] } ``` **步驟三:數據獲取** ``` MCP Host → MCP Client:「調用 get_commits 工具」 MCP Client → GitHub MCP Server: { "method": "tools/call", "params": { "name": "get_commits", "arguments": { "repo": "myproject", "since": "2025-08-09T00:00:00Z", "until": "2025-08-16T23:59:59Z" } } } GitHub MCP Server → GitHub API:實際 API 調用 GitHub API → GitHub MCP Server:返回 commits 數據 GitHub MCP Server → MCP Client:格式化的結果 MCP Client → MCP Host:結構化數據 ``` **步驟四:結果處理** ``` MCP Host: - 接收所有數據 - 進行智慧化分析 - 生成人類可讀的報告 - 返回給用戶 ``` ## 傳輸層與通訊協議 ### 支援的傳輸方式 **1. HTTP/HTTPS (最常用)** ```python # RESTful 風格的 MCP over HTTP transport = HTTPTransport("https://api.example.com/mcp") client = MCPClient(transport) ``` **2. WebSocket (即時通訊)** ```python # 適合需要雙向即時通訊的場景 transport = WebSocketTransport("wss://realtime.example.com/mcp") client = MCPClient(transport) ``` **3. Unix Domain Socket (本地通訊)** ```python # 適合同機器上的 MCP Server transport = UnixSocketTransport("/tmp/mcp-server.sock") client = MCPClient(transport) ``` **4. Standard I/O (程序間通訊)** ```python # 適合可執行程式形式的 MCP Server transport = StdioTransport("./my-mcp-server") client = MCPClient(transport) ``` ### JSON-RPC 2.0 協議層 MCP 建立在 JSON-RPC 2.0 之上,確保標準化的通訊格式: **請求格式:** ```json { "jsonrpc": "2.0", "id": "req-123", "method": "tools/call", "params": { "name": "get_file_content", "arguments": { "path": "/home/user/document.txt" } } } ``` **響應格式:** ```json { "jsonrpc": "2.0", "id": "req-123", "result": { "content": "檔案內容...", "encoding": "utf-8", "size": 1024 } } ``` **錯誤格式:** ```json { "jsonrpc": "2.0", "id": "req-123", "error": { "code": -32602, "message": "Invalid params", "data": { "details": "Path parameter is required" } } } ``` ## 連接生命週期管理 ### 連接建立流程 ``` 1. Transport 建立 ↓ 2. 能力協商 (Capability Negotiation) ↓ 3. 身份驗證 (Authentication) ↓ 4. 工具發現 (Tool Discovery) ↓ 5. 正常通訊開始 ``` **能力協商範例:** ```json Client → Server: { "jsonrpc": "2.0", "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": { "tools": {"listChanged": true}, "resources": {"subscribe": true}, "prompts": {"listChanged": true} }, "clientInfo": { "name": "Claude Desktop", "version": "1.0.0" } } } Server → Client: { "jsonrpc": "2.0", "result": { "protocolVersion": "2025-06-18", "capabilities": { "tools": {"listChanged": true}, "resources": {"listChanged": true}, "prompts": {"listChanged": false} }, "serverInfo": { "name": "GitHub MCP Server", "version": "2.1.0" } } } ``` ### 心跳與健康檢查 ```python class MCPClient: async def _maintain_connection(self): """維護連接健康""" while self.connected: try: # 發送 ping 檢查連接 await self.ping() await asyncio.sleep(30) # 30 秒心跳間隔 except ConnectionError: await self._reconnect() except Exception as e: self.logger.error(f"Health check failed: {e}") async def _reconnect(self): """重新連接邏輯""" for attempt in range(3): try: await self.connect() self.logger.info("Reconnected successfully") break except Exception as e: await asyncio.sleep(2 ** attempt) # 指數退避 ``` ## 多 Server 協調與負載均衡 ### Host 的智慧化調度 ```python class MCPHost: def __init__(self): self.servers = {} self.capabilities_map = {} async def register_server(self, name: str, client: MCPClient): """註冊新的 MCP Server""" self.servers[name] = client tools = await client.list_tools() # 建立能力映射 for tool in tools: if tool.category not in self.capabilities_map: self.capabilities_map[tool.category] = [] self.capabilities_map[tool.category].append({ "server": name, "tool": tool }) async def execute_task(self, task_description: str): """智慧化選擇最適合的 Server 執行任務""" # AI 分析任務需求 required_capabilities = self.analyze_task(task_description) # 選擇最適合的 Server best_server = self.select_best_server(required_capabilities) # 執行任務 return await self.servers[best_server].execute_task(task_description) ``` ### 容錯與備援機制 ```python class ResilientMCPClient: def __init__(self, primary_servers: List[str], backup_servers: List[str]): self.primary_servers = primary_servers self.backup_servers = backup_servers self.current_server_index = 0 async def call_tool_with_fallback(self, tool_name: str, args: dict): """帶有故障轉移的工具調用""" servers_to_try = self.primary_servers + self.backup_servers for server_uri in servers_to_try: try: client = await self.get_client(server_uri) result = await client.call_tool(tool_name, args) return result except Exception as e: self.logger.warning(f"Server {server_uri} failed: {e}") continue raise AllServersFailedError("No available servers") ``` ## 安全性與訪問控制 ### 傳輸層安全 ```python class SecureMCPTransport: def __init__(self, server_uri: str, cert_path: str = None): self.server_uri = server_uri self.ssl_context = self._create_ssl_context(cert_path) def _create_ssl_context(self, cert_path: str): context = ssl.create_default_context() if cert_path: context.load_verify_locations(cert_path) context.check_hostname = True context.verify_mode = ssl.CERT_REQUIRED return context ``` ### OAuth 2.1 整合 ```python class OAuth21MCPClient: def __init__(self, client_id: str, client_secret: str): self.oauth_client = OAuth21Client(client_id, client_secret) self.access_token = None async def authenticate(self): """執行 OAuth 2.1 認證流程""" self.access_token = await self.oauth_client.get_access_token() async def make_authenticated_request(self, method: str, params: dict): """帶有認證的請求""" headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json" } return await self._send_request(method, params, headers) ``` ## 台灣企業部署策略 ### 混合雲架構 ``` 本地 MCP Servers (敏感數據) ├─ 財務系統 MCP Server ├─ 人事系統 MCP Server └─ 內部文件 MCP Server 雲端 MCP Servers (一般服務) ├─ GitHub MCP Server ├─ Google Workspace MCP Server └─ Slack MCP Server 邊緣 MCP Servers (即時服務) ├─ IoT 設備 MCP Server ├─ 製造設備 MCP Server └─ 監控系統 MCP Server ``` ### 段階式部署建議 **第一階段:概念驗證 (1-2 個月)** - 部署 2-3 個簡單的 MCP Server - 建立基本的 MCP Client 連接 - 驗證核心功能 **第二階段:生產試點 (3-4 個月)** - 整合核心業務系統 - 建立安全和監控機制 - 培訓內部團隊 **第三階段:全面推廣 (6-12 個月)** - 擴展到所有部門 - 建立企業級治理機制 - 持續最佳化和創新 ## 效能最佳化策略 ### 連接池管理 ```python class MCPConnectionPool: def __init__(self, max_connections: int = 10): self.pool = asyncio.Queue(maxsize=max_connections) self.active_connections = {} async def get_connection(self, server_uri: str) -> MCPClient: """從連接池獲取連接""" if server_uri in self.active_connections: return self.active_connections[server_uri] if not self.pool.empty(): client = await self.pool.get() await client.connect_to(server_uri) self.active_connections[server_uri] = client return client # 建立新連接 client = MCPClient(server_uri) await client.connect() self.active_connections[server_uri] = client return client ``` ### 快取機制 ```python class CachedMCPClient: def __init__(self, cache_ttl: int = 300): # 5 分鐘 TTL self.cache = {} self.cache_ttl = cache_ttl async def list_tools_cached(self) -> List[Tool]: """帶快取的工具列表""" cache_key = "tools_list" if cache_key in self.cache: cached_time, tools = self.cache[cache_key] if time.time() - cached_time < self.cache_ttl: return tools # 快取過期,重新獲取 tools = await self.client.list_tools() self.cache[cache_key] = (time.time(), tools) return tools ``` ## 小結:架構設計的核心價值 MCP 的 Client-Server 架構不僅僅是技術實現,更是一個**智慧化化、標準化、可擴展**的生態系統設計: 1. **智慧化協調**:Host 提供高層次決策,Client 處理協議細節,Server 專注領域專業 2. **標準化通訊**:統一的 JSON-RPC 2.0 協議確保互操作性 3. **彈性部署**:支援多種傳輸方式,適應不同部署需求 4. **安全可控**:內建的認證、授權和加密機制 5. **高可用性**:故障轉移和負載均衡確保服務穩定 這個架構為台灣企業提供了一個既能保持技術領先,又能確保安全可控的 AI 整合方案。 --- **下一頁:** [2-2 三大核心機制:Tools、Resources、Prompts](/s/mcp-core-mechanisms)