Skip to main content

​
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

FeatureSync ClientAsyncClient
HTTP libraryrequestshttpx
BatchingBackground threadasyncio task
Context managerclose() / atexitasync 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.