Das Python-SDK (sdk/python, Paket ledger_sdk) hat vollständige Parität mit der TypeScript-Referenz — manuelle Erfassung plus Anthropic/OpenAI-Auto-Instrumentierung — und hat keine Abhängigkeiten außerhalb der Standardbibliothek.
Noch nicht auf PyPI veröffentlicht. Binden Sie sdk/python/ledger_sdk vorerst als Vendor-Verzeichnis in Ihr Projekt ein, oder installieren Sie es als lokale/git-Abhängigkeit:
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)
# ... die Arbeit erledigen ...
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() stoppt den Hintergrund-Flush-Thread nach einem abschließenden Flush — rufen Sie es beim Herunterfahren auf, damit gepufferte Events nicht verloren gehen.
action() — in einem Aufruf zeitmessen und erfassenresult = ledger.action(
{"action_type": "tool_call", "config": {"tool": "save_memo"}},
lambda: save_memo(path),
)
Erfasst automatisch Erfolg oder Fehlschlag und reicht die ursprüngliche Exception im Fehlerfall unverändert weiter.
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())
# ab hier unveränderter Anwendungscode — keine Ledger-Referenzen nötig
res = client.messages.create(
model="claude-fable-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Draft the client memo."}],
)
wrap_openai(ledger, client) funktioniert genauso für OpenAIs Client.
span = ledger.record(action_type="search", started_at=t0, ended_at=t1, status="success")
with ledger.within(span["span_id"]):
# alles, was hier erfasst wird, setzt parent_ref standardmäßig auf die Such-Span
client = wrap_anthropic(ledger, anthropic.Anthropic())
client.messages.create(...)
Übergeben Sie parent_ref explizit bei einem record()-Aufruf, um den ambienten Parent zu überschreiben.
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())
# oder: BatchSpanProcessor(ledger.otel_exporter())
Jede gen_ai.*-Span wird zu einem Ledger-Event; andere Spans laufen unverändert durch.