5.2 错误代码与解决方案

# 错误示例
try:
    async with sse_client("http://localhost:8000/mcp") as (read, write):
        # ...
except ConnectionRefusedError:
    print("无法连接到服务器")

# 测试连接
telnet localhost 8000
# 或
curl http://localhost:8000/mcp

try:
    result = await session.call_tool("calculator", arguments={"expression": "2 + 2"})
except SessionExpiredError:
    # 重新初始化会话
    await session.initialize()
    result = await session.call_tool("calculator", arguments={"expression": "2 + 2"})

try:
    data = await session.read_resource("data://users/123")
except ResourceNotFoundError:
    # 处理资源不存在的情况
    print("用户数据不存在")
    # 可能需要创建资源或使用默认值

try:
    result = await session.call_tool(
        "calculate_bmi", 
        arguments={"weight_kg": "not_a_number", "height_m": 1.75}
    )
except InvalidToolArgumentsError as e:
    print(f"参数错误: {e}")
    # 使用正确的参数类型重试
    result = await session.call_tool(
        "calculate_bmi", 
        arguments={"weight_kg": 70.0, "height_m": 1.75}
    )

import json

# 防止InvalidJson错误的辅助函数
def safe_json_parse(data):
    try:
        return json.loads(data)
    except json.JSONDecodeError:
        print("JSON格式无效")
        return None

try:
    result = await session.call_tool("complex_calculation", arguments={"data": large_data})
except ServerError as e:
    print(f"服务器错误: {e}")
    # 尝试使用更小的数据集或分批处理
    chunked_results = []
    for chunk in split_data(large_data, chunk_size=1000):
        chunk_result = await session.call_tool("complex_calculation", arguments={"data": chunk})
        chunked_results.append(chunk_result)

# 服务器端：分页资源
@mcp.resource("data://large-collection")
def get_large_collection(page: int = 0, page_size: int = 100) -> dict:
    """获取大型集合的一部分"""
    start = page * page_size
    end = start + page_size
    return {
        "data": large_data[start:end],
        "pagination": {
            "page": page,
            "pageSize": page_size,
            "totalItems": len(large_data),
            "totalPages": (len(large_data) + page_size - 1) // page_size
        }
    }

# 客户端：处理分页
async def fetch_all_items(session):
    all_items = []
    page = 0
    while True:
        response = await session.read_resource(
            "data://large-collection",
            arguments={"page": page, "page_size": 100}
        )
        
        content = response.content[0].json
        all_items.extend(content["data"])
        
        if page >= content["pagination"]["totalPages"] - 1:
            break
            
        page += 1
    
    return all_items

class ResilientMcpClient:
    def __init__(self, server_url, max_retries=5, retry_delay=1):
        self.server_url = server_url
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.session = None
    
    async def ensure_connected(self):
        if self.session is None or not self.session.is_active():
            retries = 0
            while retries < self.max_retries:
                try:
                    transport = await sse_client(self.server_url)
                    read, write = transport
                    self.session = ClientSession(read, write)
                    await self.session.initialize()
                    return
                except Exception as e:
                    retries += 1
                    if retries >= self.max_retries:
                        raise ConnectionError(f"无法连接到MCP服务器: {e}")
                    await asyncio.sleep(self.retry_delay * (2 ** (retries - 1)))  # 指数退避
    
    async def call_tool(self, tool_name, arguments):
        await self.ensure_connected()
        try:
            return await self.session.call_tool(tool_name, arguments)
        except SessionExpiredError:
            self.session = None
            await self.ensure_connected()
            return await self.session.call_tool(tool_name, arguments)

# 服务器端：异步工具执行
@mcp.tool()
async def long_running_operation(task_id: str) -> dict:
    """启动长时间运行的操作并返回任务ID"""
    # 将任务加入队列并立即返回
    task = start_background_task(task_id)
    return {"task_id": task_id, "status": "started"}

@mcp.tool()
async def get_task_status(task_id: str) -> dict:
    """获取任务状态"""
    status = check_task_status(task_id)
    return {"task_id": task_id, "status": status}

# 客户端：轮询任务状态
async def execute_long_running_task(session, task_id):
    # 启动任务
    await session.call_tool("long_running_operation", arguments={"task_id": task_id})
    
    # 轮询状态
    status = "started"
    while status not in ["completed", "failed"]:
        await asyncio.sleep(5)  # 等待5秒
        response = await session.call_tool("get_task_status", arguments={"task_id": task_id})
        status = response.content[0].json["status"]
    
    if status == "completed":
        # 获取结果
        result = await session.read_resource(f"data://task-results/{task_id}")
        return result
    else:
        raise Exception(f"任务失败: {task_id}")