# agents/core-logic/app.py # OSVauco-NMTMD-GCOS — FastAPI HTTP entrypoint for Cloud Run import os import sys import logging from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from pydantic import BaseModel logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Gjor ml-pakken tilgjengelig uansett cwd sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) try: from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai.types import Content, Part from agent import root_agent from agents.recommendations_engine import get_recommendations except ImportError as e: logger.error(f"Failed to import ADK dependencies: {e}") raise session_service = InMemorySessionService() APP_NAME = os.environ.get("CLOUD_RUN_SERVICE", "gcp-orchestrator") PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") BQ_BILLING_DATASET = os.environ.get("BQ_BILLING_DATASET", "billing_data") # CG6: in-memory budget store (TODO Fase-C: migrate to Firestore per-user) _budget_store: dict = {} # key: user_email | "default" @asynccontextmanager async def lifespan(app: FastAPI): logger.info(f"OSVauco agent '{APP_NAME}' starting up") yield logger.info(f"OSVauco agent '{APP_NAME}' shutting down") app = FastAPI( title="OSVauco GCP Agent", description="ADK-based multi-agent orchestrator on Cloud Run", version="1.0.0", lifespan=lifespan, ) class RunRequest(BaseModel): user_id: str session_id: str message: str class RunResponse(BaseModel): user_id: str session_id: str response: str class BudgetRequest(BaseModel): budget: float user: str = "default" async def _ensure_session(user_id: str, session_id: str): try: session = await session_service.get_session( app_name=APP_NAME, user_id=user_id, session_id=session_id ) if session is not None: return session except Exception: pass session = await session_service.create_session( app_name=APP_NAME, user_id=user_id, session_id=session_id ) logger.info(f"Created new session: {session_id} for user: {user_id}") return session # ── helse ──────────────────────────────────────────────────────────────────── @app.get("/health") async def health(): return JSONResponse({"status": "ok", "service": APP_NAME}) # ── agent run ──────────────────────────────────────────────────────────────── @app.post("/run", response_model=RunResponse) async def run(req: RunRequest): try: await _ensure_session(req.user_id, req.session_id) runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service) user_content = Content(role="user", parts=[Part(text=req.message)]) final_response = "" async for event in runner.run_async( user_id=req.user_id, session_id=req.session_id, new_message=user_content ): if event.is_final_response() and event.content: for part in event.content.parts: if part.text: final_response += part.text logger.info(f"[{req.user_id}/{req.session_id}] Response length: {len(final_response)}") return RunResponse(user_id=req.user_id, session_id=req.session_id, response=final_response) except Exception as e: logger.error(f"Agent run failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ── billing: tokens ────────────────────────────────────────────────────────── @app.get("/billing/tokens/summary") async def billing_tokens_summary(): try: from google.cloud import bigquery client = bigquery.Client(project=PROJECT_ID) query = f""" SELECT agent_name, model_name, SUM(input_tokens) AS total_input_tokens, SUM(output_tokens) AS total_output_tokens, SUM(total_tokens) AS total_tokens, SUM(estimated_cost_usd) AS total_cost_usd FROM `{PROJECT_ID}.{BQ_BILLING_DATASET}.llm_token_usage` WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) GROUP BY agent_name, model_name ORDER BY total_cost_usd DESC """ rows = [ { "agent_name": row.agent_name, "model_name": row.model_name, "total_input_tokens": row.total_input_tokens, "total_output_tokens": row.total_output_tokens, "total_tokens": row.total_tokens, "total_cost_usd": round(float(row.total_cost_usd), 6), } for row in client.query(query).result() ] return JSONResponse({"period_days": 30, "rows": rows}) except Exception as e: logger.error(f"[/billing/tokens/summary] Failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ── billing: anbefalinger ──────────────────────────────────────────────────── @app.get("/billing/recommendations") async def billing_recommendations(budget: float = 500.0): try: recommendations = get_recommendations(budget) return JSONResponse(recommendations) except Exception as e: logger.error(f"[/billing/recommendations] Failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ── billing: tjenester med SKU-detaljer (CG5) ──────────────────────────────── @app.get("/billing/by-service") async def billing_by_service(days: int = 30): try: from ml.billing_agent import BillingAgent agent = BillingAgent() return JSONResponse(agent.get_service_totals(days)) except Exception as e: logger.error(f"[/billing/by-service] Failed: {e}", exc_info=True) return JSONResponse([]) # fallback: tom liste, dashbordet krasjer ikke # ── billing: anomalier (CG6) ───────────────────────────────────────────────── @app.get("/billing/anomalies") async def billing_anomalies(): """ CG6 — Anomalideteksjon: sammenligner dagens kostnad mot 7-dagers snitt per tjeneste. Returnerer tjenester der dagens kostnad > 2x snittet. Fallback: tom liste ved feil slik at dashbordet ikke krasjer. """ try: from ml.billing_agent import BillingAgent agent = BillingAgent() return JSONResponse(agent.get_anomalies()) except Exception as e: logger.error(f"[/billing/anomalies] Failed: {e}", exc_info=True) return JSONResponse({"anomalies": []}) # ── billing: historikk for bar-chart (CG6) ─────────────────────────────────── @app.get("/billing/history") async def billing_history(days: int = 30): """ CG6 — Daglig MTD-historikk for bar-chart i dashbordet. Returnerer liste: [{date, mtd}] sortert ASC. """ try: from ml.billing_agent import BillingAgent agent = BillingAgent() return JSONResponse({"history": agent.get_daily_history(days)}) except Exception as e: logger.error(f"[/billing/history] Failed: {e}", exc_info=True) return JSONResponse({"history": []}) # ── billing: summary ───────────────────────────────────────────────────────── @app.get("/billing/summary") async def billing_summary(): try: from ml.billing_agent import BillingAgent agent = BillingAgent() return JSONResponse(agent.get_summary()) except Exception as e: logger.error(f"[/billing/summary] Failed: {e}", exc_info=True) return JSONResponse({"summary": []}) # ── billing: live / forecast ───────────────────────────────────────────────── @app.get("/billing/live") async def billing_live(): try: from ml.billing_agent import BillingAgent agent = BillingAgent() return JSONResponse(agent.get_forecast()) except Exception as e: logger.error(f"[/billing/live] Failed: {e}", exc_info=True) return JSONResponse({}) # ── billing: budsjett GET + POST (CG6) ─────────────────────────────────────── # TODO Fase-C: erstatt _budget_store med Firestore-dokument per bruker @app.get("/billing/budget") async def get_budget(user: str = "default"): """ CG6 — Hent lagret budsjett for bruker. In-memory; nullstilles ved redeploy (Firestore i Fase C). """ budget = _budget_store.get(user, 500.0) return JSONResponse({"budget": budget, "user": user}) @app.post("/billing/budget") async def set_budget(req: BudgetRequest): """ CG6 — Lagre budsjett for bruker. In-memory; nullstilles ved redeploy (Firestore i Fase C). """ if req.budget <= 0: raise HTTPException(status_code=400, detail="Budget must be > 0") _budget_store[req.user] = req.budget logger.info(f"Budget updated: {req.user} -> {req.budget}") return JSONResponse({"budget": req.budget, "user": req.user})