#!/usr/bin/env python3 """ opax-mcp/server.py — OPAX MCP Server Eksponerer hele OPAX-plattformen som MCP-tools. Kjører som egen Cloud Run-service: opax-mcp (us-central1). Auth: Bearer token via MCP_SECRET env-var (Secret Manager: mcp-server-key). Base: OPAX_BASE_URL (default https://opax.vauco.no) Tools: Billing: get_billing_summary, get_billing_forecast, get_billing_credits, get_billing_anomalies, get_billing_history, get_billing_budget, set_billing_budget Onboarding: create_invite, list_customers Notify: send_webhook, send_email, send_sms Agent: run_jason, run_emma Platform: get_health, get_build_status, get_notify_channels, get_state, get_telemetry GitHub: list_commits, get_file, search_code """ import os import sys import json import httpx from typing import Any from mcp.server.fastmcp import FastMCP # ── CONFIG ───────────────────────────────────────────────────────────────── OPAX_BASE_URL = os.environ.get("OPAX_BASE_URL", "https://opax.vauco.no") MCP_SECRET = os.environ.get("MCP_SECRET", "") GH_PAT = os.environ.get("GITHUB_PAT", "") GH_REPO = os.environ.get("GH_REPO", "vauco-saas/OSVauco") PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") mcp = FastMCP("opax-mcp") # ── HTTP HELPERS ───────────────────────────────────────────────────────────── def _opax_headers() -> dict: headers = {"Content-Type": "application/json"} if MCP_SECRET: headers["Authorization"] = f"Bearer {MCP_SECRET}" # IAP identity token (Cloud Run-to-Cloud Run) try: import google.auth import google.auth.transport.requests creds, _ = google.auth.default() creds.refresh(google.auth.transport.requests.Request()) headers["x-goog-authenticated-user-email"] = "serviceAccount:opax-mcp" # For IAP: send identity token import urllib.request meta_url = f"http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?audience={OPAX_BASE_URL}&format=full" req = urllib.request.Request(meta_url, headers={"Metadata-Flavor": "Google"}) token = urllib.request.urlopen(req, timeout=2).read().decode() headers["Authorization"] = f"Bearer {token}" except Exception: pass return headers def _gh_headers() -> dict: h = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"} if GH_PAT: h["Authorization"] = f"Bearer {GH_PAT}" return h async def _get(path: str, params: dict = None) -> Any: async with httpx.AsyncClient(timeout=20.0) as c: r = await c.get(f"{OPAX_BASE_URL}{path}", headers=_opax_headers(), params=params) r.raise_for_status() return r.json() async def _post(path: str, body: dict = None) -> Any: async with httpx.AsyncClient(timeout=30.0) as c: r = await c.post(f"{OPAX_BASE_URL}{path}", headers=_opax_headers(), json=body or {}) r.raise_for_status() return r.json() async def _gh_get(path: str, params: dict = None) -> Any: async with httpx.AsyncClient(timeout=15.0) as c: r = await c.get(f"https://api.github.com{path}", headers=_gh_headers(), params=params) r.raise_for_status() return r.json() # ────────────────────────────────────────────────────────────────────────── # BILLING TOOLS # ────────────────────────────────────────────────────────────────────────── @mcp.tool() async def get_billing_summary() -> str: """Hent GCP-kostnadsoversikt per tjeneste for inneværende måned.""" data = await _get("/billing/summary") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_billing_forecast() -> str: """Hent kostnadsprøve og daglig snitt siste 7 dager.""" data = await _get("/billing/forecast") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_billing_credits(days: int = 90) -> str: """Hent Vertex/GCP-kredittbalanse, burn rate og estimert tom-dato.""" data = await _get("/billing/credits", params={"days": days}) return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_billing_anomalies() -> str: """Detect kostnadsanomalier i GCP-forbruket.""" data = await _get("/billing/anomalies") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_billing_history() -> str: """Hent historisk daglig kostnad siste 90 dager.""" data = await _get("/billing/history") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_billing_budget() -> str: """Hent nåværende budsjettgrense (NOK).""" data = await _get("/billing/budget") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def set_billing_budget(budget: float) -> str: """Sett ny budsjettgrense i NOK. Eksempel: 1500.0""" data = await _post("/billing/budget", {"budget": budget}) return json.dumps(data, ensure_ascii=False, indent=2) # ────────────────────────────────────────────────────────────────────────── # ONBOARDING TOOLS # ────────────────────────────────────────────────────────────────────────── @mcp.tool() async def create_invite(company: str, email: str, tier: str = "starter") -> str: """ Opprett en CostGuard-invitasjon og send til prospekt. Returnerer ferdig onboarding-URL (72t gyldig). tier: starter | guard | shield | enterprise """ data = await _post("/onboard/invite", {"company": company, "email": email, "tier": tier}) return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def list_customers() -> str: """List alle onboardede CostGuard-kunder fra Firestore.""" try: import google.auth from google.cloud import firestore google.auth.default() db = firestore.Client(project=PROJECT_ID) docs = db.collection("onboarded_customers").stream() customers = [] for doc in docs: d = doc.to_dict() d["id"] = doc.id # Fjern sensitive felter d.pop("webhook_url", None) customers.append(d) return json.dumps({"customers": customers, "count": len(customers)}, ensure_ascii=False, indent=2, default=str) except Exception as e: return json.dumps({"error": str(e)}) # ────────────────────────────────────────────────────────────────────────── # NOTIFY TOOLS # ────────────────────────────────────────────────────────────────────────── @mcp.tool() async def send_webhook(url: str, title: str, body: str, event: str = "custom") -> str: """Send varsel til Slack/Teams/Google Chat webhook.""" data = await _post("/notify/webhook", {"url": url, "title": title, "body": body, "event": event}) return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def send_email(to: str, subject: str, body_html: str, event: str = "custom") -> str: """Send e-post via SendGrid. body_html kan være enkel HTML eller ren tekst.""" data = await _post("/notify/email", {"to": to, "subject": subject, "body_html": body_html, "event": event}) return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def send_sms(to: str, body: str, tier: str, event: str = "spike") -> str: """ Send SMS via Twilio. Krever Guard+-tier. to: mobilnummer i E.164-format (+4712345678) """ data = await _post("/notify/sms", {"to": to, "body": body, "event": event, "tier": tier}) return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_notify_channels() -> str: """List alle konfigurerte varslingskanaler (email, sms, webhook).""" data = await _get("/notify/channels") return json.dumps(data, ensure_ascii=False, indent=2) # ────────────────────────────────────────────────────────────────────────── # AGENT TOOLS # ────────────────────────────────────────────────────────────────────────── @mcp.tool() async def run_jason(message: str, session_id: str = "mcp", mode: str = "light") -> str: """ Kjør Jason (kundevendt agent, gemini-2.5-flash). mode: light | heavy """ data = await _post("/run", {"message": message, "user_id": "mcp", "session_id": session_id, "mode": mode}) return data.get("response", json.dumps(data)) @mcp.tool() async def run_emma(message: str, session_id: str = "mcp-emma", history: list = None) -> str: """ Kjør Emma (intern agent, Vertex/Gemma). history: [{"role": "user"|"model", "content": str}] """ data = await _post("/emma", {"message": message, "session_id": session_id, "history": history or []}) return data.get("response", json.dumps(data)) # ────────────────────────────────────────────────────────────────────────── # PLATFORM TOOLS # ────────────────────────────────────────────────────────────────────────── @mcp.tool() async def get_health() -> str: """Sjekk om OPAX Cloud Run-service er oppe.""" data = await _get("/health") return json.dumps(data, ensure_ascii=False) @mcp.tool() async def get_build_status() -> str: """Hent siste 5 Cloud Build-kjøringer og status.""" data = await _get("/opax/build-status") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_state() -> str: """Hent OPAX agent-state snapshot (telemetri, siste kjøringer).""" data = await _get("/state") return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def get_telemetry(limit: int = 20) -> str: """Hent OPAX agent-telemetri historikk.""" data = await _get("/telemetry/history", params={"limit": limit}) return json.dumps(data, ensure_ascii=False, indent=2) @mcp.tool() async def run_terminal(cmd: str) -> str: """ Kjør whitelisted OPAX terminal-kommando. Gyldige: health | billing | build | logs | help """ data = await _post("/terminal/exec", {"cmd": cmd}) return data.get("output", json.dumps(data)) # ────────────────────────────────────────────────────────────────────────── # GITHUB TOOLS # ────────────────────────────────────────────────────────────────────────── @mcp.tool() async def list_commits(branch: str = "main", per_page: int = 10) -> str: """List siste commits i OSVauco-repo.""" data = await _gh_get(f"/repos/{GH_REPO}/commits", params={"sha": branch, "per_page": per_page}) commits = [{"sha": c["sha"][:8], "message": c["commit"]["message"].split("\n")[0], "author": c["commit"]["author"]["name"], "date": c["commit"]["author"]["date"]} for c in data] return json.dumps(commits, ensure_ascii=False, indent=2) @mcp.tool() async def get_file(path: str, ref: str = "main") -> str: """Hent innholdet i en fil fra OSVauco-repo. path: f.eks. 'docs/MASTERPLAN.md'""" data = await _gh_get(f"/repos/{GH_REPO}/contents/{path}", params={"ref": ref}) import base64 content = base64.b64decode(data["content"]).decode("utf-8") return content @mcp.tool() async def list_open_issues() -> str: """List åpne issues i OSVauco-repo (OQ-lista).""" data = await _gh_get(f"/repos/{GH_REPO}/issues", params={"state": "open", "per_page": 30}) issues = [{"number": i["number"], "title": i["title"], "labels": [l["name"] for l in i["labels"]], "created_at": i["created_at"]} for i in data if "pull_request" not in i] return json.dumps(issues, ensure_ascii=False, indent=2) @mcp.tool() async def create_github_issue(title: str, body: str, labels: list = None) -> str: """Opprett et nytt GitHub issue i OSVauco-repo.""" async with httpx.AsyncClient(timeout=15.0) as c: r = await c.post( f"https://api.github.com/repos/{GH_REPO}/issues", headers=_gh_headers(), json={"title": title, "body": body, "labels": labels or []} ) r.raise_for_status() data = r.json() return json.dumps({"number": data["number"], "url": data["html_url"], "title": data["title"]}, indent=2) @mcp.tool() async def push_file(path: str, content: str, message: str, branch: str = "main") -> str: """ Opprett eller oppdater en fil i OSVauco-repo. Henter nåværende SHA automatisk hvis filen eksisterer. """ # Hent nåværende SHA sha = None try: existing = await _gh_get(f"/repos/{GH_REPO}/contents/{path}", params={"ref": branch}) sha = existing.get("sha") except Exception: pass import base64 encoded = base64.b64encode(content.encode("utf-8")).decode("ascii") payload = {"message": message, "content": encoded, "branch": branch} if sha: payload["sha"] = sha async with httpx.AsyncClient(timeout=15.0) as c: r = await c.put( f"https://api.github.com/repos/{GH_REPO}/contents/{path}", headers=_gh_headers(), json=payload ) r.raise_for_status() data = r.json() return json.dumps({"path": path, "sha": data["content"]["sha"][:8], "url": data["content"]["html_url"]}, indent=2) # ── ENTRYPOINT ───────────────────────────────────────────────────────────────── if __name__ == "__main__": mcp.run(transport="stdio")