Database Server

from mcp_server import MCPServer
from sqlalchemy import create_engine, inspect
import json

server = MCPServer("database")

# Tool: Query database
@server.tool()
def execute_query(query: str, params: dict = None):
    """Execute a SQL query and return results."""
    engine = create_engine("postgresql://localhost/mydb")
    with engine.connect() as conn:
        result = conn.execute(query, params or {})
        return [dict(row) for row in result]

# Tool: Get schema
@server.tool()
def get_schema():
    """Introspect database schema."""
    inspector = inspect(engine)
    return {
        "tables": inspector.get_table_names(),
        "views": inspector.get_view_names()
    }

if __name__ == "__main__":
    server.run()