🎬 HoRain 云小助手：个人主页
⛺️生活的理想，就是为了理想的生活!
⛳️ 推荐
前些天发现了一个超棒的服务器购买网站，性价比超高，大内存超划算！忍不住分享一下给大家。点击跳转到网站。
目录
⛳️ 推荐
Python 连接 MCP Server 全指南
📊 MCP 核心概念速览
🚀 一、环境准备与安装
1.1 安装 Python SDK
1.2 验证安装
🔌 二、连接 MCP Server 的三种方式
2.1 方式一：使用官方 Python SDK（推荐）
2.2 方式二：通过 HTTP 连接远程 MCP Server
2.3 方式三：使用 FastMCP 简化连接
🛠️ 三、完整实战示例
3.1 创建简单的 MCP Server
3.2 创建功能完整的 MCP Client
3.3 集成 LLM 的智能客户端
🌐 四、连接第三方 MCP Server
4.1 连接天聚数行 MCP 服务
4.2 连接观测云 MCP Server
🔧 五、高级配置与最佳实践
5.1 配置管理
5.2 错误处理与重试
5.3 性能监控
🚨 六、常见问题与解决方案
6.1 连接问题排查
6.2 常见错误及解决方案
📚 七、学习资源与进阶
7.1 官方资源
7.2 社区资源
7.3 进阶主题
🎯 总结
Python 连接 MCP Server 全指南
MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 推出的开放协议，旨在标准化 AI 模型与外部数据源、工具之间的交互方式。通过 MCP，你可以让大语言模型安全、可控地调用外部工具，就像给 AI 装上了"手脚"。
📊 MCP 核心概念速览
组件 | 作用 | 类比 |
|---|---|---|
MCP Server | 提供工具和数据源 | 服务提供方 |
MCP Client | 连接服务器并调用工具 | 服务消费方 |
Tools | 服务器暴露的可调用函数 | API接口 |
Resources | 只读数据源（文件、数据库等） | 数据源 |
Transport | 通信方式（stdio/HTTP/SSE） | 传输协议 |
🚀 一、环境准备与安装
1.1 安装 Python SDK
# 安装官方 MCP Python SDK pip install mcp # 或使用 FastMCP（简化版） pip install fastmcp # 使用 uv 包管理器（推荐） uv add mcp uv add httpx pydantic1.2 验证安装
import mcp print(f"MCP SDK 版本: {mcp.__version__}")🔌 二、连接 MCP Server 的三种方式
2.1 方式一：使用官方 Python SDK（推荐）
import asyncio from mcp import ClientSession from mcp.client.stdio import stdio_client, StdioServerParameters async def connect_via_stdio(): """通过 stdio 连接本地 MCP Server""" # 配置服务器参数 server_params = StdioServerParameters( command="python", # 运行命令 args=["server.py"], # 服务器脚本路径 env=None # 可选环境变量 ) # 建立连接 async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 获取可用工具列表 tools_response = await session.list_tools() print(f"可用工具: {[tool.name for tool in tools_response.tools]}") # 调用工具 result = await session.call_tool("add", {"a": 3, "b": 5}) print(f"3 + 5 = {result}") return session # 运行 asyncio.run(connect_via_stdio())2.2 方式二：通过 HTTP 连接远程 MCP Server
import httpx import asyncio from mcp.client.streamable_http import streamable_http_client from mcp.client.session import ClientSession async def connect_via_http(): """通过 HTTP 连接远程 MCP Server""" url = "https://mcp.example.com/mcp" headers = {"Authorization": "Bearer your_token"} async with httpx.AsyncClient(headers=headers, timeout=60.0) as http_client: async with streamable_http_client(url=url, http_client=http_client) as (read, write, _): async with ClientSession(read, write) as session: # 初始化会话 init_result = await session.initialize() print(f"已连接到: {init_result.serverInfo.name} v{init_result.serverInfo.version}") # 获取工具列表 tools = await session.list_tools() print(f"可用工具数量: {len(tools.tools)}") return session # 运行 asyncio.run(connect_via_http())2.3 方式三：使用 FastMCP 简化连接
import asyncio from fastmcp import Client async def connect_with_fastmcp(): """使用 FastMCP 简化连接""" # 连接到本地服务器 client = Client("python server.py") async with client: # 获取工具列表 tools = await client.list_tools() print("可用工具:", tools) # 调用工具 result = await client.call_tool('greet', {'name': '技术爱好者'}) print("调用结果:", result) # 调用资源 resource = await client.read_resource("greeting://Alice") print("资源内容:", resource) # 运行 asyncio.run(connect_with_fastmcp())🛠️ 三、完整实战示例
3.1 创建简单的 MCP Server
# server.py - 简单的数学计算服务器 from mcp.server.fastmcp import FastMCP # 创建 MCP 服务器实例 mcp = FastMCP("MathServer") @mcp.tool() def add(a: int, b: int) -> int: """两数相加""" return a + b @mcp.tool() def subtract(a: int, b: int) -> int: """两数相减""" return a - b @mcp.tool() def multiply(a: int, b: int) -> int: """两数相乘""" return a * b @mcp.tool() def divide(a: float, b: float) -> float: """两数相除""" if b == 0: raise ValueError("除数不能为零") return a / b @mcp.resource("math://constants/{name}") def get_constant(name: str) -> str: """获取数学常数""" constants = { "pi": "3.141592653589793", "e": "2.718281828459045", "golden_ratio": "1.618033988749895" } return constants.get(name, "未知常数") if __name__ == "__main__": # 启动服务器（支持多种传输方式） mcp.run() # 默认 stdio 模式 # mcp.run(transport="streamable-http", port=8000) # HTTP 模式 # mcp.run(transport="sse", port=8000) # SSE 模式3.2 创建功能完整的 MCP Client
# client.py - 功能完整的客户端 import asyncio import sys from typing import Dict, Any from mcp import ClientSession, types from mcp.client.stdio import stdio_client, StdioServerParameters class MCPClient: """MCP 客户端封装类""" def __init__(self, server_script: str): self.server_script = server_script self.session = None async def connect(self): """连接到 MCP 服务器""" server_params = StdioServerParameters( command=sys.executable, args=[self.server_script], ) transport = stdio_client(server_params) stdio, write = await transport.__aenter__() self.session = await ClientSession(stdio, write).__aenter__() await self.session.initialize() print("✅ 已成功连接到 MCP 服务器") return self async def list_tools(self) -> list: """获取所有可用工具""" if not self.session: raise RuntimeError("请先调用 connect() 方法连接服务器") response = await self.session.list_tools() return [ { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema } for tool in response.tools ] async def call_tool(self, tool_name: str, arguments: Dict[str, Any]): """调用指定工具""" if not self.session: raise RuntimeError("请先调用 connect() 方法连接服务器") try: result = await self.session.call_tool(tool_name, arguments) return result except Exception as e: print(f"❌ 调用工具 {tool_name} 失败: {e}") return None async def read_resource(self, uri: str): """读取资源""" if not self.session: raise RuntimeError("请先调用 connect() 方法连接服务器") try: response = await self.session.read_resource([types.Resource(uri=uri)]) if response.contents: return response.contents[0].text except Exception as e: print(f"❌ 读取资源 {uri} 失败: {e}") return None async def close(self): """关闭连接""" if self.session: await self.session.__aexit__(None, None, None) print("🔌 连接已关闭") async def main(): """主函数示例""" # 创建客户端 client = MCPClient("server.py") try: # 连接服务器 await client.connect() # 获取工具列表 tools = await client.list_tools() print("\n📋 可用工具列表:") for tool in tools: print(f" • {tool['name']}: {tool['description']}") # 调用工具示例 print("\n🧪 工具调用测试:") # 测试加法 result = await client.call_tool("add", {"a": 10, "b": 20}) print(f" 10 + 20 = {result}") # 测试除法 result = await client.call_tool("divide", {"a": 100, "b": 4}) print(f" 100 ÷ 4 = {result}") # 读取资源示例 print("\n📚 资源读取测试:") pi_value = await client.read_resource("math://constants/pi") print(f" π 的值: {pi_value}") e_value = await client.read_resource("math://constants/e") print(f" e 的值: {e_value}") except Exception as e: print(f"❌ 发生错误: {e}") finally: # 关闭连接 await client.close() if __name__ == "__main__": asyncio.run(main())3.3 集成 LLM 的智能客户端
# smart_client.py - 集成 LLM 的智能客户端 import asyncio import os from typing import List, Dict, Any from dotenv import load_dotenv from openai import OpenAI from mcp import ClientSession from mcp.client.stdio import stdio_client, StdioServerParameters load_dotenv() class SmartMCPClient: """集成 LLM 的智能 MCP 客户端""" def __init__(self, server_script: str): self.server_script = server_script self.session = None self.llm_client = None self.tools = [] async def connect(self): """连接到 MCP 服务器和 LLM""" # 1. 连接 MCP 服务器 server_params = StdioServerParameters( command="python", args=[self.server_script], ) transport = stdio_client(server_params) stdio, write = await transport.__aenter__() self.session = await ClientSession(stdio, write).__aenter__() await self.session.initialize() # 2. 获取工具列表并转换为 LLM 格式 tools_response = await self.session.list_tools() self.tools = [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema } } for tool in tools_response.tools ] # 3. 初始化 LLM 客户端 self.llm_client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("BASE_URL", "https://api.openai.com/v1"), ) print(f"✅ 已连接到 MCP 服务器，发现 {len(self.tools)} 个工具") return self async def process_query(self, query: str) -> str: """处理用户查询，自动调用工具""" if not self.session or not self.llm_client: raise RuntimeError("请先调用 connect() 方法") messages = [ { "role": "system", "content": "你是一个智能助手，可以调用数学计算工具。请根据用户需求选择合适的工具进行计算。" }, {"role": "user", "content": query} ] # 第一次 LLM 调用 response = self.llm_client.chat.completions.create( model="gpt-4", messages=messages, tools=self.tools, tool_choice="auto" ) message = response.choices[0].message # 检查是否需要调用工具 if message.tool_calls: for tool_call in message.tool_calls: tool_name = tool_call.function.name import json arguments = json.loads(tool_call.function.arguments) print(f"🛠️ 调用工具: {tool_name}({arguments})") # 调用 MCP 工具 tool_result = await self.session.call_tool(tool_name, arguments) # 将结果发送回 LLM messages.append(message) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(tool_result) }) # 第二次 LLM 调用（包含工具结果） second_response = self.llm_client.chat.completions.create( model="gpt-4", messages=messages ) return second_response.choices[0].message.content else: return message.content async def interactive_chat(self): """交互式聊天""" print("\n🤖 智能数学助手已启动") print("输入 '退出' 或 'exit' 结束对话") print("-" * 50) while True: try: user_input = input("\n你: ").strip() if user_input.lower() in ['退出', 'exit', 'quit']: break if not user_input: continue print("思考中...", end="", flush=True) response = await self.process_query(user_input) print(f"\n助手: {response}") except KeyboardInterrupt: print("\n\n👋 再见！") break except Exception as e: print(f"\n❌ 错误: {e}") async def close(self): """关闭连接""" if self.session: await self.session.__aexit__(None, None, None) print("🔌 连接已关闭") async def main(): client = SmartMCPClient("server.py") try: await client.connect() await client.interactive_chat() finally: await client.close() if __name__ == "__main__": asyncio.run(main())🌐 四、连接第三方 MCP Server
4.1 连接天聚数行 MCP 服务
import requests import json class TianAPI_MCP_Client: """天聚数行 MCP 服务客户端""" def __init__(self, api_key: str, toolset_name: str = "my-agent-tools"): self.base_url = f"https://mcp.tianapi.com/{toolset_name}/{api_key}" def list_tools(self): """获取可用工具列表""" response = requests.post(self.base_url, json={"method": "list_tools"}) return response.json() def call_tool(self, tool_name: str, arguments: dict): """调用工具""" payload = { "method": "call", "params": { "name": tool_name, "arguments": arguments } } response = requests.post(self.base_url, json=payload) return response.json() def get_weather(self, city: str): """查询天气（示例）""" return self.call_tool("tianqi", {"city": city}) def get_vehicle_limit(self, city: str): """查询限行（示例）""" return self.call_tool("vehiclelimit", {"city": city}) # 使用示例 def tianapi_example(): client = TianAPI_MCP_Client(api_key="your_api_key_here") # 获取工具列表 tools = client.list_tools() print("可用工具:", [tool["name"] for tool in tools.get("result", [])]) # 查询天气 weather = client.get_weather("北京") if not weather.get("error"): data = weather["result"] print(f"{data['city']} 天气:") print(f" 温度: {data['temp']}℃") print(f" 天气: {data['weather']}") print(f" 风力: {data['wind']}") print(f" 湿度: {data['humidity']}%") # 查询限行 limit = client.get_vehicle_limit("成都") if not limit.get("error"): data = limit["result"] print(f"\n{data['city']} 限行信息:") print(f" 限行: {data['limit_info']}") print(f" 区域: {data['area']}") if __name__ == "__main__": tianapi_example()4.2 连接观测云 MCP Server
import httpx import asyncio from mcp.client.streamable_http import streamable_http_client from mcp.client.session import ClientSession async def connect_guance_mcp(): """连接观测云 MCP Server""" url = "https://obsy-ai.guance.com/obsy_ai_mcp/mcp" headers = { "Authorization": "DF-API-KEY your_api_key_here", "Endpoint": "cn1" # 根据地区选择 } async with httpx.AsyncClient(headers=headers, timeout=30.0) as http_client: async with streamable_http_client(url=url, http_client=http_client) as (read, write, _): async with ClientSession(read, write) as session: # 初始化连接 await session.initialize() # 获取工具列表 tools_response = await session.list_tools() print("观测云可用工具:") for tool in tools_response.tools: print(f" • {tool.name}: {tool.description}") return session # 运行 asyncio.run(connect_guance_mcp())🔧 五、高级配置与最佳实践
5.1 配置管理
# config.py - 配置管理 import os from typing import Optional from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() @dataclass class MCPConfig: """MCP 配置类""" # 传输方式 transport: str = os.getenv("MCP_TRANSPORT", "stdio") # stdio, http, sse # HTTP 配置 http_url: Optional[str] = os.getenv("MCP_HTTP_URL") http_headers: dict = None # Stdio 配置 server_command: str = os.getenv("MCP_SERVER_COMMAND", "python") server_args: list = None # 超时设置 timeout: int = int(os.getenv("MCP_TIMEOUT", "30")) # 重试配置 max_retries: int = int(os.getenv("MCP_MAX_RETRIES", "3")) retry_delay: float = float(os.getenv("MCP_RETRY_DELAY", "1.0")) def __post_init__(self): if self.http_headers is None: self.http_headers = {} if self.server_args is None: self.server_args = ["server.py"] # 添加认证头 if api_key := os.getenv("MCP_API_KEY"): self.http_headers["Authorization"] = f"Bearer {api_key}" class MCPClientFactory: """MCP 客户端工厂""" @staticmethod def create_client(config: MCPConfig): """根据配置创建客户端""" if config.transport == "stdio": return StdioMCPClient(config) elif config.transport == "http": return HttpMCPClient(config) elif config.transport == "sse": return SSEMCPClient(config) else: raise ValueError(f"不支持的传输方式: {config.transport}") # 使用示例 config = MCPConfig( transport="http", http_url="https://api.example.com/mcp", http_headers={"X-API-Key": "your_key"} ) client = MCPClientFactory.create_client(config)5.2 错误处理与重试
import asyncio import time from typing import Callable, Any from functools import wraps def retry_with_backoff( max_retries: int = 3, initial_delay: float = 1.0, max_delay: float = 10.0, exponential_base: float = 2.0 ): """带指数退避的重试装饰器""" def decorator(func: Callable): @wraps(func) async def wrapper(*args, **kwargs): delay = initial_delay last_exception = None for attempt in range(max_retries + 1): try: return await func(*args, **kwargs) except Exception as e: last_exception = e if attempt == max_retries: break # 计算下一次重试的延迟 delay = min(delay * exponential_base, max_delay) jitter = delay * 0.1 # 添加随机抖动 actual_delay = delay + (jitter * (2 * random.random() - 1)) print(f"⚠️ 调用失败，{actual_delay:.1f}秒后重试 (尝试 {attempt + 1}/{max_retries})") await asyncio.sleep(actual_delay) raise last_exception return wrapper return decorator class RobustMCPClient: """健壮的 MCP 客户端""" def __init__(self, config: MCPConfig): self.config = config self.session = None @retry_with_backoff(max_retries=3, initial_delay=1.0) async def connect_with_retry(self): """带重试的连接""" try: if self.config.transport == "stdio": from mcp.client.stdio import stdio_client, StdioServerParameters server_params = StdioServerParameters( command=self.config.server_command, args=self.config.server_args, ) transport = stdio_client(server_params) stdio, write = await transport.__aenter__() self.session = await ClientSession(stdio, write).__aenter__() elif self.config.transport == "http": import httpx from mcp.client.streamable_http import streamable_http_client async with httpx.AsyncClient( headers=self.config.http_headers, timeout=self.config.timeout ) as http_client: async with streamable_http_client( url=self.config.http_url, http_client=http_client ) as (read, write, _): self.session = await ClientSession(read, write).__aenter__() # 初始化会话 await self.session.initialize() print("✅ 连接成功") return self except Exception as e: print(f"❌ 连接失败: {e}") raise async def safe_call_tool(self, tool_name: str, arguments: dict, fallback_value=None): """安全的工具调用""" try: result = await self.session.call_tool(tool_name, arguments) return result except Exception as e: print(f"⚠️ 工具调用失败: {e}") # 记录错误日志 self.log_error({ "tool": tool_name, "arguments": arguments, "error": str(e), "timestamp": time.time() }) # 返回降级值 if fallback_value is not None: return fallback_value raise def log_error(self, error_info: dict): """记录错误日志""" # 这里可以实现日志记录到文件、数据库等 import json print(f"📝 错误日志: {json.dumps(error_info, ensure_ascii=False)}")5.3 性能监控
import time from typing import Dict, List from dataclasses import dataclass from datetime import datetime @dataclass class ToolCallMetrics: """工具调用指标""" tool_name: str start_time: float end_time: float success: bool error_message: str = "" @property def duration(self) -> float: return self.end_time - self.start_time class MCPClientWithMetrics: """带性能监控的 MCP 客户端""" def __init__(self, base_client): self.base_client = base_client self.metrics: List[ToolCallMetrics] = [] async def call_tool_with_metrics(self, tool_name: str, arguments: dict): """带监控的工具调用""" start_time = time.time() success = False error_msg = "" try: result = await self.base_client.call_tool(tool_name, arguments) success = True return result except Exception as e: error_msg = str(e) raise finally: end_time = time.time() metric = ToolCallMetrics( tool_name=tool_name, start_time=start_time, end_time=end_time, success=success, error_message=error_msg ) self.metrics.append(metric) def get_performance_report(self) -> Dict: """获取性能报告""" if not self.metrics: return {} total_calls = len(self.metrics) successful_calls = sum(1 for m in self.metrics if m.success) failed_calls = total_calls - successful_calls durations = [m.duration for m in self.metrics] avg_duration = sum(durations) / len(durations) if durations else 0 max_duration = max(durations) if durations else 0 min_duration = min(durations) if durations else 0 # 按工具统计 tool_stats = {} for metric in self.metrics: if metric.tool_name not in tool_stats: tool_stats[metric.tool_name] = { "total": 0, "success": 0, "fail": 0, "total_duration": 0 } stats = tool_stats[metric.tool_name] stats["total"] += 1 stats["total_duration"] += metric.duration if metric.success: stats["success"] += 1 else: stats["fail"] += 1 # 计算平均时长 for tool_name, stats in tool_stats.items(): if stats["total"] > 0: stats["avg_duration"] = stats["total_duration"] / stats["total"] return { "summary": { "total_calls": total_calls, "successful_calls": successful_calls, "failed_calls": failed_calls, "success_rate": successful_calls / total_calls if total_calls > 0 else 0, "avg_duration_seconds": avg_duration, "max_duration_seconds": max_duration, "min_duration_seconds": min_duration }, "tool_statistics": tool_stats, "timestamp": datetime.now().isoformat() } def print_performance_report(self): """打印性能报告""" report = self.get_performance_report() if not report: print("📊 暂无性能数据") return summary = report["summary"] print("\n" + "="*50) print("📊 MCP 客户端性能报告") print("="*50) print(f"总调用次数: {summary['total_calls']}") print(f"成功次数: {summary['successful_calls']}") print(f"失败次数: {summary['failed_calls']}") print(f"成功率: {summary['success_rate']:.2%}") print(f"平均耗时: {summary['avg_duration_seconds']:.3f}秒") print(f"最大耗时: {summary['max_duration_seconds']:.3f}秒") print(f"最小耗时: {summary['min_duration_seconds']:.3f}秒") print("\n🔧 工具统计:") for tool_name, stats in report["tool_statistics"].items(): print(f" {tool_name}:") print(f" 调用次数: {stats['total']}") print(f" 成功率: {stats['success']}/{stats['total']} ({stats['success']/stats['total']:.2%})") print(f" 平均耗时: {stats.get('avg_duration', 0):.3f}秒") print("="*50)🚨 六、常见问题与解决方案
6.1 连接问题排查
# troubleshooting.py - 连接问题排查工具 import asyncio import sys import subprocess from typing import Optional class MCPTroubleshooter: """MCP 连接问题排查工具""" @staticmethod async def check_connection(config: dict) -> dict: """检查连接状态""" results = { "server_script_exists": False, "python_available": False, "dependencies_installed": False, "server_startup": False, "connection_test": False, "errors": [] } # 1. 检查服务器脚本是否存在 server_script = config.get("server_args", [""])[0] if server_script and os.path.exists(server_script): results["server_script_exists"] = True else: results["errors"].append(f"服务器脚本不存在: {server_script}") # 2. 检查 Python 是否可用 try: subprocess.run([sys.executable, "--version"], capture_output=True, check=True) results["python_available"] = True except: results["errors"].append("Python 不可用") # 3. 检查依赖是否安装 try: import mcp results["dependencies_installed"] = True except ImportError as e: results["errors"].append(f"依赖未安装: {e}") # 4. 测试服务器启动 if results["server_script_exists"] and results["python_available"]: try: process = subprocess.Popen( [sys.executable, server_script], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 等待一小段时间看是否启动成功 import time time.sleep(2) if process.poll() is None: results["server_startup"] = True process.terminate() else: _, stderr = process.communicate() results["errors"].append(f"服务器启动失败: {stderr}") except Exception as e: results["errors"].append(f"启动测试失败: {e}") return results @staticmethod def print_diagnostic_report(results: dict): """打印诊断报告""" print("\n" + "="*60) print("🔍 MCP 连接诊断报告") print("="*60) checks = [ ("服务器脚本", results["server_script_exists"]), ("Python 环境", results["python_available"]), ("依赖安装", results["dependencies_installed"]), ("服务器启动", results["server_startup"]), ("连接测试", results["connection_test"]) ] for check_name, status in checks: status_icon = "✅" if status else "❌" print(f"{status_icon} {check_name}") if results["errors"]: print("\n⚠️ 发现的问题:") for error in results["errors"]: print(f" • {error}") print("\n💡 建议解决方案:") for error in results["errors"]: if "脚本不存在" in error: print(" • 检查 server.py 文件路径是否正确") elif "Python 不可用" in error: print(" • 确保 Python 已正确安装并添加到 PATH") elif "依赖未安装" in error: print(" • 运行: pip install mcp") elif "服务器启动失败" in error: print(" • 检查服务器脚本是否有语法错误") print(" • 查看服务器日志获取详细信息") print("="*60) # 使用示例 async def diagnose_connection(): config = { "server_command": "python", "server_args": ["server.py"] } troubleshooter = MCPTroubleshooter() results = await troubleshooter.check_connection(config) troubleshooter.print_diagnostic_report(results) if __name__ == "__main__": asyncio.run(diagnose_connection())6.2 常见错误及解决方案
错误类型 | 可能原因 | 解决方案 |
|---|---|---|
Connection refused | 服务器未启动或端口被占用 | 检查服务器是否运行，更换端口 |
ModuleNotFoundError | 依赖未安装 |
|
TimeoutError | 网络延迟或服务器响应慢 | 增加超时时间，检查网络连接 |
JSON-RPC error | 协议版本不匹配 | 检查 MCP SDK 版本，更新到最新 |
Tool not found | 工具名称错误或未注册 | 使用 |
Invalid arguments | 参数类型或格式错误 | 检查工具的参数定义和要求 |
📚 七、学习资源与进阶
7.1 官方资源
MCP 官方文档: modelcontextprotocol.io
Python SDK GitHub: github.com/modelcontextprotocol/python-sdk
MCP 规范: spec.modelcontextprotocol.io
7.2 社区资源
Awesome MCP Servers: GitHub 上的 MCP 服务器集合
MCP Inspector: 调试和测试 MCP 服务器的工具
Claude Desktop: 内置 MCP 支持的 AI 客户端
7.3 进阶主题
自定义传输协议: 实现自己的 Transport 类
工具权限管理: 基于角色的访问控制
流式响应: 支持大型数据集的流式传输
批量操作: 优化多个工具调用的性能
缓存策略: 减少重复请求，提高性能
🎯 总结
Python 连接 MCP Server 的核心要点：
选择适合的连接方式：根据场景选择 stdio（本地）、HTTP（远程）或 SSE（实时）
使用正确的 SDK：官方
mcp库功能最全，fastmcp更易上手遵循标准流程：连接 → 初始化 → 发现工具 → 调用工具
做好错误处理：网络异常、工具调用失败、超时等都需要处理
考虑性能优化：连接池、缓存、批量操作等
通过 MCP，你可以将任何 Python 功能暴露给 AI 模型，构建强大的 AI 应用生态系统。无论是简单的工具调用，还是复杂的业务流程集成，MCP 都提供了标准化的解决方案。
❤️❤️❤️本人水平有限，如有纰漏，欢迎各位大佬评论批评指正！😄😄😄
💘💘💘如果觉得这篇文对你有帮助的话，也请给个点赞、收藏下吧，非常感谢!👍 👍 👍
🔥🔥🔥Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧！🌙🌙🌙
