MCP AI應用通信的底層機制

技術小館專注AI與Java領域的前沿技術知識庫技術小館官網

在AI應用快速發展的今天，不同AI系統之間的高效通信成為技術架構的關鍵挑戰。MCP（Model Context Protocol）作為新一代AI應用通信協議，正在重新定義AI工具生態的構建方式。

想象一下，當你需要讓ChatGPT調用外部API、訪問數據庫或執行復雜計算時，傳統的插件機制往往面臨兼容性差、擴展性有限的困境。MCP協議的出現，為這個問題提供了優雅的解決方案。通過標準化的通信接口和靈活的架構設計，MCP讓AI應用能夠無縫集成各種外部服務和工具，就像給AI裝上了"萬能工具箱"。無論是數據查詢、文件處理還是復雜業務邏輯，MCP都能讓AI應用以統一的方式調用和協作。

一、MCP協議概述

什么是MCP協議

MCP（Model Context Protocol）是一個開放標準協議，專門設計用于AI模型與外部工具和服務之間的通信。它采用JSON-RPC 2.0規范，通過標準化的消息格式實現AI應用與各種外部資源的無縫集成。

{"jsonrpc": "2.0","id": 1,"method": "tools/list","params": {}
}

MCP的核心思想是提供一個統一的接口層，讓AI模型能夠像調用本地函數一樣調用遠程服務，同時保持協議的簡潔性和擴展性。

MCP的設計目標與核心價值

MCP的設計遵循幾個核心原則：

標準化：統一的協議規范，確保不同實現之間的互操作性
安全性：內置認證和授權機制，保護敏感數據
可擴展性：支持動態工具發現和注冊
性能優化：高效的序列化和傳輸機制

這些設計目標使得MCP成為構建AI工具生態的理想選擇。

MCP在AI生態中的定位

在當前的AI技術棧中，MCP扮演著"連接器"的角色。它位于AI模型層和應用服務層之間，為兩者提供標準化的通信橋梁。

AI模型層 (ChatGPT, Claude等)↓
MCP協議層 (標準化通信接口)↓
應用服務層 (數據庫、API、文件系統等)

這種分層架構使得AI應用能夠靈活地集成各種外部資源，而不需要為每種服務開發專門的適配器。

二、MCP的核心架構

協議層設計原理

MCP采用分層設計，每一層都有明確的職責：

傳輸層：處理底層網絡通信，支持TCP、WebSocket等多種傳輸方式
協議層：實現JSON-RPC 2.0規范，處理消息的序列化和反序列化
應用層：定義具體的工具接口和業務邏輯

# MCP服務器基礎實現示例
import json
import asyncioclass MCPServer:def __init__(self):self.tools = {}async def handle_request(self, request):"""處理MCP請求"""try:method = request.get("method")params = request.get("params", {})if method == "tools/list":return {"result": list(self.tools.keys())}elif method == "tools/call":tool_name = params.get("name")tool_params = params.get("arguments", {})return await self.call_tool(tool_name, tool_params)except Exception as e:return {"error": {"code": -1, "message": str(e)}}

客戶端-服務器模型

MCP采用經典的客戶端-服務器架構，其中：

客戶端：通常是AI模型或AI應用，發起工具調用請求
服務器：提供各種工具和服務，響應客戶端的請求

# MCP客戶端示例
class MCPClient:def __init__(self, server_url):self.server_url = server_urlasync def list_tools(self):"""獲取可用工具列表"""request = {"jsonrpc": "2.0","id": 1,"method": "tools/list","params": {}}return await self.send_request(request)async def call_tool(self, tool_name, arguments):"""調用指定工具"""request = {"jsonrpc": "2.0","id": 2,"method": "tools/call","params": {"name": tool_name,"arguments": arguments}}return await self.send_request(request)

消息傳遞機制

MCP使用JSON-RPC 2.0作為消息傳遞協議，支持三種消息類型：

請求消息：客戶端向服務器發送的調用請求
響應消息：服務器返回的處理結果
通知消息：單向消息，不需要響應

