Telemetry client
import fact0
client = fact0.Client(api_key=os.environ["FACT0_API_KEY"])
tel = client.telemetry
Telemetry requires the same write-scoped f0_live_* API key as the audit client.
Start an execution
exec = tel.start_execution(
agent_id="research-bot",
agent_name="Research Bot",
trigger="user_query",
metadata={"user_id": "u_123"},
)
print(exec["id"])
Context managers (recommended)
with client.telemetry.execution(agent_id="research-bot") as ex:
# 1. Record an LLM call with prompt and session metadata
with ex.span("gpt-4o", span_type="MODEL_INVOCATION") as span:
span.complete(
model_invocation={
"model_name": "gpt-4o",
"model_provider": "openai",
"prompt_tokens": 1500,
"completion_tokens": 350,
"total_tokens": 1850,
"temperature": 0.7,
# Observability Turn & Prompt Catalog metadata:
"session_id": "session_9a2f1b",
"turn_sequence": 2,
"prompt_name": "customer-inquiry",
"prompt_version": 3,
"cost_usd": 0.00925,
}
)
# 2. Record a tool call
with ex.span("tool.search", span_type="TOOL_CALL") as span:
span.log_event("query", {"q": "fact0"})
span.complete(output={"hits": 5})
# execution auto-ended with COMPLETED status
Span types
TOOL_CALL, MODEL_INVOCATION, STATE_MUTATION, HUMAN_APPROVAL, POLICY_EVALUATION, CUSTOM
Read methods
| Method | REST |
|---|
list_executions(agent_id=, status=, page_size=, offset=) | GET /api/v1/executions |
get_execution(id) | GET /api/v1/executions/{id} |
get_spans(execution_id) | GET /api/v1/executions/{id}/spans |
get_dag(execution_id) | GET /api/v1/executions/{id}/dag |
replay(execution_id, from_sequence=0, to_sequence=0) | GET /api/v1/executions/{id}/replay |
get_span(span_id) | GET /api/v1/spans/{id} |
Low-level ingest
| Method | REST |
|---|
ingest_spans(execution_id, spans) | POST /api/v1/executions/{id}/spans |
ingest_events(execution_id, events) | POST /api/v1/executions/{id}/events |
end_execution(execution_id, status) | PUT /api/v1/executions/{id}/end |
Status values: RUNNING, COMPLETED, FAILED, CANCELLED.