"""Ultron master DB — in-process MCP giving Jarvis full Supabase/Postgres access. Ultron scrapes Google-Maps business leads and stores EVERYTHING in one Supabase Postgres (leads, plus the CRM: clients, contacts, lead_clients, app_users, notes, scan coverage, scraper status). This module lets Jarvis (and any Claude on the same box) run arbitrary SQL against it — read anything ("how many clients signed", "who contacted whom", "fetch coffee-shop leads") and write anything (add leads, create a client, make a user). Composes with the Drive + Smartlead tools: e.g. query leads -> format a CSV -> drive_write it -> later load into a Smartlead campaign. Connection mirrors Ultron's own src/supabase.js: db..supabase.co:5432, user postgres, SSL. Credentials come from the process env if present, else are read from the Ultron repo's .env (the single source of truth). MCP_ULTRONDB=0 disables it. SAFETY: full control, but a seatbelt on the irreversible stuff — DROP, TRUNCATE, and DELETE/UPDATE with no WHERE need confirm="yes". Everything else runs freely. Tools (mcp__ultron__): ultron_query run SQL (read or write) against the master DB ultron_schema list tables (with row counts) or one table's columns """ from __future__ import annotations import os import re from pathlib import Path _PROJECT = Path(__file__).resolve().parent.parent # claude-voice/ _ULTRON_ENV = Path(os.environ.get("ULTRON_ENV", str(_PROJECT.parent / ".env"))) # Ultron/.env _DANGER = re.compile(r"\b(drop|truncate)\b", re.I) def _creds() -> tuple[str, str] | None: """(host, password) from env, falling back to the Ultron repo .env.""" url = os.environ.get("SUPABASE_URL", "") pw = os.environ.get("SUPABASE_DB_PASSWORD", "") if not (url and pw) and _ULTRON_ENV.exists(): for line in _ULTRON_ENV.read_text().splitlines(): line = line.strip() if line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") k, v = k.strip(), v.strip() if k == "SUPABASE_URL" and not url: url = v elif k == "SUPABASE_DB_PASSWORD" and not pw: pw = v if not (url and pw): return None m = re.search(r"//([^.]+)\.", url) if not m: return None return f"db.{m.group(1)}.supabase.co", pw def enabled() -> bool: if os.environ.get("MCP_ULTRONDB", "1") == "0": return False return _creds() is not None def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} def _connect(): import psycopg host, pw = _creds() return psycopg.connect( f"host={host} port=5432 user=postgres password={pw} " f"dbname=postgres sslmode=require", connect_timeout=15, autocommit=True) def _needs_confirm(sql: str) -> bool: s = sql.strip() if _DANGER.search(s): return True if re.match(r"(?is)^\s*(delete|update)\b", s) and \ not re.search(r"\bwhere\b", s, re.I): return True return False def _run(sql: str) -> str: with _connect() as conn: cur = conn.execute(sql) if cur.description: # a result set (SELECT / RETURNING / RPC) cols = [d.name for d in cur.description] rows = cur.fetchmany(500) if not rows: return "(0 rows)" head = " | ".join(cols) body = "\n".join( " | ".join("" if v is None else str(v) for v in r) for r in rows) more = "\n…(showing first 500 rows)" if len(rows) == 500 else "" out = f"{head}\n{'-' * len(head)}\n{body}{more}" return out[:5000] + (" …[truncated]" if len(out) > 5000 else "") return f"OK — {cur.rowcount} row(s) affected." def build_server(): import asyncio from claude_agent_sdk import tool, create_sdk_mcp_server @tool("ultron_query", "Run SQL against the Ultron master database (Supabase Postgres) — " "full read AND write over every table: leads, clients, contacts, " "lead_clients, user_clients, lead_notes, app_users, scan_ledger, " "scraper_status. SELECT to fetch/aggregate (e.g. coffee-shop leads, " "how many clients signed, who contacted whom); INSERT/UPDATE to add " "leads, create a client, make a user; call RPCs via SELECT * FROM " "fn(...). Destructive statements (DROP/TRUNCATE, or DELETE/UPDATE with " "no WHERE) require confirm='yes'. Use ultron_schema first if unsure of " "columns.", {"sql": str, "confirm": str}) async def ultron_query(args: dict) -> dict: sql = str(args.get("sql", "")).strip() if not sql: return _text("ultron_query failed: no SQL.") if _needs_confirm(sql) and str(args.get("confirm", "")).lower() != "yes": return _text("That statement is irreversible (drop/truncate or a " "WHERE-less delete/update). Re-send with confirm='yes' " "if you really mean it.") try: return _text(await asyncio.to_thread(_run, sql)) except Exception as e: # noqa: BLE001 return _text(f"ultron_query error: {str(e)[:250]}") @tool("ultron_schema", "Inspect the database structure. No `table` → list every table with " "its row count. With `table` → that table's columns (name + type) so " "you know exactly what to query/write. Use before composing a query " "on a table you're unsure about.", {"table": str}) async def ultron_schema(args: dict) -> dict: table = str(args.get("table", "")).strip() def work(): with _connect() as conn: if not table: tabs = [r[0] for r in conn.execute( "select table_name from information_schema.tables " "where table_schema='public' order by 1").fetchall()] lines = [] for t in tabs: try: n = conn.execute( f'select count(*) from "{t}"').fetchone()[0] except Exception: # noqa: BLE001 n = "?" lines.append(f"- {t} ({n} rows)") return "Tables:\n" + "\n".join(lines) cols = conn.execute( "select column_name, data_type from " "information_schema.columns where table_schema='public' " "and table_name=%s order by ordinal_position", (table,)).fetchall() if not cols: return f"No table named {table!r}." return (f"{table} columns:\n" + "\n".join(f"- {c} ({t})" for c, t in cols)) try: return _text(await asyncio.to_thread(work)) except Exception as e: # noqa: BLE001 return _text(f"ultron_schema error: {str(e)[:250]}") return create_sdk_mcp_server( name="ultron", version="1.0.0", tools=[ultron_query, ultron_schema], )