// 請求消息示例
{"jsonrpc": "2.0","id": 1,"method": "tools/call","params": {"name": "database/query","arguments": {"sql": "SELECT * FROM users WHERE age > 18"}}
}// 響應消息示例
{"jsonrpc": "2.0","id": 1,"result": {"data": [{"id": 1, "name": "張三", "age": 25},{"id": 2, "name": "李四", "age": 30}]}
}

錯誤處理與容錯機制

MCP定義了標準的錯誤碼和錯誤處理機制：

# 錯誤處理示例
class MCPError(Exception):def __init__(self, code, message, data=None):self.code = codeself.message = messageself.data = data# 標準錯誤碼
ERROR_CODES = {-32700: "解析錯誤",-32600: "無效請求",-32601: "方法未找到",-32602: "無效參數",-32603: "內部錯誤"
}def handle_error(error_code, message):"""統一錯誤處理"""return {"jsonrpc": "2.0","error": {"code": error_code,"message": message}}

三、MCP協議詳解

請求-響應模式

MCP采用同步的請求-響應模式，每個請求都有唯一的ID，確保響應的正確匹配：

class RequestManager:def __init__(self):self.pending_requests = {}self.request_id = 0def create_request(self, method, params):"""創建新請求"""self.request_id += 1request = {"jsonrpc": "2.0","id": self.request_id,"method": method,"params": params}self.pending_requests[self.request_id] = requestreturn requestdef handle_response(self, response):"""處理響應"""request_id = response.get("id")if request_id in self.pending_requests:del self.pending_requests[request_id]return responsereturn None

數據格式與序列化

MCP使用JSON作為數據交換格式，支持復雜的數據結構：

# 數據類型示例
class MCPDataTypes:@staticmethoddef serialize_tool_result(data):"""序列化工具調用結果"""if isinstance(data, (dict, list)):return json.dumps(data, ensure_ascii=False)return str(data)@staticmethoddef deserialize_tool_params(params_str):"""反序列化工具參數"""try:return json.loads(params_str)except json.JSONDecodeError:raise MCPError(-32700, "參數格式錯誤")

認證與安全機制

MCP支持多種認證方式，確保通信安全：

# 認證機制示例
class MCPAuth:def __init__(self):self.api_keys = {}def authenticate(self, headers):"""驗證請求身份"""api_key = headers.get("Authorization")if not api_key:raise MCPError(-32001, "缺少認證信息")if api_key not in self.api_keys:raise MCPError(-32002, "無效的API密鑰")return Truedef add_api_key(self, key, permissions):"""添加API密鑰"""self.api_keys[key] = permissions

版本兼容性策略

MCP采用語義化版本控制，確保向后兼容：

# 版本兼容性處理
class MCPVersionManager:def __init__(self):self.supported_versions = ["1.0", "1.1", "1.2"]def check_compatibility(self, client_version):"""檢查版本兼容性"""major, minor = client_version.split(".")for supported in self.supported_versions:s_major, s_minor = supported.split(".")if major == s_major and int(minor) <= int(s_minor):return Truereturn Falsedef negotiate_version(self, client_versions):"""協商最佳版本"""for version in client_versions:if version in self.supported_versions:return versionreturn self.supported_versions[0]

四、MCP的實際應用場景

AI應用集成外部API

MCP最常見的應用場景是讓AI模型調用外部API服務：

# 天氣API集成示例
class WeatherTool:def __init__(self, api_key):self.api_key = api_keyasync def get_weather(self, city):"""獲取城市天氣信息"""url = f"https://api.weatherapi.com/v1/current.json"params = {"key": self.api_key,"q": city,"aqi": "no"}async with aiohttp.ClientSession() as session:async with session.get(url, params=params) as response:data = await response.json()return {"city": city,"temperature": data["current"]["temp_c"],"condition": data["current"]["condition"]["text"]}# 注冊到MCP服務器
mcp_server = MCPServer()
weather_tool = WeatherTool("your_api_key")
mcp_server.tools["weather/get"] = weather_tool.get_weather

數據庫連接與查詢

