El SDK de Python (sdk/python, paquete ledger_sdk) tiene paridad completa con la referencia de TypeScript — captura manual más auto-instrumentación de Anthropic/OpenAI — y no depende de nada fuera de la librería estándar.
Aún no está publicado en PyPI. Por ahora, incorpore sdk/python/ledger_sdk a su proyecto, o instálelo como una dependencia local/git:
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)
# ... haga el trabajo ...
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() detiene el hilo de flush en segundo plano tras un flush final — llámelo al apagar la aplicación para que no se pierdan los eventos en buffer.
action() — cronometrar y registrar en una sola llamadaresult = ledger.action(
{"action_type": "tool_call", "config": {"tool": "save_memo"}},
lambda: save_memo(path),
)
Registra el éxito o el fallo automáticamente y relanza la excepción original sin modificar en caso de fallo.
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())
# código de la aplicación sin modificar a partir de aquí — no hace falta ninguna referencia al ledger
res = client.messages.create(
model="claude-fable-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Draft the client memo."}],
)
wrap_openai(ledger, client) funciona igual para el cliente de OpenAI.
span = ledger.record(action_type="search", started_at=t0, ended_at=t1, status="success")
with ledger.within(span["span_id"]):
# todo lo que se registre aquí dentro toma el span de búsqueda como parent_ref por defecto
client = wrap_anthropic(ledger, anthropic.Anthropic())
client.messages.create(...)
Pase parent_ref explícitamente en una llamada a record() para anular el padre ambiental.
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())
# o: BatchSpanProcessor(ledger.otel_exporter())
Cada span gen_ai.* se convierte en un evento del ledger; el resto de spans pasan sin modificar.