Function events

app/api/agentops/api/routes/v1.py:165–252

(request: Request, supabase: AsyncSupabaseClient)

@router.post("/events")
async def events(request: Request, supabase: AsyncSupabaseClient):
    try:
        api_key = request.headers.get("X-Agentops-Auth")

        sessions, data = await asyncio.gather(
            supabase.table("sessions").select("id").eq("api_key", api_key).limit(1).single().execute(),
            request.json(),
        )
        # premium_status = await get_premium_status(supabase, sessions['id'])
        premium_status = False

        session_id = data.get("session_id")
        session_ids_for_project = [session["id"] for session in sessions]
        if session_id not in session_ids_for_project:
            raise RuntimeError("Invalid API Key for session")

    except RuntimeError as e:
        message = {"message": f"/events: Error posting event: {e}"}
        logger.error(message)
        return JSONResponse(message, status_code=401)

    except InvalidModelError as e:
        message = {"message": f"/events: Invalid model while posting event: {e}"}
        return JSONResponse(message, status_code=401)

    try:
        actions = []
        llms = []
        tools = []
        errors = []
        additional_cost: Decimal | None = Decimal(0)
        additional_events = 0
        additional_prompt_tokens = 0
        additional_completion_tokens = 0
        for event in data.get("events"):
            additional_events += 1
            if event["event_type"] == "llms":
                llm = await event_handlers.handle_llms(event, premium_status, session_id)
                cost = llm.get("cost")
                if cost is not None:
                    additional_cost += Decimal(cost)
                additional_prompt_tokens += llm["prompt_tokens"]
                additional_completion_tokens += llm["completion_tokens"]
                llms.append(llm)
            elif event["event_type"] == "tools":
                tools.append(await event_handlers.handle_tools(event, session_id))
            # TODO: move into an /errors endpoint?
            elif event["event_type"] == "errors":
                errors.append(await event_handlers.handle_errors(event, session_id))
            else:
                actions.append(await event_handlers.handle_actions(event, session_id))

        if additional_cost == Decimal(0):
            additional_cost = None

        inserts = []
        if len(actions) != 0:
            inserts.append(supabase.table("actions").upsert(actions).execute())