chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

CI6: fix _opax_headers() — bruk identity token (metadata server) istedet for access token for Cloud Run auth

Browse Source
This commit is contained in:
chrischristiansen-glitch 2026-06-10 07:39:36 +02:00
parent 3bb95c8af9
commit 8013bd8b6f
1 changed files with 33 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
agents/tools/mcp_tools.py
View File

@ -19,6 +19,7 @@ from google.adk.tools.mcp_tool.mcp_toolset import (
)
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
OPAX_MCP_URL = os.environ.get("MCP_SERVER_URL", "https://opax-mcp-zjbqp3prqq-uc.a.run.app")
# --- BigQuery MCP (Google managed, OAuth) ---
BIGQUERY_MCP_URL = f"https://bigquery.googleapis.com/mcp/projects/{PROJECT_ID}"
@ -59,19 +60,42 @@ def get_maps_mcp_toolset() -> MCPToolset:
# Bruker httpx over REST i stedet for MCPToolset siden opax-mcp er en
# vanlig FastAPI-server, ikke en MCP streamable-server med session-handshake.
OPAX_MCP_URL = os.environ.get("MCP_SERVER_URL", "https://opax-mcp-zjbqp3prqq-uc.a.run.app")
def _opax_identity_token() -> str:
"""
Hent Cloud Run identity token fra GCE metadata server.
Kjører på Cloud Run / GCE — audience må matche opax-mcp sin URL.
Fallback til ADC access token for lokal utvikling.
"""
import httpx
metadata_url = (
"http://metadata.google.internal/computeMetadata/v1/instance"
f"/service-accounts/default/identity?audience={OPAX_MCP_URL}&format=full"
)
try:
resp = httpx.get(
metadata_url,
headers={"Metadata-Flavor": "Google"},
timeout=5,
)
if resp.status_code == 200 and resp.text.strip():
return resp.text.strip()
except Exception:
pass
# Fallback: ADC access token (lokal dev)
credentials, _ = google.auth.default()
credentials.refresh(google.auth.transport.requests.Request())
return credentials.token
def _opax_headers() -> dict:
"""Bygg auth-headers for opax-mcp: Cloud Run identity token + MCP-nøkkel."""
import google.auth.transport.requests
credentials, _ = google.auth.default()
credentials.refresh(google.auth.transport.requests.Request())
return {
"Authorization": f"Bearer {credentials.token}",
"Authorization": f"Bearer {_opax_identity_token()}",
"X-MCP-Key": os.environ.get("MCP_SECRET", ""),
"Content-Type": "application/json",
}
def push_static(file_path: str, content: str, content_type: str = "text/html; charset=utf-8") -> dict:
"""Last opp en statisk fil (HTML/CSS/JS) til opax.vauco.no via GCS. file_path er relativ sti, f.eks. static/jason.html."""
import httpx
@ -81,6 +105,7 @@ def push_static(file_path: str, content: str, content_type: str = "text/html; ch
json={"file_path": file_path, "content": content, "content_type": content_type},
timeout=30,
)
resp.raise_for_status()
return resp.json()
def get_build_status() -> dict:
@ -91,6 +116,7 @@ def get_build_status() -> dict:
headers=_opax_headers(),
timeout=15,
)
resp.raise_for_status()
return resp.json()
def get_logs(lines: int = 50, severity: str = "DEFAULT") -> dict:
@ -102,6 +128,7 @@ def get_logs(lines: int = 50, severity: str = "DEFAULT") -> dict:
params={"lines": lines, "severity": severity},
timeout=15,
)
resp.raise_for_status()
return resp.json()
def deploy_service(branch: str = "main", trigger_id: str = "") -> dict:
@ -113,6 +140,7 @@ def deploy_service(branch: str = "main", trigger_id: str = "") -> dict:
json={"branch": branch, "trigger_id": trigger_id},
timeout=15,
)
resp.raise_for_status()
return resp.json()
def get_opax_tools() -> list: