MCP Client 调试与问题定位技巧

import logging
import json
import time
import os
from typing import Dict, Any, Optional
from dataclasses import dataclass

@dataclass
class LogEntry:
    """日志条目"""
    level: str
    message: str
    timestamp: float
    request_id: Optional[str] = None
    module: Optional[str] = None
    component: Optional[str] = None
    context: Optional[Dict[str, Any]] = None
    error: Optional[str] = None

class SmartLogger:
    """智能日志系统"""
    
    def __init__(self, name: str, log_level: str = "INFO"):
        """
        初始化智能日志系统
        
        Args:
            name: 日志名称
            log_level: 日志级别
        """
        self.name = name
        self.log_level = log_level
        
        # 创建日志记录器
        self.logger = logging.getLogger(name)
        self.logger.setLevel(getattr(logging, log_level.upper()))
        
        # 创建控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(getattr(logging, log_level.upper()))
        
        # 创建文件处理器
        log_file = f"{name}_{time.strftime('%Y-%m-%d')}.log"
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.DEBUG)  # 文件日志记录所有级别
        
        # 创建格式化器
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        
        # 设置格式化器
        console_handler.setFormatter(formatter)
        file_handler.setFormatter(formatter)
        
        # 添加处理器
        self.logger.addHandler(console_handler)
        self.logger.addHandler(file_handler)
        
        # 结构化日志存储
        self.logs = []
        
    def set_level(self, level: str):
        """
        设置日志级别
        
        Args:
            level: 日志级别
        """
        self.log_level = level
        log_level = getattr(logging, level.upper())
        self.logger.setLevel(log_level)
        for handler in self.logger.handlers:
            if isinstance(handler, logging.StreamHandler):
                handler.setLevel(log_level)
    
    def log(self, level: str, message: str, request_id: Optional[str] = None,
            module: Optional[str] = None, component: Optional[str] = None,
            context: Optional[Dict[str, Any]] = None, error: Optional[str] = None):
        """
        """
        记录日志
        
        Args:
            level: 日志级别
            message: 日志消息
            request_id: 请求 ID
            module: 模块名称
            component: 组件名称
            context: 上下文信息
            error: 错误信息
        """
        # 创建日志条目
        log_entry = LogEntry(
            level=level,
            message=message,
            timestamp=time.time(),
            request_id=request_id,
            module=module,
            component=component,
            context=context,
            error=error
        )
        
        # 添加到日志列表
        self.logs.append(log_entry)
        
        # 限制日志列表大小
        if len(self.logs) > 10000:
            self.logs = self.logs[-10000:]
        
        # 记录到日志系统
        log_func = getattr(self.logger, level.lower())
        
        # 格式化日志消息
        log_message = message
        if request_id:
            log_message = f"[{request_id}] {log_message}"
        if module:
            log_message = f"[{module}] {log_message}"
        if component:
            log_message = f"[{component}] {log_message}"
        if context:
            log_message = f"{log_message} | Context: {json.dumps(context, ensure_ascii=False)}"
        if error:
            log_message = f"{log_message} | Error: {error}"
        
        log_func(log_message)
    
    def debug(self, message: str, **kwargs):
        """记录 DEBUG 级别日志"""
        self.log("DEBUG", message, **kwargs)
    
    def info(self, message: str, **kwargs):
        """记录 INFO 级别日志"""
        self.log("INFO", message, **kwargs)
    
    def warning(self, message: str, **kwargs):
        """记录 WARNING 级别日志"""
        self.log("WARNING", message, **kwargs)
    
    def error(self, message: str, **kwargs):
        """记录 ERROR 级别日志"""
        self.log("ERROR", message, **kwargs)
    
    def critical(self, message: str, **kwargs):
        """记录 CRITICAL 级别日志"""
        self.log("CRITICAL", message, **kwargs)
    
    def get_logs(self, level: Optional[str] = None,
                request_id: Optional[str] = None,
                module: Optional[str] = None,
                component: Optional[str] = None,
                start_time: Optional[float] = None,
                end_time: Optional[float] = None) -> list:
        """
        获取日志列表
        
        Args:
            level: 日志级别过滤
            request_id: 请求 ID 过滤
            module: 模块过滤
            component: 组件过滤
            start_time: 开始时间
            end_time: 结束时间
            
        Returns:
            过滤后的日志列表
        """
        filtered_logs = self.logs.copy()
        
        # 按级别过滤
        if level:
            filtered_logs = [log for log in filtered_logs if log.level.upper() == level.upper()]
        
        # 按请求 ID 过滤
        if request_id:
            filtered_logs = [log for log in filtered_logs if log.request_id == request_id]
        
        # 按模块过滤
        if module:
            filtered_logs = [log for log in filtered_logs if log.module == module]
        
        # 按组件过滤
        if component:
            filtered_logs = [log for log in filtered_logs if log.component == component]
        
        # 按时间范围过滤
        if start_time:
            filtered_logs = [log for log in filtered_logs if log.timestamp >= start_time]
        if end_time:
            filtered_logs = [log for log in filtered_logs if log.timestamp <= end_time]
        
        return filtered_logs
    
    def clear_logs(self):
        """清空日志列表"""
        self.logs.clear()

import asyncio
import json
from typing import Dict, Any, Optional
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel

app = FastAPI(title="MCP Client 调试 API")

# 模拟 MCP Client 实例
class MockMCPClient:
    """模拟 MCP Client 实例"""
    
    def __init__(self):
        self.config = {
            "server_url": "http://localhost:8000/mcp",
            "timeout": 30.0,
            "retry_count": 3,
            "log_level": "INFO",
            "debug_enabled": False
        }
        self.status = {
            "is_connected": True,
            "active_requests": 0,
            "total_requests": 100,
            "successful_requests": 95,
            "error_count": 5,
            "uptime": 3600
        }
    
    def get_config(self) -> Dict[str, Any]:
        """获取配置"""
        return self.config.copy()
    
    def update_config(self, updates: Dict[str, Any]) -> Dict[str, Any]:
        """更新配置"""
        self.config.update(updates)
        return self.config.copy()
    
    def get_status(self) -> Dict[str, Any]:
        """获取状态"""
        return self.status.copy()
    
    def set_debug_mode(self, enabled: bool) -> bool:
        """设置调试模式"""
        self.config["debug_enabled"] = enabled
        return enabled
    
    def get_debug_info(self) -> Dict[str, Any]:
        """获取调试信息"""
        return {
            "config": self.config,
            "status": self.status,
            "debug_data": {
                "active_calls": [],
                "recent_errors": [],
                "performance_metrics": {}
            }
        }

# 创建模拟实例
mcp_client = MockMCPClient()

# 调试 API 模型
class ConfigUpdate(BaseModel):
    """配置更新模型"""
    updates: Dict[str, Any]

class DebugMode(BaseModel):
    """调试模式模型"""
    enabled: bool

# API 端点
@app.get("/api/debug/config", response_model=Dict[str, Any])
def get_config():
    """获取 MCP Client 配置"""
    return mcp_client.get_config()

@app.put("/api/debug/config", response_model=Dict[str, Any])
def update_config(config_update: ConfigUpdate):
    """更新 MCP Client 配置"""
    return mcp_client.update_config(config_update.updates)

@app.get("/api/debug/status", response_model=Dict[str, Any])
def get_status():
    """获取 MCP Client 状态"""
    return mcp_client.get_status()

@app.get("/api/debug/info", response_model=Dict[str, Any])
def get_debug_info():
    """获取详细调试信息"""
    return mcp_client.get_debug_info()

@app.post("/api/debug/debug_mode", response_model=Dict[str, bool])
def set_debug_mode(debug_mode: DebugMode):
    """设置调试模式"""
    result = mcp_client.set_debug_mode(debug_mode.enabled)
    return {"enabled": result}

@app.post("/api/debug/restart", response_model=Dict[str, str])
def restart_client():
    """重启 MCP Client"""
    # 模拟重启
    return {"message": "MCP Client 已重启"}

@app.get("/api/debug/logs", response_model=Dict[str, Any])
def get_logs(
    level: Optional[str] = Query(None, description="日志级别过滤"),
    request_id: Optional[str] = Query(None, description="请求 ID 过滤"),
    limit: int = Query(100, description="日志数量限制")
):
    """获取日志"""
    # 模拟日志数据
    logs = [
        {
            "timestamp": "2026-01-01T12:00:00Z",
            "level": "INFO",
            "message": "MCP Client 启动",
            "request_id": None,
            "module": "client",
            "component": "core"
        },
        {
            "timestamp": "2026-01-01T12:01:00Z",
            "level": "DEBUG",
            "message": "连接到 MCP Server",
            "request_id": "req-123",
            "module": "client",
            "component": "connector"
        },
        {
            "timestamp": "2026-01-01T12:02:00Z",
            "level": "ERROR",
            "message": "调用工具失败",
            "request_id": "req-456",
            "module": "client",
            "component": "executor",
            "error": "工具执行超时"
        }
    ]
    
    # 过滤日志
    filtered_logs = logs
    if level:
        filtered_logs = [log for log in filtered_logs if log["level"] == level.upper()]
    if request_id:
        filtered_logs = [log for log in filtered_logs if log["request_id"] == request_id]
    
    # 限制数量
    filtered_logs = filtered_logs[:limit]
    
    return {
        "logs": filtered_logs,
        "total": len(filtered_logs),
        "limit": limit
    }

@app.get("/health")
def health_check():
    """健康检查"""
    return {"status": "healthy"}

# 运行调试服务器
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

import os
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.aiohttp_client import AioHttpClientInstrumentor

# 初始化分布式追踪
def init_tracing(service_name: str = "mcp-client"):
    """
    初始化分布式追踪
    
    Args:
        service_name: 服务名称
    """
    # 创建资源
    resource = Resource(
        attributes={
            SERVICE_NAME: service_name
        }
    )
    
    # 创建追踪提供程序
    tracer_provider = TracerProvider(resource=resource)
    trace.set_tracer_provider(tracer_provider)
    
    # 创建 Jaeger 导出器
    jaeger_exporter = JaegerExporter(
        agent_host_name="localhost",
        agent_port=6831,
    )
    
    # 创建批处理 span 处理器
    span_processor = BatchSpanProcessor(jaeger_exporter)
    tracer_provider.add_span_processor(span_processor)
    
    # 初始化请求工具的自动追踪
    RequestsInstrumentor().instrument()
    AioHttpClientInstrumentor().instrument()
    
    print(f"分布式追踪已初始化，服务名称: {service_name}")

# 获取追踪器
def get_tracer(name: str) -> trace.Tracer:
    """
    获取追踪器
    
    Args:
        name: 追踪器名称
        
    Returns:
        追踪器实例
    """
    return trace.get_tracer(name)

# 示例：使用追踪器
async def example_with_tracing():
    """使用追踪器的示例"""
    tracer = get_tracer("mcp-client-example")
    
    # 创建根 span
    with tracer.start_as_current_span("mcp-client-request") as root_span:
        root_span.set_attribute("request.type", "tool_call")
        root_span.set_attribute("client.version", "2.0")
        
        # 模拟工具调用
        with tracer.start_as_current_span("tool-selection") as selection_span:
            selection_span.set_attribute("tool.count", 5)
            selection_span.set_attribute("selected.tool", "weather_api")
            # 模拟工具选择逻辑
            await asyncio.sleep(0.1)
        
        # 模拟工具执行
        with tracer.start_as_current_span("tool-execution") as execution_span:
            execution_span.set_attribute("tool.name", "weather_api")
            execution_span.set_attribute("tool.params.city", "beijing")
            # 模拟工具执行逻辑
            await asyncio.sleep(0.2)
        
        # 模拟结果处理
        with tracer.start_as_current_span("result-processing") as processing_span:
            processing_span.set_attribute("result.status", "success")
            # 模拟结果处理逻辑
            await asyncio.sleep(0.05)
        
        root_span.set_attribute("request.status", "success")
        print("带有追踪的请求处理完成")

# 运行示例
if __name__ == "__main__":
    # 初始化追踪
    init_tracing()
    
    # 运行示例
    import asyncio
    asyncio.run(example_with_tracing())

# 示例：MCP Client 调试与问题定位示例
import asyncio
import json
from smart_logger import SmartLogger
from distributed_tracing import init_tracing, get_tracer

async def example_debugging():
    """调试与问题定位示例"""
    print("=== MCP Client 调试与问题定位示例 ===")
    
    # 1. 初始化日志系统
    logger = SmartLogger("mcp-client", log_level="DEBUG")
    logger.info("初始化 MCP Client 调试系统")
    
    # 2. 初始化分布式追踪
    init_tracing()
    tracer = get_tracer("mcp-client-debug")
    
    # 3. 模拟 MCP Client 请求处理
    request_id = "debug-req-001"
    
    with tracer.start_as_current_span("mcp-client-debug-request") as root_span:
        root_span.set_attribute("request_id", request_id)
        root_span.set_attribute("debug_mode", True)
        
        # 记录请求开始
        logger.info("处理 MCP 请求", request_id=request_id, module="client", component="core")
        
        try:
            # 模拟工具调用流程
            logger.debug("开始工具选择", request_id=request_id, module="client", component="tool-selector")
            
            with tracer.start_as_current_span("tool-selection") as selection_span:
                selection_span.set_attribute("request_id", request_id)
                selection_span.set_attribute("tool_count", 3)
                
                # 模拟工具选择
                await asyncio.sleep(0.1)
                selected_tool = "weather_api"
                
                logger.debug(f"选择工具: {selected_tool}", request_id=request_id, 
                           module="client", component="tool-selector")
                selection_span.set_attribute("selected_tool", selected_tool)
            
            # 模拟工具执行
            logger.debug(f"开始执行工具: {selected_tool}", request_id=request_id, 
                       module="client", component="tool-executor")
            
            with tracer.start_as_current_span("tool-execution") as execution_span:
                execution_span.set_attribute("request_id", request_id)
                execution_span.set_attribute("tool_name", selected_tool)
                
                # 模拟工具执行错误
                raise Exception("工具执行超时")
                
        except Exception as e:
            # 记录错误
            logger.error(f"工具调用失败: {e}", request_id=request_id, 
                       module="client", component="core", error=str(e))
            
            # 设置错误属性
            root_span.set_attribute("request_status", "error")
            root_span.set_attribute("error_type", type(e).__name__)
            root_span.set_attribute("error_message", str(e))
            
            # 模拟错误处理
            logger.debug("开始错误处理", request_id=request_id, module="client", component="error-handler")
            
            with tracer.start_as_current_span("error-handling") as error_span:
                error_span.set_attribute("request_id", request_id)
                error_span.set_attribute("error_type", type(e).__name__)
                
                # 模拟错误处理逻辑
                await asyncio.sleep(0.1)
                
                # 记录错误处理结果
                logger.info("错误处理完成，返回默认结果", request_id=request_id, 
                           module="client", component="error-handler")
        
        # 记录请求结束
        logger.info("MCP 请求处理完成", request_id=request_id, module="client", component="core")
    
    # 4. 展示日志查询功能
    print("\n=== 日志查询示例 ===")
    
    # 查询 ERROR 级别日志
    error_logs = logger.get_logs(level="ERROR")
    print(f"ERROR 级别日志数量: {len(error_logs)}")
    for log in error_logs:
        print(f"- [{log.level}] {log.message} | {log.timestamp}")
    
    # 根据请求 ID 查询日志
    req_logs = logger.get_logs(request_id=request_id)
    print(f"\n请求 {request_id} 的日志数量: {len(req_logs)}")
    for log in req_logs[:3]:  # 只显示前 3 条
        print(f"- [{log.level}] {log.message} | {log.module} | {log.component}")
    
    # 5. 模拟调试 API 调用
    print("\n=== 调试 API 调用示例 ===")
    print("调试 API 提供以下功能:")
    print("- GET /api/debug/config - 获取配置")
    print("- PUT /api/debug/config - 更新配置")
    print("- GET /api/debug/status - 获取状态")
    print("- GET /api/debug/info - 获取调试信息")
    print("- POST /api/debug/debug_mode - 设置调试模式")
    print("- GET /api/debug/logs - 查询日志")
    print("- POST /api/debug/restart - 重启客户端")

async def main():
    """主函数"""
    await example_debugging()

if __name__ == "__main__":
    asyncio.run(main())