MCP可以安全地讓AI模型訪問數據庫：

# 數據庫工具示例
class DatabaseTool:def __init__(self, connection_string):self.connection_string = connection_stringasync def query_database(self, sql, params=None):"""執行數據庫查詢"""try:async with asyncpg.connect(self.connection_string) as conn:result = await conn.fetch(sql, *params or [])return [dict(row) for row in result]except Exception as e:raise MCPError(-32603, f"數據庫查詢錯誤: {str(e)}")async def execute_sql(self, sql, params=None):"""執行SQL語句"""try:async with asyncpg.connect(self.connection_string) as conn:result = await conn.execute(sql, *params or [])return {"affected_rows": result}except Exception as e:raise MCPError(-32603, f"SQL執行錯誤: {str(e)}")# 注冊數據庫工具
db_tool = DatabaseTool("postgresql://user:pass@localhost/db")
mcp_server.tools["database/query"] = db_tool.query_database
mcp_server.tools["database/execute"] = db_tool.execute_sql

文件系統操作

MCP支持安全的文件系統操作：

# 文件系統工具示例
class FileSystemTool:def __init__(self, base_path):self.base_path = Path(base_path)async def read_file(self, file_path):"""讀取文件內容"""full_path = self.base_path / file_pathif not full_path.exists():raise MCPError(-32003, "文件不存在")try:with open(full_path, 'r', encoding='utf-8') as f:return {"content": f.read(), "path": str(full_path)}except Exception as e:raise MCPError(-32603, f"文件讀取錯誤: {str(e)}")async def write_file(self, file_path, content):"""寫入文件內容"""full_path = self.base_path / file_pathtry:full_path.parent.mkdir(parents=True, exist_ok=True)with open(full_path, 'w', encoding='utf-8') as f:f.write(content)return {"success": True, "path": str(full_path)}except Exception as e:raise MCPError(-32603, f"文件寫入錯誤: {str(e)}")# 注冊文件系統工具
fs_tool = FileSystemTool("/safe/path")
mcp_server.tools["filesystem/read"] = fs_tool.read_file
mcp_server.tools["filesystem/write"] = fs_tool.write_file

自定義工具開發

開發者可以創建自定義工具來擴展AI的能力：

# 自定義計算工具示例
class CalculatorTool:def __init__(self):self.supported_operations = ['+', '-', '*', '/', '**']async def calculate(self, expression):"""安全計算數學表達式"""try:# 安全檢查，只允許基本數學運算allowed_chars = set('0123456789+-*/.() ')if not all(c in allowed_chars for c in expression):raise ValueError("表達式包含不允許的字符")result = eval(expression)return {"expression": expression, "result": result}except Exception as e:raise MCPError(-32602, f"計算錯誤: {str(e)}")async def get_statistics(self, numbers):"""計算統計信息"""try:nums = [float(n) for n in numbers]return {"mean": sum(nums) / len(nums),"max": max(nums),"min": min(nums),"count": len(nums)}except Exception as e:raise MCPError(-32602, f"統計計算錯誤: {str(e)}")# 注冊計算工具
calc_tool = CalculatorTool()
mcp_server.tools["calculator/compute"] = calc_tool.calculate
mcp_server.tools["calculator/stats"] = calc_tool.get_statistics

五、MCP與其他協議的對比

與OpenAI插件協議的差異

MCP相比OpenAI插件協議具有以下優勢：

標準化程度更高：MCP是開放標準，不依賴特定廠商
擴展性更強：支持動態工具發現和注冊
安全性更好：內置認證和授權機制

# OpenAI插件 vs MCP對比示例
# OpenAI插件方式
class OpenAIPlugin:def __init__(self):self.manifest = {"schema_version": "v1","name_for_model": "weather_plugin","name_for_human": "天氣查詢插件","description_for_model": "獲取城市天氣信息","description_for_human": "查詢指定城市的天氣情況","auth": {"type": "none"},"api": {"type": "openapi", "url": "https://api.example.com/openapi.yaml"},"logo_url": "https://example.com/logo.png","contact_email": "support@example.com","legal_info_url": "https://example.com/legal"}# MCP方式
class MCPTool:def __init__(self):self.tool_info = {"name": "weather/get","description": "獲取城市天氣信息","parameters": {"type": "object","properties": {"city": {"type": "string", "description": "城市名稱"}},"required": ["city"]}}

