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

fix(CI5): bytt til google.auth REST-kall for regional Cloud Build

Browse Source
This commit is contained in:
chrischristiansen-glitch 2026-06-10 04:57:54 +02:00
parent 3af4cbb7cc
commit be2e3b4583
1 changed files with 46 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
agents/mcp_server/server.py
View File

@ -1,5 +1,5 @@
# agents/mcp_server/server.py # agents/mcp_server/server.py
# OPAX-MCP — Vauco sin egen MCP-server / CI-kanal mot GCP # OPAX-MCP — Vauco sin eigen MCP-server / CI-kanal mot GCP
import os import os
import logging import logging
@ -25,14 +25,27 @@ def verify_token(x_mcp_key: str = Header(default="")):
return True return True
def _cloudbuild_client(): def _get_access_token() -> str:
"""Returnerer CloudBuildClient koblet til regional endpoint.""" import google.auth
from google.cloud.devtools import cloudbuild_v1 import google.auth.transport.requests
from google.api_core.client_options import ClientOptions creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
endpoint = f"{REGION}-cloudbuild.googleapis.com" creds.refresh(google.auth.transport.requests.Request())
return cloudbuild_v1.CloudBuildClient( return creds.token
client_options=ClientOptions(api_endpoint=endpoint)
def _list_builds_rest(project_id: str, region: str, trigger_id: str, page_size: int = 1):
"""Kaller Cloud Build v1 REST direkte mot regional endpoint."""
import urllib.request
import json
token = _get_access_token()
url = (
f"https://cloudbuild.googleapis.com/v1/"
f"projects/{project_id}/locations/{region}/builds"
f"?filter=trigger_id%3D%22{trigger_id}%22&pageSize={page_size}"
) )
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
class PushStaticRequest(BaseModel): class PushStaticRequest(BaseModel):
@ -94,39 +107,36 @@ async def push_static(req: PushStaticRequest, _: bool = Depends(verify_token)):
@app.get("/tools/get_build_status") @app.get("/tools/get_build_status")
async def get_build_status(_: bool = Depends(verify_token)): async def get_build_status(_: bool = Depends(verify_token)):
""" """
Henter siste build fra osvauco-agent-main-trigger. Henter siste build fra osvauco-agent-main-trigger via REST mot
Bruker regional endpoint: {region}-cloudbuild.googleapis.com cloudbuild.googleapis.com/v1/projects/{p}/locations/{r}/builds
og parent: projects/{project}/locations/{region}
""" """
try: try:
from google.cloud.devtools import cloudbuild_v1 data = _list_builds_rest(PROJECT_ID, REGION, BUILD_TRIGGER_ID, page_size=1)
client = _cloudbuild_client() builds = data.get("builds", [])
parent = f"projects/{PROJECT_ID}/locations/{REGION}"
request = cloudbuild_v1.ListBuildsRequest(
parent=parent,
filter=f'trigger_id="{BUILD_TRIGGER_ID}"',
page_size=1,
)
builds = list(client.list_builds(request=request))
if not builds: if not builds:
return {"status": "unknown", "message": f"Ingen builds for trigger {BUILD_TRIGGER_ID} i {REGION}"} return {"status": "unknown", "message": f"Ingen builds for trigger {BUILD_TRIGGER_ID} i {REGION}"}
b = builds[0] b = builds[0]
status_map = {1:"queued",2:"working",3:"success",4:"failure", status_str = b.get("status", "unknown").lower()
5:"internal_error",6:"timeout",7:"cancelled"} subs = b.get("substitutions", {})
status_str = status_map.get(int(b.status), "unknown") start = b.get("startTime")
import time finish = b.get("finishTime")
duration_s = None duration_s = None
if b.finish_time and b.start_time: if start and finish:
duration_s = int(b.finish_time.seconds - b.start_time.seconds) from datetime import datetime, timezone
elif b.start_time: fmt = "%Y-%m-%dT%H:%M:%S.%fZ"
duration_s = int(time.time() - b.start_time.seconds) try:
t0 = datetime.strptime(start, fmt).replace(tzinfo=timezone.utc)
t1 = datetime.strptime(finish, fmt).replace(tzinfo=timezone.utc)
duration_s = int((t1 - t0).total_seconds())
except Exception:
pass
return { return {
"status": status_str, "status": status_str,
"build_id": b.id, "build_id": b.get("id", ""),
"branch": (b.substitutions or {}).get("BRANCH_NAME", "main"), "branch": subs.get("BRANCH_NAME", "main"),
"commit": (b.substitutions or {}).get("SHORT_SHA", ""), "commit": subs.get("SHORT_SHA", ""),
"duration_s": duration_s, "duration_s": duration_s,
"log_url": b.log_url or "", "log_url": b.get("logUrl", ""),
"region": REGION, "region": REGION,
"trigger_id": BUILD_TRIGGER_ID, "trigger_id": BUILD_TRIGGER_ID,
} }
@ -139,7 +149,10 @@ async def get_build_status(_: bool = Depends(verify_token)):
async def deploy_service(req: DeployServiceRequest, _: bool = Depends(verify_token)): async def deploy_service(req: DeployServiceRequest, _: bool = Depends(verify_token)):
try: try:
from google.cloud.devtools import cloudbuild_v1 from google.cloud.devtools import cloudbuild_v1
client = _cloudbuild_client() 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 BUILD_TRIGGER_ID trigger_id = req.trigger_id or BUILD_TRIGGER_ID
subs = {"BRANCH_NAME": req.branch, **req.substitutions} subs = {"BRANCH_NAME": req.branch, **req.substitutions}
response = client.run_build_trigger( response = client.run_build_trigger(