The Python SDK (sdk/python, package ledger_sdk) has full parity with the TypeScript reference — manual capture plus Anthropic/OpenAI auto-instrumentation — and depends on nothing outside the standard library.
Not yet published to PyPI. Vendor sdk/python/ledger_sdk into your project for now, or install it as a local/git dependency:
pip install <path-to-repo>/sdk/python
from datetime import datetime, timezone
from ledger_sdk import Ledger
ledger = Ledger(
api_url="http://127.0.0.1:4010",
ingest_key="...",
agent_id="urn:agent:research-assistant",
on_error=lambda msg: print("[ledger]", msg),
)
t0 = datetime.now(timezone.utc)
# ... do the work ...
t1 = datetime.now(timezone.utc)
ledger.record(
action_type="llm_call",
provider="anthropic",
model="claude-fable-5",
started_at=t0,
ended_at=t1,
status="success",
usage={"tokens_in": 420, "tokens_out": 180},
payloads=[
{"kind": "prompt", "content": "Summarize the policy for the client memo."},
{"kind": "response", "content": "The policy provides that…"},
],
trigger={"type": "human", "subject": "person@example.com"},
)
ledger.flush()
ledger.close()
close() stops the background flush thread after a final flush — call it at shutdown so buffered events aren't lost.
action() — time and record in one callresult = ledger.action(
{"action_type": "tool_call", "config": {"tool": "save_memo"}},
lambda: save_memo(path),
)
Records success or failure automatically and re-raises the original exception untouched on failure.
from ledger_sdk import Ledger
from ledger_sdk.instrument import wrap_anthropic
import anthropic
ledger = Ledger(api_url="http://127.0.0.1:4010", ingest_key="...", agent_id="urn:agent:research-assistant")
ledger.set_trigger("human", "person@example.com")
client = wrap_anthropic(ledger, anthropic.Anthropic())
# unmodified application code from here down — no ledger references needed
res = client.messages.create(
model="claude-fable-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Draft the client memo."}],
)
wrap_openai(ledger, client) works the same way for OpenAI's client.
span = ledger.record(action_type="search", started_at=t0, ended_at=t1, status="success")
with ledger.within(span["span_id"]):
# anything recorded in here defaults its parent_ref to the search span
client = wrap_anthropic(ledger, anthropic.Anthropic())
client.messages.create(...)
Pass parent_ref explicitly on a record() call to override the ambient parent.
blob_id = ledger.upload_blob(file_bytes, "application/pdf")
ledger.record(
action_type="data_read",
payloads=[{"kind": "artifact", "blob_ref": blob_id}],
started_at=t0, ended_at=t1, status="success",
)
provider.add_span_processor(ledger.otel_span_processor())
# or: BatchSpanProcessor(ledger.otel_exporter())
Every gen_ai.* span becomes a ledger event; other spans pass through untouched.