test_improved_mcp.py

#!/usr/bin/env python3
"""
测试改进的MCP实现
"""

import asyncio
import logging
from python_agent.mcp.mcp_manager_improved import ImprovedMcpManager
from python_agent.core.config import AgentConfig

# 设置详细日志
logging.basicConfig(level=logging.DEBUG, format='%(name)s - %(levelname)s - %(message)s')

async def test_improved_mcp():
    print("🚀 测试改进的MCP实现")
    print("=" * 50)
    
    try:
        # 加载配置
        config = AgentConfig.load_from_file('python_agent/mcp_config.yaml')
        
        # 创建改进的MCP管理器
        mcp_config = {'mcp_servers': config.mcp_servers}
        manager = ImprovedMcpManager(mcp_config)
        
        print(f"📋 配置了 {len(manager.connections)} 个MCP服务器")
        
        # 显示服务器状态
        for name, connection in manager.connections.items():
            enabled = "启用" if connection.server.enabled else "禁用"
            print(f"  - {name}: {enabled} ({connection.server.command})")
        
        # 启动服务器
        print("\\n🔄 启动MCP服务器...")
        await manager.start_all_servers()
        
        # 等待一下让连接稳定
        await asyncio.sleep(2)
        
        # 检查连接状态
        print("\\n📊 连接状态:")
        status = manager.get_connection_status()
        for name, info in status.items():
            print(f"  - {name}: {info['status']} (工具: {info['tools_count']})")
            if info['error_message']:
                print(f"    错误: {info['error_message']}")
        
        # 获取工具
        print("\\n🛠️ 可用的MCP工具:")
        tools = manager.get_all_tools()
        print(f"总共发现 {len(tools)} 个MCP工具:")
        
        for i, tool in enumerate(tools, 1):
            dangerous = " [危险]" if tool.is_dangerous() else ""
            print(f"  {i}. {tool.get_name()}{dangerous}")
            print(f"     描述: {tool.get_description()}")
            
            if i >= 5:  # 只显示前5个工具
                remaining = len(tools) - 5
                if remaining > 0:
                    print(f"  ... 还有 {remaining} 个工具")
                break
        
        # 健康检查
        print("\\n🏥 健康检查:")
        health = await manager.health_check()
        print(f"  整体状态: {health['overall_status']}")
        print(f"  总服务器: {health['total_servers']}")
        print(f"  已启用: {health['enabled_servers']}")
        print(f"  已连接: {health['connected_servers']}")
        print(f"  有错误: {health['error_servers']}")
        print(f"  总工具: {health['total_tools']}")
        
        # 如果有可用的工具，尝试执行一个简单的测试
        if tools:
            print("\\n🧪 测试工具执行:")
            
            # 找一个相对安全的工具进行测试
            test_tool = None
            for tool in tools:
                if 'evaluate' in tool.get_name() or 'info' in tool.get_name():
                    test_tool = tool
                    break
            
            if test_tool:
                print(f"测试工具: {test_tool.get_name()}")
                try:
                    # 根据工具类型准备测试参数
                    if 'evaluate' in test_tool.get_name():
                        test_input = {"script": "console.log('MCP Test'); 'Hello from MCP'"}
                    else:
                        test_input = {}
                    
                    result = await test_tool.execute(test_input)
                    
                    if result.success:
                        print(f"✅ 工具执行成功:")
                        print(f"   输出: {result.output[:200]}...")
                    else:
                        print(f"⚠️ 工具执行失败: {result.error}")
                        
                except Exception as e:
                    print(f"❌ 工具测试出错: {e}")
            else:
                print("没有找到合适的测试工具")
        
        # 清理
        print("\\n🧹 清理连接...")
        await manager.disconnect_all_servers()
        
        print("\\n✅ 测试完成!")
        
        # 总结
        if tools:
            print(f"\\n🎉 成功！MCP集成工作正常:")
            print(f"  - 连接了 {health['connected_servers']} 个服务器")
            print(f"  - 发现了 {len(tools)} 个工具")
            print(f"  - 可以集成到Python Agent中使用")
        else:
            print(f"\\n⚠️ 部分成功：MCP架构正常但没有可用工具")
            print(f"  - 可能需要调试特定的MCP服务器连接")
        
        return len(tools) > 0
        
    except Exception as e:
        print(f"❌ 测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False

async def main():
    print("🎯 改进的Python Agent MCP集成测试")
    print("=" * 60)
    
    success = await test_improved_mcp()
    
    if success:
        print("\\n🎊 恭喜！改进的MCP实现成功工作！")
    else:
        print("\\n🔧 需要进一步调试MCP连接问题")

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