Async audit client

For FastAPI, asyncio agents, and other async runtimes:

import asyncio
import fact0

async def main():
    async with fact0.AsyncClient(api_key="f0_live_...") as client:
        await client.audit.log(
            actor={"id": "agent-1", "type": "agent"},
            action="tool.invoke",
            resource={"id": "run_abc", "type": "agent.execution"},
            outcome="success",
        )
        await client.audit.flush()
        events = await client.audit.list_events(page_size=10)
        print(events["total"])

asyncio.run(main())

Differences from sync client

Feature	Sync `Client`	`AsyncClient`
HTTP library	`requests`	`httpx`
Batching	Background thread	`asyncio` task
Context manager	`close()` / `atexit`	`async with`
Read/verify/export	✓	✓

Telemetry methods are also available on AsyncClient.telemetry.

Async Telemetry

When using AsyncClient, telemetry execution and span context managers are asynchronous and run completely non-blocking in the background:

import asyncio
import fact0

async def main():
    async with fact0.AsyncClient(api_key="f0_live_...") as client:
        # Asynchronous execution tracing context manager
        async with client.telemetry.execution(agent_id="research-bot") as ex:
            # Asynchronous span context manager
            async with ex.span("tool.search", span_type="TOOL_CALL") as span:
                await span.log_event("query", {"q": "fact0"})
                await span.complete(output={"hits": 5})

asyncio.run(main())

FastAPI lifespan pattern

from contextlib import asynccontextmanager
from fastapi import FastAPI
import fact0

audit_client: fact0.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global audit_client
    audit_client = fact0.AsyncClient(api_key="...")
    yield
    await audit_client.close()

app = FastAPI(lifespan=lifespan)

See the FastAPI integration for middleware setup.