Function v2_create_session

app/api/agentops/api/routes/v2.py:194–288

(request: Request, supabase: AsyncSupabaseClient)

@router.post("/create_session")
async def v2_create_session(request: Request, supabase: AsyncSupabaseClient):
    try:
        api_key = request.headers.get("X-Agentops-Api-Key")
        parent_key = request.headers.get("X-Agentops-Parent-Key")

        validate_uuid(api_key)

        tasks = [
            supabase.table("projects").select("id").eq("api_key", api_key).execute(),
            request.json(),
        ]

        if parent_key:
            tasks.append(supabase.table("projects").select("id").eq("api_key", parent_key).execute())
            project_response, data, project_secondary_response = await asyncio.gather(*tasks)
            project_secondary = (
                project_secondary_response.data[0] if project_secondary_response.data else None
            )
        else:
            project_response, data = await asyncio.gather(*tasks)
            project_secondary = None

        project = project_response.data[0] if project_response.data else None

        logger.debug(data)

        if project is None:
            raise RuntimeError("Invalid API Key")

        session = {
            "id": data["session"]["session_id"],
            "project_id": project["id"],
            "init_timestamp": data["session"].get("init_timestamp", None)
            or datetime.now(timezone.utc).isoformat(),
            "end_timestamp": data["session"].get("end_timestamp", None),
            "tags": data["session"].get("tags", None),
            "end_state": data["session"].get("end_state", None),
            "end_state_reason": data["session"].get("end_state_reason", None),
            "video": data["session"].get("video", None),
            "host_env": data["session"].get("host_env", None),
        }

        if project_secondary:
            session["project_id_secondary"] = project_secondary["id"]

        await supabase.table("sessions").insert(session).execute()
        await supabase.table("stats").insert({"session_id": session["id"]}).execute()
        await export.create_session(session)

        token = generate_jwt(session["id"], jwt_secret)
        logger.info(colored(f"Completed request for {session['id']}", "yellow"))
        return JSONResponse(
            {
                "status": "Success",
                "jwt": token,
                "session_url": f'{app_url}/drilldown?session_id={data["session"]["session_id"]}',
            }
        )