與REST API的對比分析

MCP相比傳統REST API的優勢：

統一接口：所有工具通過相同的協議調用
類型安全：支持強類型參數定義
錯誤處理：標準化的錯誤碼和錯誤信息

# REST API vs MCP對比
# 傳統REST API調用
async def rest_api_call():async with aiohttp.ClientSession() as session:# 需要為每個API維護不同的調用方式weather_response = await session.get("https://api.weather.com/current?city=北京")db_response = await session.post("https://api.database.com/query", json={"sql": "SELECT * FROM users"})file_response = await session.get("https://api.files.com/read?path=/data.txt")# MCP統一調用
async def mcp_call():client = MCPClient("mcp://localhost:3000")# 所有工具使用相同的調用方式weather = await client.call_tool("weather/get", {"city": "北京"})users = await client.call_tool("database/query", {"sql": "SELECT * FROM users"})content = await client.call_tool("filesystem/read", {"path": "/data.txt"})

性能與擴展性優勢

MCP在性能和擴展性方面的優勢：

# 性能對比示例
import time
import asyncioclass PerformanceTest:def __init__(self):self.mcp_client = MCPClient("mcp://localhost:3000")async def test_mcp_performance(self):"""測試MCP性能"""start_time = time.time()# 并發調用多個工具tasks = [self.mcp_client.call_tool("weather/get", {"city": "北京"}),self.mcp_client.call_tool("database/query", {"sql": "SELECT * FROM users"}),self.mcp_client.call_tool("calculator/compute", {"expression": "2+2*3"})]results = await asyncio.gather(*tasks)end_time = time.time()return {"total_time": end_time - start_time,"results": results}async def test_scalability(self):"""測試擴展性"""# 動態添加新工具new_tool = {"name": "custom/process","description": "自定義數據處理工具","handler": lambda data: {"processed": data.upper()}}# 無需重啟服務器即可使用新工具result = await self.mcp_client.call_tool("custom/process", {"data": "hello world"})return result

六、MCP開發實踐指南

環境搭建與配置

開始MCP開發需要搭建基礎環境：

# 環境配置示例
# requirements.txt
mcp==1.0.0
aiohttp==3.8.5
asyncpg==0.28.0
pydantic==2.0.0# 基礎配置文件
import yamlclass MCPConfig:def __init__(self, config_path="mcp_config.yaml"):with open(config_path, 'r', encoding='utf-8') as f:self.config = yaml.safe_load(f)@propertydef server_host(self):return self.config.get("server", {}).get("host", "localhost")@propertydef server_port(self):return self.config.get("server", {}).get("port", 3000)@propertydef auth_token(self):return self.config.get("auth", {}).get("token")@propertydef tools_config(self):return self.config.get("tools", {})# mcp_config.yaml
server:host: localhostport: 3000debug: trueauth:token: "your_secret_token"enabled: truetools:weather:api_key: "your_weather_api_key"enabled: truedatabase:connection_string: "postgresql://user:pass@localhost/db"enabled: truefilesystem:base_path: "/safe/path"enabled: true

基礎客戶端開發

開發MCP客戶端的基本步驟：

