# 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)