"""
Enhanced MCP Server with PostgreSQL Integration Resources and Tools
Exposes comprehensive database functions as both MCP resources and tools for LLM interaction
"""
import asyncio
import logging
import sys
from pathlib import Path
from typing import Any, Dict
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from mcp.server import Server
from mcp.types import Tool, Resource
import mcp.server.stdio
from services.mcp_resources_service import MCPResourcesService
from services.mcp_tools_service import MCPToolsService
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class EnhancedMCPServer:
"""Enhanced MCP server with comprehensive PostgreSQL integration via Resources and Tools"""
def __init__(self, config: Dict[str, Any] = None):
self.config = config or {}
self.server = Server("enhanced-mcp-database-server")
# Initialize both resources and tools services
self.resources_service = MCPResourcesService(self.config)
self.tools_service = MCPToolsService(self.config)
self._register_resources()
self._register_tools()
self._register_handlers()
def _register_resources(self):
"""Register all MCP resources from resource definitions"""
try:
# Get all available resources from the service
available_resources = self.resources_service.list_resources()
for resource_def in available_resources:
resource = Resource(
uri=resource_def["uri"],
name=resource_def["name"],
description=resource_def["description"],
mimeType=resource_def["mimeType"]
)
# Note: MCP server handles resource registration automatically
# when resources are listed via list_resources handler
logger.info(f"MCP resources ready: {len(available_resources)} resources available")
except Exception as e:
logger.error(f"Error registering resources: {e}")
def _register_tools(self):
"""Register all MCP tools from tool definitions"""
try:
# Load tool definitions for the list_tools handler
tool_info = self.tools_service.get_available_tools()
self.available_tools = tool_info.get("tools", [])
logger.info(f"MCP tools loaded: {len(self.available_tools)} tools")
except Exception as e:
logger.error(f"Error loading tools: {e}")
self.available_tools = []
def _register_handlers(self):
"""Register both resource and tool handlers"""
@self.server.list_resources()
async def handle_list_resources() -> list[Resource]:
"""Handle MCP resource listing"""
try:
logger.info("Listing available MCP resources")
available_resources = self.resources_service.list_resources()
resources = []
for resource_def in available_resources:
resource = Resource(
uri=resource_def["uri"],
name=resource_def["name"],
description=resource_def["description"],
mimeType=resource_def["mimeType"]
)
resources.append(resource)
logger.info(f"Listed {len(resources)} resources")
return resources
except Exception as e:
logger.error(f"Error listing resources: {e}")
return []
@self.server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Handle MCP resource reading"""
try:
logger.info(f"Reading resource: {uri}")
result = await self.resources_service.read_resource(uri)
return result.get("text", "{}")
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
return f'{{"error": "{str(e)}", "uri": "{uri}"}}'
@self.server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""Handle MCP tool listing"""
try:
logger.info("Listing available MCP tools")
tools = []
for tool_def in self.available_tools:
tool = Tool(
name=tool_def["name"],
description=tool_def["description"],
inputSchema=tool_def["parameters"]
)
tools.append(tool)
logger.info(f"Listed {len(tools)} tools")
return tools
except Exception as e:
logger.error(f"Error listing tools: {e}")
return []
@self.server.call_tool()
async def handle_tool_call(name: str, arguments: Dict[str, Any]) -> list[Any]:
"""Handle MCP tool calls by routing to the tools service"""
try:
logger.info(f"Handling tool call: {name} with arguments: {arguments}")
# Route to tools service
result = await self.tools_service.handle_tool_call(name, arguments)
logger.info(f"Tool call {name} completed successfully")
# Return result in MCP expected format
return [result]
except Exception as e:
logger.error(f"Error handling tool call {name}: {e}")
return [{
"success": False,
"error": str(e),
"tool_name": name,
"arguments": arguments
}]
logger.info("Resource and tool handlers registered successfully")
async def start(self):
"""Start the MCP server"""
logger.info("Starting Enhanced MCP Database Server...")
logger.info("Available databases: db1, db2, db3")
logger.info("Ready to handle database resources and tools")
logger.info("Resources: Static database information (schemas, stats, table lists)")
logger.info("Tools: Active database operations (SQL execution, search, analysis)")
async def main():
"""Main entry point"""
# Initialize and start the enhanced MCP server
server = EnhancedMCPServer()
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.start()
await server.server.run(
read_stream,
write_stream,
None
)
if __name__ == "__main__":
asyncio.run(main())
