PostgreSQL MCP AllAccess

#!/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())