OSVauco/agents/core-logic/app.py

# agents/core-logic/app.py
# OSVauco-NMTMD-GCOS — FastAPI HTTP entrypoint for Cloud Run

import os
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__)

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")


@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


async def _ensure_session(user_id: str, session_id: str):
    """Await get_session; create if missing or raises."""
    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


@app.get("/health")
async def health():
    return JSONResponse({"status": "ok", "service": APP_NAME})


@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))


@app.get("/billing/tokens/summary")
async def billing_tokens_summary():
    """
    CG3e — Aggreger LLM token-bruk og estimert kostnad per agent siste 30 dager.
    Returnerer JSON-liste sortert etter total_cost DESC.
    """
    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
        """

        results = client.query(query).result()
        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 results
        ]
        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))

@app.get("/billing/recommendations")
async def billing_recommendations(budget: float = 500.0):
    """
    CG4 — Kjører anbefalings- og anomali-motoren.
    Returnerer en liste med anbefalinger og estimert besparelse.
    """
    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))