Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.fact0.io/llms.txt

Use this file to discover all available pages before exploring further.

Telemetry client

import ledgerflow

client = ledgerflow.Client(api_key=os.environ["LEDGERFLOW_API_KEY"])
tel = client.telemetry
Telemetry routes are currently unauthenticated for local/demo use. API-key auth is planned.

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"])
with client.telemetry.execution(agent_id="research-bot") as ex:
    with ex.span("tool.search", span_type="TOOL_CALL") as span:
        span.log_event("query", {"q": "ledgerflow"})
        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

MethodREST
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

MethodREST
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.