Le SDK Python (sdk/python, package ledger_sdk) a une parité complète avec la référence TypeScript — capture manuelle plus auto-instrumentation Anthropic/OpenAI — et ne dépend de rien en dehors de la bibliothèque standard.
Pas encore publié sur PyPI. Pour l'instant, intégrez sdk/python/ledger_sdk dans votre projet, ou installez-le comme dépendance locale/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)
# ... effectuer le travail ...
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() arrête le thread de flush en arrière-plan après un flush final — appelez-le à l'arrêt afin que les événements en tampon ne soient pas perdus.
action() — chronométrer et enregistrer en un seul appelresult = ledger.action(
{"action_type": "tool_call", "config": {"tool": "save_memo"}},
lambda: save_memo(path),
)
Enregistre automatiquement le succès ou l'échec et relève l'exception d'origine sans la modifier en cas d'échec.
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())
# code applicatif non modifié à partir d'ici — aucune référence au registre nécessaire
res = client.messages.create(
model="claude-fable-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Draft the client memo."}],
)
wrap_openai(ledger, client) fonctionne de la même manière pour le client OpenAI.
span = ledger.record(action_type="search", started_at=t0, ended_at=t1, status="success")
with ledger.within(span["span_id"]):
# tout ce qui est enregistré ici a par défaut le span de recherche comme parent_ref
client = wrap_anthropic(ledger, anthropic.Anthropic())
client.messages.create(...)
Passez parent_ref explicitement sur un appel à record() pour remplacer le parent ambiant.
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())
# ou : BatchSpanProcessor(ledger.otel_exporter())
Chaque span gen_ai.* devient un événement du registre ; les autres spans passent sans modification.