#!/usr/bin/env python3
"""Simple test client for MCP servers."""
import asyncio
import json
import sys
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client
from mcp import StdioServerParameters
async def test_server(server_command: list[str]) -> None:
"""Test an MCP server by connecting and calling its capabilities.
Args:
server_command: Command to start the server
"""
print(f"🔍 Testing server: {' '.join(server_command)}")
try:
server_params = StdioServerParameters(
command=server_command[0],
args=server_command[1:] if len(server_command) > 1 else []
)
async with stdio_client(server_params) as streams:
async with ClientSession(streams[0], streams[1]) as session:
# Initialize the session
await session.initialize()
print("✅ Server connected successfully!")
# List available tools
tools_result = await session.list_tools()
print(f"🔧 Available tools ({len(tools_result.tools)}):")
for tool in tools_result.tools:
print(f" - {tool.name}: {tool.description}")
# List available resources
try:
resources_result = await session.list_resources()
print(f"📁 Available resources ({len(resources_result.resources)}):")
for resource in resources_result.resources:
print(f" - {resource.name} ({resource.uri}): {resource.description}")
except Exception as e:
print(f"📁 Resources: Error listing - {e}")
# List available prompts
try:
prompts_result = await session.list_prompts()
print(f"💬 Available prompts ({len(prompts_result.prompts)}):")
for prompt in prompts_result.prompts:
print(f" - {prompt.name}: {prompt.description}")
except Exception as e:
print(f"💬 Prompts: Error listing - {e}")
# Test a simple tool if available
if tools_result.tools:
first_tool = tools_result.tools[0]
print(f"\n🧪 Testing tool: {first_tool.name}")
# Try calling with minimal arguments
try:
if first_tool.name == "add":
result = await session.call_tool(first_tool.name, {"a": 5, "b": 3})
elif first_tool.name == "calculator":
result = await session.call_tool(first_tool.name, {"operation": "add", "a": 5, "b": 3})
else:
# Try with empty arguments for other tools
result = await session.call_tool(first_tool.name, {})
print(f"✅ Tool result: {result.content}")
except Exception as e:
print(f"⚠️ Tool test failed: {e}")
print("\n🎉 Server test completed successfully!")
except Exception as e:
print(f"❌ Server test failed: {e}")
async def main() -> None:
"""Main test function."""
if len(sys.argv) < 2:
print("Usage: python test_client.py <server_type>")
print(" server_type: basic, advanced, or enterprise")
sys.exit(1)
server_type = sys.argv[1].lower()
server_commands = {
"basic": ["uv", "run", "python", "src/mcp_server_hero/examples/basic_server.py"],
"advanced": ["uv", "run", "python", "src/mcp_server_hero/examples/advanced_server.py"],
"enterprise": ["uv", "run", "python", "src/mcp_server_hero/examples/enterprise_server.py"],
}
if server_type not in server_commands:
print(f"❌ Unknown server type: {server_type}")
print("Available types: basic, advanced, enterprise")
sys.exit(1)
await test_server(server_commands[server_type])
if __name__ == "__main__":
asyncio.run(main())