Supabase MCP Server - Self-Hosted Edition

#!/usr/bin/env python3 """ Supabase MCP Server (FastMCP + Smithery) """ from typing import Optional import os from pydantic import BaseModel, Field from mcp.server.fastmcp import Context, FastMCP from smithery.decorators import smithery class ConfigSchema(BaseModel): SUPABASE_URL: str = Field("", description="URL de votre projet Supabase") SUPABASE_ANON_KEY: str = Field("", description="Clé anonyme Supabase") SUPABASE_SERVICE_ROLE_KEY: Optional[str] = Field(None, description="Clé service (optionnel)") SUPABASE_AUTH_JWT_SECRET: Optional[str] = Field(None, description="JWT secret (optionnel)") @smithery.server(config_schema=ConfigSchema) def create_server() -> FastMCP: server = FastMCP("Supabase MCP Server") def effective_config(ctx: Context): cfg = ctx.session_config supabase_url = getattr(cfg, "SUPABASE_URL", None) or os.getenv("SUPABASE_URL", "") anon_key = getattr(cfg, "SUPABASE_ANON_KEY", None) or os.getenv("SUPABASE_ANON_KEY", "") service_key = getattr(cfg, "SUPABASE_SERVICE_ROLE_KEY", None) or os.getenv("SUPABASE_SERVICE_ROLE_KEY", "") jwt_secret = getattr(cfg, "SUPABASE_AUTH_JWT_SECRET", None) or os.getenv("SUPABASE_AUTH_JWT_SECRET", "") return supabase_url, anon_key, service_key, jwt_secret @server.tool() def execute_sql(sql: str, ctx: Context) -> str: supabase_url, anon_key, _, _ = effective_config(ctx) if not supabase_url or not anon_key: return "Configuration Supabase manquante (SUPABASE_URL, SUPABASE_ANON_KEY)." return f"SQL reçu ({len(sql)} chars). Connexion prête pour {supabase_url}." @server.tool() def list_tables(ctx: Context) -> str: supabase_url, _, _, _ = effective_config(ctx) if not supabase_url: return "Configuration Supabase manquante." return "Tables listées (démo)." @server.tool() def check_health(ctx: Context) -> str: supabase_url, _, _, _ = effective_config(ctx) if not supabase_url: return "Configuration Supabase manquante." return "OK" # --- Database / Extensions / Stats --- @server.tool() def list_extensions(ctx: Context) -> str: cfg = ctx.session_config if not cfg.SUPABASE_URL: return "Configuration Supabase manquante." return "Extensions listées (démo)." @server.tool() def get_database_stats(ctx: Context) -> str: return "Stats DB (démo)." @server.tool() def get_database_connections(ctx: Context) -> str: return "Connexions DB (démo)." @server.tool() def apply_migration(version: str, ctx: Context) -> str: return f"Migration appliquée: {version} (démo)." @server.tool() def list_migrations(ctx: Context) -> str: return "Migrations listées (démo)." @server.tool() def create_migration(name: str, ctx: Context) -> str: return f"Migration créée: {name} (démo)." @server.tool() def validate_migration(version: str, ctx: Context) -> str: return f"Migration validée: {version} (démo)." @server.tool() def push_migrations(ctx: Context) -> str: return "Migrations poussées (démo)." @server.tool() def smart_migration(ctx: Context) -> str: return "Migration intelligente (démo)." @server.tool() def auto_migrate(ctx: Context) -> str: return "Auto-migrate (démo)." @server.tool() def sync_schema(ctx: Context) -> str: return "Schéma synchronisé (démo)." @server.tool() def inspect_schema(ctx: Context) -> str: return "Schéma inspecté (démo)." @server.tool() def import_schema(sql: str, ctx: Context) -> str: return "Schéma importé (démo)." @server.tool() def execute_psql(command: str, ctx: Context) -> str: return f"psql exécuté: {command[:80]} (démo)." @server.tool() def auto_create_indexes(ctx: Context) -> str: return "Index auto-créés (démo)." @server.tool() def vacuum_analyze(ctx: Context) -> str: return "VACUUM ANALYZE (démo)." # --- Types & Codegen --- @server.tool() def generate_typescript_types(ctx: Context) -> str: return "Types TypeScript générés (démo)." @server.tool() def generate_crud_api(ctx: Context) -> str: return "API CRUD générée (démo)." # --- Logs & Docs & Monitoring --- @server.tool() def get_logs(service: Optional[str] = None, ctx: Context = None) -> str: return f"Logs récupérés (service={service}) (démo)." @server.tool() def search_docs(query: str, ctx: Context) -> str: return f"Recherche docs: {query[:60]} (démo)." @server.tool() def metrics_dashboard(ctx: Context) -> str: return "Dashboard métriques (démo)." @server.tool() def analyze_performance(ctx: Context) -> str: return "Analyse performance (démo)." @server.tool() def analyze_rls_coverage(ctx: Context) -> str: return "Couverture RLS (démo)." @server.tool() def audit_security(ctx: Context) -> str: return "Audit sécurité (démo)." # --- Auth --- @server.tool() def list_auth_users(ctx: Context) -> str: return "Utilisateurs auth listés (démo)." @server.tool() def get_auth_user(id: Optional[str] = None, email: Optional[str] = None, ctx: Context = None) -> str: return f"Auth user récupéré (id={id}, email={email}) (démo)." @server.tool() def create_auth_user(email: str, password: str, ctx: Context) -> str: return f"Utilisateur créé: {email} (démo)." @server.tool() def delete_auth_user(id: str, ctx: Context) -> str: return f"Utilisateur supprimé: {id} (démo)." @server.tool() def update_auth_user(id: str, ctx: Context) -> str: return f"Utilisateur mis à jour: {id} (démo)." @server.tool() def manage_roles(ctx: Context) -> str: return "Rôles gérés (démo)." @server.tool() def manage_rls_policies(ctx: Context) -> str: return "Politiques RLS gérées (démo)." @server.tool() def verify_jwt_secret(ctx: Context) -> str: return "JWT secret présent (démo)." # --- Storage --- @server.tool() def list_storage_buckets(ctx: Context) -> str: return "Buckets listés (démo)." @server.tool() def list_storage_objects(bucket_id: str, ctx: Context) -> str: return f"Objets listés pour bucket={bucket_id} (démo)." @server.tool() def manage_storage_policies(ctx: Context) -> str: return "Politiques Storage gérées (démo)." # --- Realtime --- @server.tool() def list_realtime_publications(ctx: Context) -> str: return "Publications realtime listées (démo)." @server.tool() def manage_realtime(ctx: Context) -> str: return "Realtime géré (démo)." @server.tool() def create_subscription(channel: str, ctx: Context) -> str: return f"Subscription créée: {channel} (démo)." @server.tool() def delete_subscription(channel: str, ctx: Context) -> str: return f"Subscription supprimée: {channel} (démo)." # --- Admin / Secrets / Webhooks / Triggers / Functions --- @server.tool() def manage_extensions(ctx: Context) -> str: return "Extensions gérées (démo)." @server.tool() def manage_functions(ctx: Context) -> str: return "Fonctions gérées (démo)." @server.tool() def manage_triggers(ctx: Context) -> str: return "Triggers gérés (démo)." @server.tool() def manage_webhooks(ctx: Context) -> str: return "Webhooks gérés (démo)." @server.tool() def rebuild_hooks(ctx: Context) -> str: return "Hooks reconstruits (démo)." @server.tool() def manage_secrets(ctx: Context) -> str: return "Secrets gérés (démo)." # --- Divers utilitaires --- @server.tool() def cache_management(ctx: Context) -> str: return "Cache géré (démo)." @server.tool() def environment_management(ctx: Context) -> str: return "Environnement géré (démo)." @server.tool() def get_project_url(ctx: Context) -> str: return "URL projet (démo)." @server.tool() def get_anon_key(ctx: Context) -> str: return "Anon key (démo)." @server.tool() def get_service_key(ctx: Context) -> str: return "Service key (démo)." return server if __name__ == "__main__": srv = create_server() print("Serveur FastMCP prêt (Smithery). Outils:", list(srv._tools.keys()))