# MCP客户端极简教程：快速接入大模型API的完整指南
MCP（Model Context Protocol）客户端是连接大模型与外部工具的关键桥梁。本文将手把手教你从零开始构建MCP客户端，实现与大模型API的无缝对接。
## MCP客户端基础架构
理解MCP协议的核心概念是构建客户端的第一步。
```python
# mcp_client/core.py
import json
import asyncio
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
import aiohttp
@dataclass
class MCPRequest:
"""MCP请求数据结构"""
jsonrpc: str = "2.0"
id: Optional[str] = None
method: str = ""
params: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict[str, Any]:
"""转换为字典格式"""
<"www.suzhou.gov.cn.felli.cn">
<"www.changsha.gov.cn.felli.cn">
<"www.baoding.gov.cn.felli.cn">
data = {
"jsonrpc": self.jsonrpc,
"method": self.method
}
if self.id:
data["id"] = self.id
if self.params:
data["params"] = self.params
return data
@dataclass
class MCPResponse:
"""MCP响应数据结构"""
jsonrpc: str
id: Optional[str]
result: Optional[Dict[str, Any]] = None
error: Optional[Dict[str, Any]] = None
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'MCPResponse':
"""从字典创建响应对象"""
return cls(
jsonrpc=data.get("jsonrpc", "2.0"),
id=data.get("id"),
result=data.get("result"),
error=data.get("error")
)
def is_success(self) -> bool:
"""判断请求是否成功"""
return self.error is None
class MCPClient:
"""MCP客户端核心类"""
def __init__(self, server_url: str):
self.server_url = server_url
self.session: Optional[aiohttp.ClientSession] = None
self.request_id = 0
async def connect(self):
"""连接到MCP服务器"""
self.session = aiohttp.ClientSession()
async def close(self):
"""关闭连接"""
<"www.handan.gov.cn.felli.cn">
<"www.wuxi.gov.cn.felli.cn">
<"www.qinhuangdao.gov.cn.felli.cn">
if self.session:
await self.session.close()
async def send_request(self, method: str, params: Dict[str, Any] = None) -> MCPResponse:
"""发送MCP请求"""
if not self.session:
raise RuntimeError("客户端未连接，请先调用connect()方法")
self.request_id += 1
request = MCPRequest(
id=str(self.request_id),
method=method,
params=params or {}
)
try:
async with self.session.post(
self.server_url,
json=request.to_dict(),
headers={"Content-Type": "application/json"}
) as response:
if response.status == 200:
data = await response.json()
return MCPResponse.from_dict(data)
else:
return MCPResponse(
jsonrpc="2.0",
id=request.id,
error={
"code": -32000,
"message": f"HTTP错误: {response.status}"
}
)
except Exception as e:
return MCPResponse(
jsonrpc="2.0",
id=request.id,
error={
"code": -32603,
"message": f"请求失败: {str(e)}"
}
)
```
## 基础功能实现
实现MCP协议的核心方法，建立与服务器的完整通信。
```python
<"www.dongguan.gov.cn.felli.cn">
<"www.cangzhou.gov.cn.felli.cn">
<"www.langfang.gov.cn.felli.cn">
# mcp_client/basic_operations.py
from .core import MCPClient, MCPResponse
from typing import List, Dict, Any
class BasicMCPClient(MCPClient):
"""基础MCP客户端功能"""
async def initialize(self, client_name: str = "Python MCP Client") -> MCPResponse:
"""初始化连接"""
params = {
"protocolVersion": "2024-11-07",
"capabilities": {
"roots": {"listChanged": True},
"tools": {"listChanged": True}
},
"clientInfo": {
"name": client_name,
"version": "1.0.0"
}
}
return await self.send_request("initialize", params)
async def list_tools(self) -> List[Dict[str, Any]]:
"""获取可用工具列表"""
response = await self.send_request("tools/list")
if response.is_success() and response.result:
return response.result.get("tools", [])
return []
async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""调用工具"""
params = {
"name": tool_name,
"arguments": arguments
}
response = await self.send_request("tools/call", params)
if response.is_success() and response.result:
return response.result.get("content")
else:
error_msg = response.error.get("message", "未知错误") if response.error else "未知错误"
raise Exception(f"工具调用失败: {error_msg}")
async def list_resources(self) -> List[Dict[str, Any]]:
<"www.shenyang.gov.cn.felli.cn">
<"www.jinzhou.gov.cn.felli.cn">
<"www.yingkou.gov.cn.felli.cn">
"""获取资源列表"""
response = await self.send_request("resources/list")
if response.is_success() and response.result:
return response.result.get("resources", [])
return []
async def read_resource(self, resource_uri: str) -> Any:
"""读取资源"""
params = {
"uri": resource_uri
}
response = await self.send_request("resources/read", params)
if response.is_success() and response.result:
return response.result.get("contents")
else:
error_msg = response.error.get("message", "未知错误") if response.error else "未知错误"
raise Exception(f"资源读取失败: {error_msg}")
```
## 工具调用封装
提供更友好的工具调用接口，简化开发流程。
```python
# mcp_client/tool_wrappers.py
from typing import Dict, Any, List
<"www.changchun.gov.cn.felli.cn">
<"www.haerbin.gov.cn.felli.cn">
<"www.kunming.gov.cn.felli.cn">
import json
class ToolManager:
"""工具管理器"""
def __init__(self, client: BasicMCPClient):
self.client = client
self.available_tools: List[Dict[str, Any]] = []
async def refresh_tools(self):
"""刷新工具列表"""
self.available_tools = await self.client.list_tools()
def get_tool_info(self, tool_name: str) -> Dict[str, Any]:
"""获取工具信息"""
for tool in self.available_tools:
if tool.get("name") == tool_name:
return tool
raise ValueError(f"工具未找到: {tool_name}")
async def execute_tool(self, tool_name: str, **kwargs) -> Any:
"""执行工具"""
return await self.client.call_tool(tool_name, kwargs)
def list_tool_names(self) -> List[str]:
"""获取所有工具名称"""
return [tool.get("name") for tool in self.available_tools]
def describe_tool(self, tool_name: str) -> str:
"""获取工具描述"""
tool_info = self.get_tool_info(tool_name)
description = tool_info.get("description", "无描述")
parameters = tool_info.get("inputSchema", {}).get("properties", {})
param_descriptions = []
for param_name, param_info in parameters.items():
param_type = param_info.get("type", "unknown")
param_desc = param_info.get("description", "无描述")
param_descriptions.append(f" - {param_name} ({param_type}): {param_desc}")
params_text = "\n".join(param_descriptions) if param_descriptions else " 无参数"
return f"{description}\n参数:\n{params_text}"
# 预定义工具包装器
<"www.dalian.gov.cn.felli.cn">
<"www.quanzhou.gov.cn.felli.cn">
<"www.shijiazhuang.gov.cn.felli.cn">
class CommonTools:
"""常用工具包装器"""
def __init__(self, tool_manager: ToolManager):
self.tm = tool_manager
async def web_search(self, query: str, max_results: int = 5) -> str:
"""网络搜索"""
return await self.tm.execute_tool("web_search", query=query, max_results=max_results)
async def calculator(self, expression: str) -> float:
"""计算器"""
return await self.tm.execute_tool("calculator", expression=expression)
async def file_read(self, filepath: str) -> str:
"""读取文件"""
return await self.tm.execute_tool("file_read", filepath=filepath)
async def file_write(self, filepath: str, content: str) -> str:
"""写入文件"""
return await self.tm.execute_tool("file_write", filepath=filepath, content=content)
async def get_weather(self, location: str) -> Dict[str, Any]:
"""获取天气"""
return await self.tm.execute_tool("get_weather", location=location)
```
## 完整客户端实现
整合所有功能，提供开箱即用的完整客户端。
```python
# mcp_client/complete_client.py
import asyncio
from typing import Dict, Any, List
from .basic_operations import BasicMCPClient
from .tool_wrappers import ToolManager, CommonTools
class CompleteMCPClient:
"""完整的MCP客户端"""
def __init__(self, server_url: str):
self.base_client = BasicMCPClient(server_url)
self.tool_manager = ToolManager(self.base_client)
self.common_tools = CommonTools(self.tool_manager)
self.is_connected = False
async def connect(self, client_name: str = "MCP Client"):
"""连接到服务器"""
await self.base_client.connect()
# 初始化连接
init_result = await self.base_client.initialize(client_name)
if not init_result.is_success():
await self.base_client.close()
raise ConnectionError(f"初始化失败: {init_result.error}")
# 刷新工具列表
await self.tool_manager.refresh_tools()
self.is_connected = True
print(f"✅ 成功连接到MCP服务器")
print(f"? 可用工具: {len(self.tool_manager.available_tools)}个")
async def close(self):
"""关闭连接"""
await self.base_client.close()
self.is_connected = False
<"www.nanning.gov.cn.felli.cn">
<"www.panjin.gov.cn.felli.cn">
<"www.tieling.gov.cn.felli.cn">
print("? 连接已关闭")
async def list_tools(self) -> List[str]:
"""列出所有工具"""
if not self.is_connected:
raise RuntimeError("客户端未连接")
return self.tool_manager.list_tool_names()
async def get_tool_help(self, tool_name: str) -> str:
"""获取工具帮助信息"""
if not self.is_connected:
raise RuntimeError("客户端未连接")
return self.tool_manager.describe_tool(tool_name)
async def execute(self, tool_name: str, **kwargs) -> Any:
"""执行工具"""
if not self.is_connected:
raise RuntimeError("客户端未连接")
return await self.tool_manager.execute_tool(tool_name, **kwargs)
# 便捷方法
async def search(self, query: str, max_results: int = 5) -> str:
"""搜索（便捷方法）"""
return await self.common_tools.web_search(query, max_results)
async def calculate(self, expression: str) -> float:
"""计算（便捷方法）"""
return await self.common_tools.calculator(expression)
async def read_file(self, filepath: str) -> str:
"""读取文件（便捷方法）"""
return await self.common_tools.file_read(filepath)
async def write_file(self, filepath: str, content: str) -> str:
"""写入文件（便捷方法）"""
return await self.common_tools.file_write(filepath, content)
# 上下文管理器支持
class MCPClientSession:
"""MCP客户端会话（上下文管理器）"""
def __init__(self, server_url: str, client_name: str = "MCP Client"):
self.client = CompleteMCPClient(server_url)
self.client_name = client_name
async def __aenter__(self):
await self.client.connect(self.client_name)
return self.client
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.client.close()
<"www.taiyuan.gov.cn.felli.cn">
<"www.jilin.gov.cn.felli.cn">
<"www.siping.gov.cn.felli.cn">
```
## 使用示例和演示
通过具体示例展示客户端的使用方法。
```python
# examples/basic_usage.py
import asyncio
import sys
import os
# 添加客户端路径
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from mcp_client.complete_client import CompleteMCPClient, MCPClientSession
async def demo_basic_operations():
"""基础操作演示"""
print("? MCP客户端基础操作演示")
print("=" * 50)
# 方法1：使用上下文管理器（推荐）
async with MCPClientSession("http://localhost:8080/mcp") as client:
# 列出所有工具
tools = await client.list_tools()
print(f"? 可用工具: {tools}")
# 获取工具帮助
if tools:
help_text = await client.get_tool_help(tools[0])
print(f"? 工具帮助:\n{help_text}")
# 执行计算
try:
result = await client.calculate("(125 + 37) * 2")
print(f"? 计算结果: {result}")
except Exception as e:
print(f"❌ 计算失败: {e}")
print("演示完成！")
async def demo_advanced_usage():
"""高级用法演示"""
print("\n? MCP客户端高级用法演示")
print("=" * 50)
# 方法2：手动管理连接
client = CompleteMCPClient("http://localhost:8080/mcp")
try:
await client.connect("高级演示客户端")
# 批量执行多个工具
tasks = [
client.search("人工智能最新发展"),
client.calculate("3.14 * 25"),
# 添加更多任务...
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"❌ 任务 {i+1} 失败: {result}")
else:
print(f"✅ 任务 {i+1} 成功: {result}")
finally:
await client.close()
async def demo_custom_tools():
<"www.liaoyuan.gov.cn.felli.cn">
<"www.mudanjiang.gov.cn.felli.cn">
<"www.yantai.gov.cn.felli.cn">
"""自定义工具演示"""
print("\n? 自定义工具演示")
print("=" * 50)
async with MCPClientSession("http://localhost:8080/mcp") as client:
# 查看所有工具
tools = await client.list_tools()
print("可用工具:", tools)
# 尝试执行各种工具
for tool_name in tools:
try:
print(f"\n尝试执行工具: {tool_name}")
help_text = await client.get_tool_help(tool_name)
print(f"工具描述: {help_text}")
# 根据工具类型执行不同的参数
if "search" in tool_name.lower():
result = await client.execute(tool_name, query="Python编程")
print(f"搜索结果: {result}")
elif "calc" in tool_name.lower() or "math" in tool_name.lower():
result = await client.execute(tool_name, expression="2 + 3 * 4")
print(f"计算结果: {result}")
elif "file" in tool_name.lower() and "read" in tool_name.lower():
result = await client.execute(tool_name, filepath="example.txt")
print(f"文件内容: {result}")
except Exception as e:
print(f"工具 {tool_name} 执行失败: {e}")
# 交互式客户端
async def interactive_client():
"""交互式客户端"""
print("? MCP交互式客户端")
print("=" * 50)
print("输入 'help' 查看帮助, 'exit' 退出")
client = CompleteMCPClient("http://localhost:8080/mcp")
try:
await client.connect("交互式客户端")
while True:
try:
user_input = input("\n? 请输入命令: ").strip()
if user_input.lower() in ['exit', 'quit']:
break
elif user_input.lower() == 'help':
print_help()
elif user_input.lower() == 'list':
tools = await client.list_tools()
print("可用工具:", tools)
elif user_input.startswith('help '):
tool_name = user_input[5:].strip()
help_text = await client.get_tool_help(tool_name)
print(help_text)
elif user_input.startswith('search '):
query = user_input[7:].strip()
result = await client.search(query)
print(f"搜索结果: {result}")
elif user_input.startswith('calc '):
expression = user_input[5:].strip()
result = await client.calculate(expression)
print(f"计算结果: {result}")
else:
print("未知命令，输入 'help' 查看可用命令")
except KeyboardInterrupt:
break
except Exception as e:
print(f"错误: {e}")
finally:
await client.close()
def print_help():
"""打印帮助信息"""
help_text = """
可用命令:
help - 显示此帮助信息
list - 列出所有可用工具
help <工具名> - 显示特定工具的帮助
search <查询> - 执行搜索
calc <表达式> - 执行计算
exit - 退出程序
"""
print(help_text)
# 主函数
async def main():
"""主函数"""
if len(sys.argv) > 1:
if sys.argv[1] == "interactive":
await interactive_client()
elif sys.argv[1] == "demo":
await demo_basic_operations()
await demo_advanced_usage()
await demo_custom_tools()
else:
print("用法: python -m examples.basic_usage [interactive|demo]")
else:
await demo_basic_operations()
if __name__ == "__main__":
<"www.tangshan.gov.cn.felli.cn">
<"www.changzhou.gov.cn.felli.cn">
<"www.nantong.gov.cn.felli.cn">
asyncio.run(main())
```
## 错误处理和调试
提供完善的错误处理和调试功能。
```python
# mcp_client/debug_utils.py
import logging
from typing import Dict, Any
from .core import MCPClient, MCPResponse
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("MCPClient")
class DebugMCPClient(MCPClient):
"""带调试功能的MCP客户端"""
async def send_request(self, method: str, params: Dict[str, Any] = None) -> MCPResponse:
"""发送请求（带调试信息）"""
logger.info(f"? 发送请求: {method}")
logger.debug(f"请求参数: {params}")
start_time = asyncio.get_event_loop().time()
response = await super().send_request(method, params)
end_time = asyncio.get_event_loop().time()
logger.info(f"? 收到响应: {method} (耗时: {(end_time - start_time)*1000:.2f}ms)")
if response.is_success():
logger.debug(f"响应结果: {response.result}")
else:
logger.error(f"请求失败: {response.error}")
return response
def enable_debug_logging():
"""启用调试日志"""
logging.getLogger("MCPClient").setLevel(logging.DEBUG)
def create_client_with_retry(server_url: str, max_retries: int = 3):
"""创建带重试机制的客户端"""
class RetryMCPClient(DebugMCPClient):
async def send_request_with_retry(self, method: str, params: Dict[str, Any] = None) -> MCPResponse:
"""带重试的请求发送"""
for attempt in range(max_retries):
try:
response = await self.send_request(method, params)
if response.is_success():
return response
elif attempt < max_retries - 1:
wait_time = 2 ** attempt # 指数退避
logger.warning(f"请求失败，{wait_time}秒后重试...")
await asyncio.sleep(wait_time)
except Exception as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
logger.warning(f"请求异常，{wait_time}秒后重试: {e}")
await asyncio.sleep(wait_time)
else:
raise
return response
return RetryMCPClient(server_url)
# 使用示例
async def debug_demo():
"""调试演示"""
# 启用调试日志
enable_debug_logging()
# 创建带重试的客户端
client = create_client_with_retry("http://localhost:8080/mcp")
try:
await client.connect()
# 使用重试机制发送请求
response = await client.send_request_with_retry("tools/list")
if response.is_success():
print("工具列表获取成功！")
finally:
await client.close()
```
通过本文的完整指南，你可以快速构建功能完善的MCP客户端，实现与大模型API的高效交互。这个极简客户端提供了完整的MCP协议支持、友好的工具调用接口和强大的调试功能，是接入大模型生态的理想起点。