# agents/mcp_server/server.py # OPAX-MCP — Vauco sin egen MCP-server / CI-kanal mot GCP # CI1a: push_static, get_build_status, deploy_service # Deploy: Cloud Run `opax-mcp` (separat fra osvauco-agent) import os import json import logging from contextlib import asynccontextmanager from typing import Any from fastapi import FastAPI, HTTPException, Depends, Header from fastapi.responses import JSONResponse from pydantic import BaseModel logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") REGION = os.environ.get("REGION", "us-central1") STATIC_BUCKET = os.environ.get("STATIC_BUCKET", "opax-vauco-static") CLOUD_RUN_SERVICE = os.environ.get("CLOUD_RUN_SERVICE", "osvauco-agent") MCP_SECRET = os.environ.get("MCP_SECRET", "") # Secret Manager: mcp-server-key # ── Auth ─────────────────────────────────────────────────────────────────────────── def verify_token(x_mcp_key: str = Header(default="")): """Enkel API-nøkkel auth. Byttes ut med IAP når tjenesten er på Cloud Run bak LB.""" if MCP_SECRET and x_mcp_key != MCP_SECRET: raise HTTPException(status_code=401, detail="Ugyldig MCP-nøkkel") return True # ── Models ────────────────────────────────────────────────────────────────────────── class PushStaticRequest(BaseModel): file_path: str # Lokal sti relativt til repo-rot, f.eks. "static/jason.html" content: str # Fil-innhold (tekst/HTML) content_type: str = "text/html; charset=utf-8" cache_control: str = "public, max-age=300" class DeployServiceRequest(BaseModel): branch: str = "main" # Branch å bygge fra trigger_id: str = "" # Cloud Build trigger ID (tom = hent fra env) substitutions: dict = {} class ToolCallRequest(BaseModel): """MCP-standard tool call envelope.""" tool: str params: dict = {} # ── Lifespan ────────────────────────────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): logger.info(f"OPAX-MCP starting | project={PROJECT_ID} | bucket={STATIC_BUCKET}") yield logger.info("OPAX-MCP shutting down") app = FastAPI( title="OPAX-MCP", description="Vauco sin egen MCP-server — CI/CD-kanal mot GCP", version="1.0.0", lifespan=lifespan, ) # ── Health ───────────────────────────────────────────────────────────────────────────── @app.get("/health") async def health(): return {"status": "ok", "service": "opax-mcp", "project": PROJECT_ID} # ── MCP Tool Registry ────────────────────────────────────────────────────────────────── @app.get("/mcp/tools") async def list_tools(_: bool = Depends(verify_token)): """MCP tool manifest — returnerer alle tilgjengelige verktøy med schema.""" return { "tools": [ { "name": "push_static", "description": "Last opp en statisk fil (HTML/CSS/JS) direkte til GCS-bucketen som serverer opax.vauco.no. Erstatter manuell gsutil-upload.", "parameters": { "file_path": {"type": "string", "description": "Relativ sti, f.eks. static/jason.html"}, "content": {"type": "string", "description": "Fil-innhold som tekst"}, "content_type": {"type": "string", "default": "text/html; charset=utf-8"}, "cache_control": {"type": "string", "default": "public, max-age=300"}, }, "required": ["file_path", "content"], }, { "name": "get_build_status", "description": "Hent status på siste Cloud Build-kjøring for prosjektet. Returnerer status, commit, varighet og logg-URL.", "parameters": {}, "required": [], }, { "name": "deploy_service", "description": "Trigger Cloud Build for å bygge og deploye osvauco-agent til Cloud Run. Kan spesifisere branch og substitutions.", "parameters": { "branch": {"type": "string", "default": "main"}, "trigger_id": {"type": "string", "description": "Cloud Build trigger ID (valgfri — bruker env-default hvis tom)"}, "substitutions": {"type": "object", "default": {}}, }, "required": [], }, { "name": "get_logs", "description": "Hent siste Cloud Run-logger for osvauco-agent.", "parameters": { "lines": {"type": "integer", "default": 50, "description": "Antall logglinjer å returnere"}, "severity": {"type": "string", "default": "DEFAULT", "description": "ERROR | WARNING | INFO | DEFAULT"}, }, "required": [], }, ] } # ── Tool: push_static ───────────────────────────────────────────────────────────────────── @app.post("/tools/push_static") async def push_static(req: PushStaticRequest, _: bool = Depends(verify_token)): """ Last opp statisk fil til GCS. Fil-sti mappes direkte: static/jason.html → gs://{STATIC_BUCKET}/static/jason.html """ try: from google.cloud import storage client = storage.Client(project=PROJECT_ID) bucket = client.bucket(STATIC_BUCKET) blob = bucket.blob(req.file_path) blob.cache_control = req.cache_control blob.upload_from_string( req.content.encode("utf-8"), content_type=req.content_type, ) public_url = f"https://storage.googleapis.com/{STATIC_BUCKET}/{req.file_path}" logger.info(f"[push_static] Uploaded {req.file_path} ({len(req.content)} bytes)") return { "ok": True, "file_path": req.file_path, "bucket": STATIC_BUCKET, "bytes": len(req.content.encode("utf-8")), "url": public_url, } except Exception as e: logger.error(f"[push_static] Failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ── Tool: get_build_status ───────────────────────────────────────────────────────────────────── @app.get("/tools/get_build_status") async def get_build_status(_: bool = Depends(verify_token)): """ Hent siste Cloud Build-kjøring fra us-central1. """ try: from google.cloud.devtools import cloudbuild_v1 from google.api_core.client_options import ClientOptions client = cloudbuild_v1.CloudBuildClient( client_options=ClientOptions( api_endpoint=f"{REGION}-cloudbuild.googleapis.com" ) ) request = cloudbuild_v1.ListBuildsRequest( project_id=PROJECT_ID, filter='trigger_id!=""', page_size=5, ) builds = list(client.list_builds(request=request)) if not builds: return {"status": "unknown", "message": "Ingen builds funnet"} b = builds[0] status_map = {1:"queued",2:"working",3:"success",4:"failure", 5:"internal_error",6:"timeout",7:"cancelled"} status_str = status_map.get(int(b.status), "unknown") import time duration_s = None if b.start_time and b.finish_time: duration_s = int(b.finish_time.seconds - b.start_time.seconds) elif b.start_time: duration_s = int(time.time() - b.start_time.seconds) return { "status": status_str, "build_id": b.id, "branch": (b.substitutions or {}).get("BRANCH_NAME", "main"), "commit": (b.substitutions or {}).get("SHORT_SHA", ""), "duration_s": duration_s, "log_url": b.log_url or "", "region": REGION, } except Exception as e: logger.error(f"[get_build_status] Failed: {e}", exc_info=True) return {"status": "error", "message": str(e)} # ── Tool: deploy_service ────────────────────────────────────────────────────────────────────── @app.post("/tools/deploy_service") async def deploy_service(req: DeployServiceRequest, _: bool = Depends(verify_token)): """ Trigger Cloud Build for å bygge + deploye osvauco-agent. HITL-gate: logger kallet og returnerer build-ID for oppfølging. """ try: from google.cloud.devtools import cloudbuild_v1 from google.api_core.client_options import ClientOptions client = cloudbuild_v1.CloudBuildClient( client_options=ClientOptions( api_endpoint=f"{REGION}-cloudbuild.googleapis.com" ) ) trigger_id = req.trigger_id or os.environ.get("CLOUD_BUILD_TRIGGER_ID", "") if not trigger_id: raise HTTPException( status_code=400, detail="trigger_id må settes i request eller CLOUD_BUILD_TRIGGER_ID env" ) subs = {"BRANCH_NAME": req.branch, **req.substitutions} response = client.run_build_trigger( project_id=PROJECT_ID, trigger_id=trigger_id, source=cloudbuild_v1.RepoSource( branch_name=req.branch, substitutions=subs, ), ) build_id = response.metadata.build.id if hasattr(response, 'metadata') else "ukjent" logger.info(f"[deploy_service] Triggered build {build_id} from branch {req.branch}") return { "ok": True, "build_id": build_id, "branch": req.branch, "trigger_id": trigger_id, "message": "Build triggered. Kall GET /tools/get_build_status for status.", } except HTTPException: raise except Exception as e: logger.error(f"[deploy_service] Failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ── Tool: get_logs ─────────────────────────────────────────────────────────────────────────────── @app.get("/tools/get_logs") async def get_logs( lines: int = 50, severity: str = "DEFAULT", _: bool = Depends(verify_token) ): """ Hent siste Cloud Run-logger for osvauco-agent. Returnerer logglinjer som liste. """ try: from google.cloud import logging as gcloud_logging client = gcloud_logging.Client(project=PROJECT_ID) filter_str = ( f'resource.type="cloud_run_revision" ' f'resource.labels.service_name="{CLOUD_RUN_SERVICE}"' ) if severity and severity != "DEFAULT": filter_str += f' severity>={severity}' entries = list(client.list_entries( filter_=filter_str, order_by=gcloud_logging.DESCENDING, page_size=lines, )) log_lines = [] for entry in entries: payload = entry.payload if isinstance(payload, dict): text = payload.get("message", str(payload)) else: text = str(payload) log_lines.append({ "timestamp": str(entry.timestamp), "severity": str(entry.severity), "text": text[:400], }) return {"service": CLOUD_RUN_SERVICE, "count": len(log_lines), "logs": log_lines} except Exception as e: logger.error(f"[get_logs] Failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ── MCP unified tool call endpoint ──────────────────────────────────────────────────────────── @app.post("/mcp/call") async def mcp_call(req: ToolCallRequest, _: bool = Depends(verify_token)): tool = req.tool params = req.params if tool == "push_static": r = PushStaticRequest(**params) return await push_static(r, True) elif tool == "get_build_status": return await get_build_status(True) elif tool == "deploy_service": r = DeployServiceRequest(**params) return await deploy_service(r, True) elif tool == "get_logs": return await get_logs( lines=params.get("lines", 50), severity=params.get("severity", "DEFAULT"), _=True ) else: raise HTTPException(status_code=404, detail=f"Ukjent tool: {tool}. Kall GET /mcp/tools for liste.")