#!/usr/bin/env python3
"""
Test MCP client connection to Docker SSE endpoint
"""
import asyncio
import httpx
from mcp import ClientSession
from mcp.client.sse import sse_client
async def test_mcp_connection():
"""Test MCP SSE connection and execute a query"""
print("🔌 Connecting to MCP server at http://localhost:3000/sse...")
async with httpx.AsyncClient() as http_client:
async with sse_client("http://localhost:3000/sse", http_client) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the session
await session.initialize()
print("✅ Connected to MCP server!")
# List available tools
tools_result = await session.list_tools()
print(f"\n📋 Available tools: {len(tools_result.tools)}")
for tool in tools_result.tools[:5]: # Show first 5
print(f" - {tool.name}")
# Execute a test query
print("\n🔍 Executing test query...")
result = await session.call_tool(
"execute_query",
{
"query": "SELECT current_database(), current_user, NOW() as current_time",
"format": "table"
}
)
print(f"✅ Query result:\n{result.content[0].text}")
# List schemas
print("\n📊 Listing schemas...")
schemas_result = await session.call_tool(
"list_schemas",
{"include_system": False}
)
print(f"✅ Schemas:\n{schemas_result.content[0].text}")
# Get connection stats
print("\n📈 Connection stats...")
stats_result = await session.call_tool(
"get_connection_stats",
{}
)
print(f"✅ Stats:\n{stats_result.content[0].text}")
if __name__ == "__main__":
asyncio.run(test_mcp_connection())
