Test MCP server integration.
(server_name: Optional[str] = None)
| 22 | |
| 23 | |
| 24 | async def test_server_integration(server_name: Optional[str] = None): |
| 25 | """Test MCP server integration.""" |
| 26 | |
| 27 | print("\n🔧 MCP Server Integration Test") |
| 28 | print("=" * 50) |
| 29 | |
| 30 | # Import required components |
| 31 | from mcp_cli.tools.manager import ToolManager |
| 32 | |
| 33 | # Load server config to find available servers |
| 34 | config_file = "server_config.json" |
| 35 | try: |
| 36 | with open(config_file, "r") as f: |
| 37 | config = json.load(f) |
| 38 | available_servers = list(config.get("mcpServers", {}).keys()) |
| 39 | except Exception as e: |
| 40 | print(f"❌ Failed to load config: {e}") |
| 41 | return False |
| 42 | |
| 43 | if not available_servers: |
| 44 | print("❌ No servers configured in server_config.json") |
| 45 | return False |
| 46 | |
| 47 | # Use provided server or first available |
| 48 | if not server_name: |
| 49 | server_name = available_servers[0] |
| 50 | print(f"ℹ️ No server specified, using: {server_name}") |
| 51 | elif server_name not in available_servers: |
| 52 | print(f"❌ Server '{server_name}' not found") |
| 53 | print(f" Available servers: {', '.join(available_servers)}") |
| 54 | return False |
| 55 | |
| 56 | print(f"📡 Testing server: {server_name}") |
| 57 | |
| 58 | # Initialize ToolManager |
| 59 | print("✅ Initializing ToolManager...", end="") |
| 60 | tool_manager = ToolManager(config_file, [server_name]) |
| 61 | if not await tool_manager.initialize(): |
| 62 | print(" ❌") |
| 63 | return False |
| 64 | print(" ✓") |
| 65 | |
| 66 | # Try to get protocol version |
| 67 | protocol_version = "unknown" |
| 68 | try: |
| 69 | if hasattr(tool_manager, "stream_manager") and hasattr( |
| 70 | tool_manager.stream_manager, "streams" |
| 71 | ): |
| 72 | if len(tool_manager.stream_manager.streams) > 0: |
| 73 | stream = tool_manager.stream_manager.streams[0] |
| 74 | if hasattr(stream, "protocol_version"): |
| 75 | protocol_version = stream.protocol_version |
| 76 | elif hasattr(stream, "_protocol_version"): |
| 77 | protocol_version = stream._protocol_version |
| 78 | elif hasattr(stream, "client") and hasattr( |
| 79 | stream.client, "protocol_version" |
| 80 | ): |
| 81 | protocol_version = stream.client.protocol_version |
no test coverage detected