# 基礎MCP客戶端實現
import asyncio
import aiohttp
import json
from typing import Dict, Anyclass MCPClient:def __init__(self, server_url: str, auth_token: str = None):self.server_url = server_urlself.auth_token = auth_tokenself.session = Noneself.request_id = 0async def __aenter__(self):"""異步上下文管理器入口"""self.session = aiohttp.ClientSession()return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):"""異步上下文管理器出口"""if self.session:await self.session.close()async def send_request(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:"""發送MCP請求"""self.request_id += 1request = {"jsonrpc": "2.0","id": self.request_id,"method": method,"params": params or {}}headers = {"Content-Type": "application/json"}if self.auth_token:headers["Authorization"] = f"Bearer {self.auth_token}"async with self.session.post(self.server_url,json=request,headers=headers) as response:result = await response.json()if "error" in result:raise MCPError(result["error"]["code"],result["error"]["message"])return result["result"]async def list_tools(self) -> list:"""獲取可用工具列表"""return await self.send_request("tools/list")async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:"""調用指定工具"""return await self.send_request("tools/call", {"name": tool_name,"arguments": arguments})# 使用示例
async def main():async with MCPClient("http://localhost:3000", "your_token") as client:# 獲取可用工具tools = await client.list_tools()print(f"可用工具: {tools}")# 調用天氣工具weather = await client.call_tool("weather/get", {"city": "北京"})print(f"天氣信息: {weather}")# 調用數據庫工具users = await client.call_tool("database/query", {"sql": "SELECT * FROM users LIMIT 5"})print(f"用戶數據: {users}")if __name__ == "__main__":asyncio.run(main())

自定義工具實現

創建自定義MCP工具的完整示例：

# 自定義工具實現
from typing import Dict, Any, Optional
import asyncio
import jsonclass CustomTool:"""自定義MCP工具基類"""def __init__(self, name: str, description: str):self.name = nameself.description = descriptionself.parameters = {}def add_parameter(self, name: str, param_type: str, description: str, required: bool = False):"""添加工具參數"""self.parameters[name] = {"type": param_type,"description": description,"required": required}async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:"""執行工具邏輯（子類需要重寫）"""raise NotImplementedError("子類必須實現execute方法")def get_info(self) -> Dict[str, Any]:"""獲取工具信息"""return {"name": self.name,"description": self.description,"parameters": {"type": "object","properties": self.parameters,"required": [name for name, param in self.parameters.items() if param.get("required", False)]}}# 圖像處理工具示例
class ImageProcessingTool(CustomTool):def __init__(self):super().__init__("image/process", "圖像處理工具")self.add_parameter("image_url", "string", "圖像URL", required=True)self.add_parameter("operation", "string", "處理操作(resize/rotate/filter)", required=True)self.add_parameter("parameters", "object", "處理參數")async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:"""執行圖像處理"""image_url = arguments["image_url"]operation = arguments["operation"]params = arguments.get("parameters", {})# 模擬圖像處理if operation == "resize":width = params.get("width", 800)height = params.get("height", 600)return {"status": "success","operation": "resize","new_size": f"{width}x{height}","processed_url": f"{image_url}_resized"}elif operation == "rotate":angle = params.get("angle", 90)return {"status": "success","operation": "rotate","angle": angle,"processed_url": f"{image_url}_rotated"}else:raise MCPError(-32602, f"不支持的操作: {operation}")# 數據分析工具示例
class DataAnalysisTool(CustomTool):def __init__(self):super().__init__("data/analyze", "數據分析工具")self.add_parameter("data", "array", "要分析的數據", required=True)self.add_parameter("analysis_type", "string", "分析類型(statistics/trend/correlation)", required=True)async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:"""執行數據分析"""data = arguments["data"]analysis_type = arguments["analysis_type"]if not isinstance(data, list):raise MCPError(-32602, "數據必須是數組格式")if analysis_type == "statistics":return self._calculate_statistics(data)elif analysis_type == "trend":return self._analyze_trend(data)elif analysis_type == "correlation":return self._calculate_correlation(data)else:raise MCPError(-32602, f"不支持的分析類型: {analysis_type}")def _calculate_statistics(self, data: list) -> Dict[str, Any]:"""計算統計信息"""if not data:return {"error": "數據為空"}numeric_data = [float(x) for x in data if isinstance(x, (int, float))]if not numeric_data:return {"error": "沒有有效的數值數據"}return {"count": len(numeric_data),"sum": sum(numeric_data),"mean": sum(numeric_data) / len(numeric_data),"min": min(numeric_data),"max": max(numeric_data),"median": sorted(numeric_data)[len(numeric_data) // 2]}def _analyze_trend(self, data: list) -> Dict[str, Any]:"""分析趨勢"""if len(data) < 2:return {"error": "數據點不足，無法分析趨勢"}numeric_data = [float(x) for x in data if isinstance(x, (int, float))]if len(numeric_data) < 2:return {"error": "數值數據點不足"}# 簡單趨勢分析first_half = numeric_data[:len(numeric_data)//2]second_half = numeric_data[len(numeric_data)//2:]first_avg = sum(first_half) / len(first_half)second_avg = sum(second_half) / len(second_half)trend = "上升" if second_avg > first_avg else "下降" if second_avg < first_avg else "平穩"return {"trend": trend,"change_rate": (second_avg - first_avg) / first_avg * 100,"first_half_avg": first_avg,"second_half_avg": second_avg}def _calculate_correlation(self, data: list) -> Dict[str, Any]:"""計算相關性（簡化版）"""if len(data) < 4:return {"error": "數據點不足，無法計算相關性"}# 假設數據是成對的if len(data) % 2 != 0:return {"error": "數據必須是成對的"}pairs = [(data[i], data[i+1]) for i in range(0, len(data), 2)]x_values = [pair[0] for pair in pairs if isinstance(pair[0], (int, float))]y_values = [pair[1] for pair in pairs if isinstance(pair[1], (int, float))]if len(x_values) != len(y_values) or len(x_values) < 2:return {"error": "有效的數據對不足"}# 簡化的相關系數計算n = len(x_values)sum_x = sum(x_values)sum_y = sum(y_values)sum_xy = sum(x * y for x, y in zip(x_values, y_values))sum_x2 = sum(x * x for x in x_values)sum_y2 = sum(y * y for y in y_values)numerator = n * sum_xy - sum_x * sum_ydenominator = ((n * sum_x2 - sum_x * sum_x) * (n * sum_y2 - sum_y * sum_y)) ** 0.5if denominator == 0:correlation = 0else:correlation = numerator / denominatorreturn {"correlation_coefficient": correlation,"data_points": n,"interpretation": "強正相關" if correlation > 0.7 else "中等正相關" if correlation > 0.3 else"弱正相關" if correlation > 0 else"強負相關" if correlation < -0.7 else"中等負相關" if correlation < -0.3 else"弱負相關" if correlation < 0 else "無相關"}# 注冊自定義工具
def register_custom_tools(mcp_server: MCPServer):"""注冊自定義工具到MCP服務器"""# 注冊圖像處理工具image_tool = ImageProcessingTool()mcp_server.tools[image_tool.name] = image_tool.execute# 注冊數據分析工具data_tool = DataAnalysisTool()mcp_server.tools[data_tool.name] = data_tool.executereturn mcp_server

調試與測試技巧

MCP開發的調試和測試策略：

# 調試和測試工具
import logging
import unittest
from unittest.mock import Mock, patchclass MCPDebugger:def __init__(self):self.logger = logging.getLogger("mcp_debugger")self.logger.setLevel(logging.DEBUG)# 添加控制臺處理器handler = logging.StreamHandler()formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')handler.setFormatter(formatter)self.logger.addHandler(handler)def log_request(self, request: Dict[str, Any]):"""記錄請求日志"""self.logger.info(f"發送請求: {json.dumps(request, ensure_ascii=False, indent=2)}")def log_response(self, response: Dict[str, Any]):"""記錄響應日志"""self.logger.info(f"收到響應: {json.dumps(response, ensure_ascii=False, indent=2)}")def log_error(self, error: Exception):"""記錄錯誤日志"""self.logger.error(f"發生錯誤: {str(error)}", exc_info=True)# MCP測試框架
class MCPTestCase(unittest.TestCase):def setUp(self):"""測試設置"""self.mcp_server = MCPServer()self.mcp_client = MCPClient("http://localhost:3000")self.debugger = MCPDebugger()async def test_tool_registration(self):"""測試工具注冊"""# 注冊測試工具async def test_tool(arguments):return {"result": "test_success", "input": arguments}self.mcp_server.tools["test/tool"] = test_tool# 驗證工具列表tools = await self.mcp_client.list_tools()self.assertIn("test/tool", tools)async def test_tool_execution(self):"""測試工具執行"""# 注冊測試工具async def calculator(arguments):a = arguments.get("a", 0)b = arguments.get("b", 0)operation = arguments.get("operation", "+")if operation == "+":return {"result": a + b}elif operation == "-":return {"result": a - b}elif operation == "*":return {"result": a * b}elif operation == "/":if b == 0:raise MCPError(-32602, "除數不能為零")return {"result": a / b}else:raise MCPError(-32602, f"不支持的操作: {operation}")self.mcp_server.tools["calculator/compute"] = calculator# 測試各種運算test_cases = [({"a": 10, "b": 5, "operation": "+"}, 15),({"a": 10, "b": 5, "operation": "-"}, 5),({"a": 10, "b": 5, "operation": "*"}, 50),({"a": 10, "b": 5, "operation": "/"}, 2),]for args, expected in test_cases:result = await self.mcp_client.call_tool("calculator/compute", args)self.assertEqual(result["result"], expected)async def test_error_handling(self):"""測試錯誤處理"""# 注冊會拋出錯誤的工具async def error_tool(arguments):raise MCPError(-32602, "測試錯誤")self.mcp_server.tools["test/error"] = error_tool# 測試錯誤處理with self.assertRaises(MCPError) as context:await self.mcp_client.call_tool("test/error", {})self.assertEqual(context.exception.code, -32602)self.assertEqual(context.exception.message, "測試錯誤")# 性能測試
class MCPPerformanceTest:def __init__(self):self.mcp_client = MCPClient("http://localhost:3000")async def benchmark_tool_calls(self, tool_name: str, arguments: Dict[str, Any], iterations: int = 100):"""基準測試工具調用性能"""import timestart_time = time.time()# 并發執行多次調用tasks = [self.mcp_client.call_tool(tool_name, arguments)for _ in range(iterations)]results = await asyncio.gather(*tasks)end_time = time.time()total_time = end_time - start_timeavg_time = total_time / iterationsreturn {"total_time": total_time,"iterations": iterations,"average_time": avg_time,"throughput": iterations / total_time,"success_rate": len([r for r in results if "error" not in r]) / len(results)}async def stress_test(self, concurrent_requests: int = 50):"""壓力測試"""async def make_request(request_id: int):try:result = await self.mcp_client.call_tool("test/stress", {"id": request_id})return {"id": request_id, "status": "success", "result": result}except Exception as e:return {"id": request_id, "status": "error", "error": str(e)}tasks = [make_request(i) for i in range(concurrent_requests)]results = await asyncio.gather(*tasks)success_count = len([r for r in results if r["status"] == "success"])error_count = len([r for r in results if r["status"] == "error"])return {"total_requests": concurrent_requests,"successful_requests": success_count,"failed_requests": error_count,"success_rate": success_count / concurrent_requests}# 使用示例
async def run_tests():"""運行所有測試"""# 單元測試test_case = MCPTestCase()await test_case.test_tool_registration()await test_case.test_tool_execution()await test_case.test_error_handling()# 性能測試perf_test = MCPPerformanceTest()benchmark_result = await perf_test.benchmark_tool_calls("calculator/compute", {"a": 10, "b": 5, "operation": "+"}, 100)print(f"性能測試結果: {benchmark_result}")stress_result = await perf_test.stress_test(50)print(f"壓力測試結果: {stress_result}")if __name__ == "__main__":asyncio.run(run_tests())

通過以上完整的MCP開發指南，你可以看到MCP協議不僅是一個技術標準，更是一個完整的生態系統。它通過標準化的接口、靈活的架構設計和豐富的工具生態，為AI應用開發提供了強大的基礎設施支持。

無論是簡單的API調用，還是復雜的業務邏輯處理，MCP都能提供統一、安全、高效的解決方案。隨著AI技術的不斷發展，MCP協議將繼續演進，為構建更智能、更強大的AI應用生態系統貢獻力量。