431 lines
20 KiB
Python
431 lines
20 KiB
Python
"""opax-mcp — MCP Streamable HTTP server for OPAX/Vauco
|
|
Transport: MCP Streamable HTTP (JSON-RPC 2.0) on POST /
|
|
Auth: Authorization: Bearer <secret> OR X-MCP-Secret: <secret> OR api-key: <secret>
|
|
"""
|
|
import os
|
|
import json
|
|
import uuid
|
|
import httpx
|
|
import base64
|
|
import google.auth
|
|
import google.auth.transport.requests
|
|
import google.oauth2.id_token
|
|
from fastapi import FastAPI, Request, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from typing import Any, Optional
|
|
import logging
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI(title="opax-mcp", version="3.2.0")
|
|
|
|
OPAX_BASE_URL = os.environ.get("OPAX_BASE_URL", "https://opax.vauco.no")
|
|
OSVAUCO_AGENT_URL = os.environ.get("OSVAUCO_AGENT_URL", "")
|
|
OPAX_IAP_CLIENT_ID = os.environ.get("OPAX_IAP_CLIENT_ID", "")
|
|
MCP_SECRET = os.environ.get("MCP_SECRET", "")
|
|
GITEA_URL = os.environ.get("GITEA_URL", "http://34.59.131.162:3000")
|
|
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
|
|
GITEA_REPO = os.environ.get("GITEA_REPO", "chris/OSVauco")
|
|
GOOGLE_CLOUD_PROJECT = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
|
|
|
|
# Emma-vm Ollama — direkte tilkobling
|
|
OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://34.13.238.133:11434")
|
|
EMMA_MODEL = os.environ.get("EMMA_MODEL", "gemma3:27b")
|
|
EMMA_FAST_MODEL = os.environ.get("EMMA_FAST_MODEL", "gemma3:4b")
|
|
EMMA_LIGHT_MODEL = os.environ.get("EMMA_LIGHT_MODEL", "qwen2.5:3b")
|
|
|
|
logger.info(f"OPAX_IAP_CLIENT_ID: {OPAX_IAP_CLIENT_ID[:10]}...")
|
|
logger.info(f"OSVAUCO_AGENT_URL: {OSVAUCO_AGENT_URL}")
|
|
logger.info(f"OLLAMA_BASE_URL: {OLLAMA_BASE_URL}")
|
|
logger.info(f"EMMA_MODEL: {EMMA_MODEL}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _verify_auth(request: Request):
|
|
if not MCP_SECRET:
|
|
return
|
|
auth = request.headers.get("Authorization", "")
|
|
if auth.startswith("Bearer ") and auth[7:] == MCP_SECRET:
|
|
return
|
|
if request.headers.get("X-MCP-Secret") == MCP_SECRET:
|
|
return
|
|
if request.headers.get("api-key") == MCP_SECRET:
|
|
return
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OPAX helpers (for opax.vauco.no via IAP)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _opax_identity_token() -> str:
|
|
"""Hent identity token for opax.vauco.no (Cloud Run IAP/IAM)."""
|
|
try:
|
|
metadata_url = (
|
|
"http://metadata.google.internal/computeMetadata/v1/instance"
|
|
f"/service-accounts/default/identity?audience={OPAX_IAP_CLIENT_ID}&format=full"
|
|
)
|
|
resp = httpx.get(metadata_url, headers={"Metadata-Flavor": "Google"}, timeout=5)
|
|
resp.raise_for_status()
|
|
token = resp.text.strip()
|
|
if not token:
|
|
raise ValueError("Empty token returned from metadata server")
|
|
return token
|
|
except Exception as e:
|
|
logger.error(f"[IAP TOKEN ERROR] {type(e).__name__}: {e}")
|
|
raise
|
|
|
|
def _opax_headers() -> dict:
|
|
return {"Authorization": f"Bearer {_opax_identity_token()}", "Content-Type": "application/json"}
|
|
|
|
|
|
async def _opax_get(path: str) -> Any:
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.get(f"{OPAX_BASE_URL}{path}", headers=_opax_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def _opax_post(path: str, body: dict) -> Any:
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.post(f"{OPAX_BASE_URL}{path}", json=body, headers=_opax_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OSVauco Agent helpers (for direct service-to-service calls)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _osvauco_agent_identity_token() -> str:
|
|
"""Hent identity token for direkte kall til osvauco-agent."""
|
|
try:
|
|
audience = os.environ.get("OSVAUCO_AGENT_URL")
|
|
if not audience:
|
|
raise ValueError("OSVAUCO_AGENT_URL environment variable is not set.")
|
|
auth_req = google.auth.transport.requests.Request()
|
|
token = google.oauth2.id_token.fetch_id_token(auth_req, audience)
|
|
if not token:
|
|
raise ValueError("Empty token returned from fetch_id_token")
|
|
return token
|
|
except Exception as e:
|
|
logger.error(f"[SERVICE-TO-SERVICE TOKEN ERROR] {type(e).__name__}: {e}")
|
|
raise
|
|
|
|
def _osvauco_agent_headers() -> dict:
|
|
return {"Authorization": f"Bearer {_osvauco_agent_identity_token()}", "Content-Type": "application/json"}
|
|
|
|
async def _osvauco_agent_post(path: str, body: dict) -> Any:
|
|
"""Kaller osvauco-agent direkte med service-to-service auth."""
|
|
url = f"{os.environ.get('OSVAUCO_AGENT_URL')}{path}"
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.post(url, json=body, headers=_osvauco_agent_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Ollama helpers — direkte mot emma-gpu-vm
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _ollama_chat(model: str, prompt: str, system: str = "") -> dict:
|
|
"""Kall Ollama chat-API direkte på emma-gpu-vm."""
|
|
messages = []
|
|
if system:
|
|
messages.append({"role": "system", "content": system})
|
|
messages.append({"role": "user", "content": prompt})
|
|
|
|
payload = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"stream": False,
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=120) as c:
|
|
r = await c.post(f"{OLLAMA_BASE_URL}/api/chat", json=payload)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
return {
|
|
"model": model,
|
|
"response": data.get("message", {}).get("content", ""),
|
|
"done": data.get("done", True),
|
|
"total_duration_ms": round(data.get("total_duration", 0) / 1e6),
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"[OLLAMA ERROR] model={model} {type(e).__name__}: {e}")
|
|
raise
|
|
|
|
|
|
async def _ollama_models() -> list:
|
|
"""List alle modeller tilgjengelig på emma-gpu-vm."""
|
|
async with httpx.AsyncClient(timeout=10) as c:
|
|
r = await c.get(f"{OLLAMA_BASE_URL}/api/tags")
|
|
r.raise_for_status()
|
|
return r.json().get("models", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gitea helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _gitea_headers() -> dict:
|
|
return {"Authorization": f"token {GITEA_TOKEN}", "Content-Type": "application/json", "Accept": "application/json"}
|
|
|
|
|
|
async def _gitea_get(path: str) -> Any:
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.get(f"{GITEA_URL}/api/v1{path}", headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def _gitea_post(path: str, body: dict) -> Any:
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.post(f"{GITEA_URL}/api/v1{path}", json=body, headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def _gitea_put(path: str, body: dict) -> Any:
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.put(f"{GITEA_URL}/api/v1{path}", json=body, headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool implementations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def get_billing_summary(p): return await _opax_get("/billing/summary")
|
|
async def get_billing_forecast(p): return await _opax_get("/billing/forecast")
|
|
async def get_billing_credits(p): return await _opax_get("/billing/credits")
|
|
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 get_billing_tokens_by_module(p): return await _opax_get("/telemetry/history")
|
|
async def set_billing_budget(p):
|
|
return await _opax_post("/billing/budget", {"budget": p.get("amount", 500)})
|
|
|
|
async def create_invite(p):
|
|
return await _opax_post("/onboard/invite", {
|
|
"company": p.get("company", p.get("name", "")),
|
|
"email": p.get("email"),
|
|
"tier": p.get("tier", "starter"),
|
|
})
|
|
|
|
async def list_customers(p): return await _opax_get("/state")
|
|
async def send_webhook(p):
|
|
return await _opax_post("/notify/webhook", {
|
|
"url": p.get("url", ""), "event": p.get("event", "custom"),
|
|
"title": p.get("title", "OPAX varsel"), "body": p.get("message", p.get("body", "")),
|
|
})
|
|
|
|
async def send_email(p):
|
|
return await _opax_post("/notify/email", {"to": p.get("to"), "subject": p.get("subject"), "event": p.get("event", "digest")})
|
|
|
|
async def send_sms(p):
|
|
return await _osvauco_agent_post("/notify/sms", {"to": p.get("to"), "body": p.get("message", p.get("body", "")), "event": p.get("event", "spike"), "tier": p.get("tier", "guard")})
|
|
|
|
async def get_notify_channels(p): return await _opax_get("/notify/channels")
|
|
|
|
async def run_jason(p):
|
|
return await _osvauco_agent_post("/run", {"message": p.get("prompt", p.get("message", "")), "user_id": p.get("user_id", "opax"), "session_id": p.get("session_id", "mcp"), "mode": p.get("mode", "light")})
|
|
|
|
async def run_emma(p):
|
|
"""Emma Vauger — Gemma 3 27B via Ollama på emma-gpu-vm. Primær lokal AI."""
|
|
return await _ollama_chat(
|
|
model=EMMA_MODEL,
|
|
prompt=p.get("prompt", p.get("message", "")),
|
|
system=p.get("system", "Du er Emma Vauger, en intelligent og hjelpsom AI-assistent for Vauco. Du er lokal, rask og priveråd.")
|
|
)
|
|
|
|
async def run_emma_fast(p):
|
|
"""Emma Fast — Gemma 3 4B via Ollama. Rask inferens for enkle oppgaver."""
|
|
return await _ollama_chat(
|
|
model=EMMA_FAST_MODEL,
|
|
prompt=p.get("prompt", p.get("message", "")),
|
|
system=p.get("system", "Du er en rask og konsis AI-assistent. Svar kort og presist.")
|
|
)
|
|
|
|
async def run_qwen(p):
|
|
"""Qwen 2.5 3B via Ollama. Lett hjelper for enkle spørsmål og kodeoppgaver."""
|
|
return await _ollama_chat(
|
|
model=EMMA_LIGHT_MODEL,
|
|
prompt=p.get("prompt", p.get("message", "")),
|
|
system=p.get("system", "")
|
|
)
|
|
|
|
async def list_emma_models(p):
|
|
"""List alle Ollama-modeller tilgjengelig på emma-gpu-vm."""
|
|
return await _ollama_models()
|
|
|
|
async def get_health(p): return await _opax_get("/health")
|
|
async def get_build_status(p): return await _opax_get("/opax/build-status")
|
|
async def get_state(p): return await _opax_get("/state")
|
|
async def get_telemetry(p): return await _opax_get("/telemetry/history")
|
|
|
|
async def run_terminal(p):
|
|
return await _opax_post("/terminal/exec", {"cmd": p.get("command", p.get("cmd", "help"))})
|
|
|
|
async def tui_command(p):
|
|
"""Kjør shell-kommando via TUI-bro."""
|
|
return await _opax_post("/tui-command", p)
|
|
|
|
async def list_commits(p):
|
|
return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/commits?limit={p.get('limit', 10)}")
|
|
|
|
async def get_file(p):
|
|
return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/contents/{p.get('path', '')}?ref={p.get('ref', 'main')}")
|
|
|
|
async def list_open_issues(p):
|
|
return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/issues?state=open&limit=20")
|
|
|
|
async def create_issue(p):
|
|
return await _gitea_post(f"/repos/{p.get('repo', GITEA_REPO)}/issues", {"title": p.get("title"), "body": p.get("body", "")})
|
|
|
|
async def push_file(p):
|
|
repo = p.get("repo", GITEA_REPO)
|
|
path = p.get("path")
|
|
content = base64.b64encode(p.get("content", "").encode()).decode()
|
|
body = {
|
|
"message": p.get("message", f"chore: update {path} via opax-mcp"),
|
|
"content": content,
|
|
"branch": p.get("branch", "main"),
|
|
}
|
|
if not p.get("sha"):
|
|
try:
|
|
existing = await _gitea_get(f"/repos/{repo}/contents/{path}?ref={p.get('branch','main')}")
|
|
body["sha"] = existing["sha"]
|
|
except Exception:
|
|
pass
|
|
else:
|
|
body["sha"] = p["sha"]
|
|
if "sha" in body:
|
|
return await _gitea_put(f"/repos/{repo}/contents/{path}", body)
|
|
return await _gitea_post(f"/repos/{repo}/contents/{path}", body)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool registry + MCP schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TOOLS = {
|
|
"get_billing_summary": (get_billing_summary, "Hent billing-sammendrag for OPAX", {}),
|
|
"get_billing_forecast": (get_billing_forecast, "Hent billing-prognose", {}),
|
|
"get_billing_credits": (get_billing_credits, "Hent gjenværende kreditter", {}),
|
|
"get_billing_anomalies": (get_billing_anomalies, "Hent billing-anomalier", {}),
|
|
"get_billing_history": (get_billing_history, "Hent billing-historikk", {}),
|
|
"get_billing_budget": (get_billing_budget, "Hent nåværende budsjett", {}),
|
|
"get_billing_tokens_by_module":(get_billing_tokens_by_module,"Hent token-forbruk per modul", {}),
|
|
"set_billing_budget": (set_billing_budget, "Sett månedlig budsjett", {"type":"object","properties":{"amount":{"type":"number","description":"Budsjett i NOK"}},"required":[]}),
|
|
"create_invite": (create_invite, "Inviter ny kunde til OPAX", {"type":"object","properties":{"company":{"type":"string"},"email":{"type":"string"},"tier":{"type":"string"}},"required":["email"]}),
|
|
"list_customers": (list_customers, "List alle kunder", {}),
|
|
"send_webhook": (send_webhook, "Send webhook-varsling", {"type":"object","properties":{"url":{"type":"string"},"title":{"type":"string"},"message":{"type":"string"}},"required":["url"]}),
|
|
"send_email": (send_email, "Send e-post", {"type":"object","properties":{"to":{"type":"string"},"subject":{"type":"string"}},"required":["to","subject"]}),
|
|
"send_sms": (send_sms, "Send SMS", {"type":"object","properties":{"to":{"type":"string"},"message":{"type":"string"}},"required":["to","message"]}),
|
|
"get_notify_channels": (get_notify_channels, "Hent varslingkanaler", {}),
|
|
"run_jason": (run_jason, "Kjør Jason-agenten med en prompt", {"type":"object","properties":{"prompt":{"type":"string"},"mode":{"type":"string"}},"required":["prompt"]}),
|
|
|
|
# — Emma Vauger — direkte Ollama på emma-gpu-vm (34.13.238.133:11434)
|
|
"run_emma": (run_emma, "Emma Vauger (gemma3:27b) — primær lokal AI, ingen quota", {"type":"object","properties":{"prompt":{"type":"string"},"system":{"type":"string","description":"Valgfri systemprompt"}},"required":["prompt"]}),
|
|
"run_emma_fast": (run_emma_fast, "Emma Fast (gemma3:4b) — rask versjon for enkle oppgaver", {"type":"object","properties":{"prompt":{"type":"string"},"system":{"type":"string"}},"required":["prompt"]}),
|
|
"run_qwen": (run_qwen, "Qwen 2.5 3B — lett hjelper og kodeoppgaver", {"type":"object","properties":{"prompt":{"type":"string"},"system":{"type":"string"}},"required":["prompt"]}),
|
|
"list_emma_models": (list_emma_models, "List alle Ollama-modeller tilgjengelig på emma-gpu-vm", {}),
|
|
|
|
"get_health": (get_health, "Hent helsestatus for OPAX", {}),
|
|
"get_build_status": (get_build_status, "Hent siste build-status", {}),
|
|
"get_state": (get_state, "Hent platform-tilstand", {}),
|
|
"get_telemetry": (get_telemetry, "Hent telemetri-historikk", {}),
|
|
"tui_command": (tui_command, "Kjør shell-kommando via TUI-bro og få stdout/stderr/exit_code tilbake", {"type":"object","properties":{"command":{"type":"string","description":"Shell-kommando å kjøre"},"timeout":{"type":"integer","description":"Timeout i sekunder, default 30"}},"required":["command"]}),
|
|
"run_terminal": (run_terminal, "Kjør terminalkommando på VM", {"type":"object","properties":{"command":{"type":"string"}},"required":["command"]}),
|
|
"list_commits": (list_commits, "List siste commits i Gitea-repo", {}),
|
|
"get_file": (get_file, "Hent fil fra Gitea-repo", {"type":"object","properties":{"path":{"type":"string"}},"required":["path"]}),
|
|
"list_open_issues": (list_open_issues, "List åpne issues i Gitea-repo", {}),
|
|
"create_issue": (create_issue, "Opprett nytt issue i Gitea-repo", {"type":"object","properties":{"title":{"type":"string"},"body":{"type":"string"}},"required":["title"]}),
|
|
"push_file": (push_file, "Opprett eller oppdater fil i Gitea-repo", {"type":"object","properties":{"path":{"type":"string"},"content":{"type":"string"},"message":{"type":"string"}},"required":["path","content"]}),
|
|
}
|
|
|
|
|
|
def _tool_list_result():
|
|
return [
|
|
{
|
|
"name": name,
|
|
"description": desc,
|
|
"inputSchema": schema if schema else {"type": "object", "properties": {}},
|
|
}
|
|
for name, (_, desc, schema) in TOOLS.items()
|
|
]
|
|
|
|
|
|
def _jsonrpc_ok(req_id, result):
|
|
return {"jsonrpc": "2.0", "id": req_id, "result": result}
|
|
|
|
|
|
def _jsonrpc_err(req_id, code, message):
|
|
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP Streamable HTTP endpoint — POST /
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.post("/")
|
|
async def mcp_handler(request: Request):
|
|
_verify_auth(request)
|
|
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
return JSONResponse(_jsonrpc_err(None, -32700, "Parse error"), status_code=400)
|
|
|
|
method = body.get("method", "")
|
|
req_id = body.get("id")
|
|
params = body.get("params", {})
|
|
|
|
if method == "initialize":
|
|
return JSONResponse(_jsonrpc_ok(req_id, {
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": {"name": "opax-mcp", "version": "3.2.0"},
|
|
}))
|
|
|
|
if method == "tools/list":
|
|
return JSONResponse(_jsonrpc_ok(req_id, {"tools": _tool_list_result()}))
|
|
|
|
if method == "tools/call":
|
|
tool_name = params.get("name") or params.get("tool")
|
|
tool_args = params.get("arguments", params.get("params", {}))
|
|
|
|
entry = TOOLS.get(tool_name)
|
|
if not entry:
|
|
return JSONResponse(_jsonrpc_err(req_id, -32601, f"Unknown tool: {tool_name}"))
|
|
|
|
handler, _, _ = entry
|
|
try:
|
|
result = await handler(tool_args)
|
|
except Exception as e:
|
|
logger.error(f"[MCP HANDLER ERROR] {type(e).__name__}: {e}")
|
|
return JSONResponse(_jsonrpc_err(req_id, -32000, str(e)))
|
|
|
|
return JSONResponse(_jsonrpc_ok(req_id, {
|
|
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]
|
|
}))
|
|
|
|
if method.startswith("notifications/"):
|
|
return JSONResponse(status_code=202, content={})
|
|
|
|
return JSONResponse(_jsonrpc_err(req_id, -32601, f"Method not found: {method}"), status_code=404)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Health (keepalive)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok", "service": "opax-mcp", "version": "3.2.0", "ollama": OLLAMA_BASE_URL, "emma_model": EMMA_MODEL}
|