"""opax-mcp — MCP tool server for OPAX/Vauco Transport: HTTP (FastAPI + uvicorn) for Cloud Run Alle tools kalles av Jason/Emma internt — ingen 3part. """ import os import httpx from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel from typing import Optional, Any app = FastAPI(title="opax-mcp", version="1.0.0") OPAX_BASE_URL = os.environ.get("OPAX_BASE_URL", "https://opax.vauco.no") MCP_SECRET = os.environ.get("MCP_SECRET", "") GH_REPO = os.environ.get("GH_REPO", "vauco-saas/OSVauco") GITHUB_PAT = os.environ.get("GITHUB_PAT", "") GOOGLE_CLOUD_PROJECT = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") def _auth_check(authorization: Optional[str]): if MCP_SECRET and authorization != f"Bearer {MCP_SECRET}": raise HTTPException(status_code=401, detail="Unauthorized") # --------------------------------------------------------------------------- # Health # --------------------------------------------------------------------------- @app.get("/health") async def health(): return {"status": "ok", "service": "opax-mcp"} # --------------------------------------------------------------------------- # Tool request/response schema # --------------------------------------------------------------------------- class ToolRequest(BaseModel): tool: str params: dict = {} @app.post("/tools/call") async def call_tool( req: ToolRequest, authorization: Optional[str] = Header(default=None) ): _auth_check(authorization) handler = TOOLS.get(req.tool) if not handler: raise HTTPException(status_code=404, detail=f"Unknown tool: {req.tool}") result = await handler(req.params) return {"tool": req.tool, "result": result} @app.get("/tools") async def list_tools(authorization: Optional[str] = Header(default=None)): _auth_check(authorization) return {"tools": list(TOOLS.keys())} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def _opax_get(path: str) -> Any: async with httpx.AsyncClient(timeout=30) as client: r = await client.get(f"{OPAX_BASE_URL}{path}") r.raise_for_status() return r.json() async def _opax_post(path: str, body: dict) -> Any: async with httpx.AsyncClient(timeout=30) as client: r = await client.post(f"{OPAX_BASE_URL}{path}", json=body) r.raise_for_status() return r.json() async def _gh_get(path: str) -> Any: headers = {"Authorization": f"token {GITHUB_PAT}", "Accept": "application/vnd.github+json"} async with httpx.AsyncClient(timeout=30) as client: r = await client.get(f"https://api.github.com{path}", headers=headers) r.raise_for_status() return r.json() async def _gh_post(path: str, body: dict) -> Any: headers = {"Authorization": f"token {GITHUB_PAT}", "Accept": "application/vnd.github+json"} async with httpx.AsyncClient(timeout=30) as client: r = await client.post(f"https://api.github.com{path}", json=body, headers=headers) r.raise_for_status() return r.json() # --------------------------------------------------------------------------- # Billing tools # --------------------------------------------------------------------------- async def get_billing_summary(p): return await _opax_get("/billing/summary") async def get_billing_forecast(p): return await _opax_get("/billing/tokens/estimate") async def get_billing_credits(p): return await _opax_get("/billing/summary") async def get_billing_anomalies(p): return await _opax_get("/billing/anomalies") async def get_billing_history(p): return await _opax_get("/billing/history") async def get_billing_budget(p): return await _opax_get("/billing/budget") async def set_billing_budget(p): return await _opax_post("/billing/budget", {"amount": p.get("amount"), "currency": p.get("currency", "USD")}) # --------------------------------------------------------------------------- # Onboarding tools # --------------------------------------------------------------------------- async def create_invite(p): return await _opax_post("/onboard/invite", { "email": p.get("email"), "name": p.get("name", ""), "tier": p.get("tier", "starter") }) async def list_customers(p): return await _opax_get("/onboard/customers") # --------------------------------------------------------------------------- # Notify tools # --------------------------------------------------------------------------- async def send_webhook(p): return await _opax_post("/notify/webhook", {"message": p.get("message"), "url": p.get("url", "")}) async def send_email(p): return await _opax_post("/notify/email", {"to": p.get("to"), "subject": p.get("subject"), "body": p.get("body")}) async def send_sms(p): return await _opax_post("/notify/sms", {"to": p.get("to"), "message": p.get("message")}) async def get_notify_channels(p): return await _opax_get("/notify/channels") # --------------------------------------------------------------------------- # Agent tools # --------------------------------------------------------------------------- async def run_jason(p): return await _opax_post("/run", {"prompt": p.get("prompt"), "agent": "jason"}) async def run_emma(p): return await _opax_post("/run", {"prompt": p.get("prompt"), "agent": "emma"}) # --------------------------------------------------------------------------- # Platform tools # --------------------------------------------------------------------------- async def get_health(p): return await _opax_get("/health") async def get_build_status(p): return await _opax_get("/build-status") async def get_state(p): return await _opax_get("/state") async def get_telemetry(p): return await _opax_get("/billing/tokens/by-module") async def run_terminal(p): return await _opax_post("/terminal/exec", {"command": p.get("command")}) # --------------------------------------------------------------------------- # GitHub tools # --------------------------------------------------------------------------- async def list_commits(p): repo = p.get("repo", GH_REPO) return await _gh_get(f"/repos/{repo}/commits?per_page={p.get('limit', 10)}") async def get_file(p): repo = p.get("repo", GH_REPO) path = p.get("path", "") return await _gh_get(f"/repos/{repo}/contents/{path}") async def list_open_issues(p): repo = p.get("repo", GH_REPO) return await _gh_get(f"/repos/{repo}/issues?state=open&per_page=20") async def create_github_issue(p): repo = p.get("repo", GH_REPO) return await _gh_post(f"/repos/{repo}/issues", { "title": p.get("title"), "body": p.get("body", "") }) async def push_file(p): """Oppdater en enkelt fil i GitHub via contents API.""" repo = p.get("repo", GH_REPO) path = p.get("path") import base64 content_b64 = base64.b64encode(p.get("content", "").encode()).decode() body = { "message": p.get("message", f"chore: oppdater {path} via opax-mcp"), "content": content_b64, } if p.get("sha"): body["sha"] = p["sha"] return await _gh_post(f"/repos/{repo}/contents/{path}", body) # --------------------------------------------------------------------------- # Tool registry # --------------------------------------------------------------------------- TOOLS = { # Billing "get_billing_summary": get_billing_summary, "get_billing_forecast": get_billing_forecast, "get_billing_credits": get_billing_credits, "get_billing_anomalies": get_billing_anomalies, "get_billing_history": get_billing_history, "get_billing_budget": get_billing_budget, "set_billing_budget": set_billing_budget, # Onboarding "create_invite": create_invite, "list_customers": list_customers, # Notify "send_webhook": send_webhook, "send_email": send_email, "send_sms": send_sms, "get_notify_channels": get_notify_channels, # Agents "run_jason": run_jason, "run_emma": run_emma, # Platform "get_health": get_health, "get_build_status": get_build_status, "get_state": get_state, "get_telemetry": get_telemetry, "run_terminal": run_terminal, # GitHub "list_commits": list_commits, "get_file": get_file, "list_open_issues": list_open_issues, "create_github_issue": create_github_issue, "push_file": push_file, }