Python SDK

Full guide for integrating ZizkaDB in Python agents — FastAPI services, LangChain, CrewAI, notebooks, or scripts.

Prerequisites

Python 3.10+. Works in venv, conda, or system Python.

bashpython3 --version # 3.10 or higher

Choose your environment

Managed cloud: Sign up → Settings → Create API key (zizkadb_live_…).

Self-host: use host="http://localhost:8000" (dev key auto-sent). See Self-host docs → click Open my dashboard → on the login page.

Install

bashpip install "zizkadb-sdk>=0.5.0"

Install zizkadb-sdk on PyPI — not the unrelated agentdb package.

Initialize the client

pythonfrom zizkadb import ZizkaDB

# Managed cloud (db.zizka.ai)
db = ZizkaDB("zizkadb_live_xxxx")

# Self-hosted
# db = ZizkaDB(host="http://localhost:8000")

Log your first event

The SDK is async-first. Wrap calls in async def main() and run with asyncio.run(main()).

pythonimport asyncio
from zizkadb import ZizkaDB

db = ZizkaDB("zizkadb_live_xxxx")

async def main():
    result = await db.log(
        agent="support-bot",
        event="tool_call",
        data={"tool": "search", "query": "pricing"},
        session_id="sess_abc123",   # group one conversation
    )
    print("Saved:", result.event_id)

asyncio.run(main())

Link events causally

Pass parent_id so you can call why() later.

pythonasync def main():
    msg = await db.log(
        agent="support-bot", event="user_message",
        data={"text": "why is my bill $200?"},
        session_id="sess_abc123",
    )
    tool = await db.log(
        agent="support-bot", event="tool_call",
        data={"tool": "get_billing"},
        parent_id=msg.event_id,
        session_id="sess_abc123",
    )
    resp = await db.log(
        agent="support-bot", event="assistant_response",
        data={"text": "I found a duplicate charge."},
        parent_id=tool.event_id,
        session_id="sess_abc123",
    )
    chain = await db.why(resp.event_id)
    chain.print()

asyncio.run(main())

Use search, replay, and memory

pythonasync def main():
    # Search: keyword out of the box, hybrid with embeddings (mode="keyword"|"semantic"|"hybrid")
    hits = await db.search("billing complaint", agent="support-bot", limit=5)

    # Safe retries and batches
    await db.log("support-bot", "tool_call", {"tool": "refund"}, idempotency_key="req-42")
    await db.log_batch([
        {"agent": "support-bot", "event": "user_message", "data": {"text": "hi"}},
        {"agent": "support-bot", "event": "llm_response", "data": {"text": "hello"}, "parent_index": 0},
    ])

    # Prove the history is untouched; keep chain_head elsewhere as an anchor
    check = await db.verify_chain("support-bot")
    print(check["intact"], check["chain_head"])

    # Reconstruct logged state at a time
    from datetime import datetime, timezone
    state = await db.at("support-bot", datetime(2026, 5, 1, 15, 0, tzinfo=timezone.utc))

    # Prompt-ready memory block
    ctx = await db.context_for(
        agent="support-bot",
        task="user asking about refund policy",
        max_tokens=2000,
    )

    # Drift vs baseline (needs enough sessions with session_id)
    baseline = await db.baseline("support-bot", recent_window=50)
    print(baseline)

asyncio.run(main())

Framework example: OpenAI tool loop

pythonimport asyncio
from openai import AsyncOpenAI
from zizkadb import ZizkaDB

db = ZizkaDB("zizkadb_live_xxxx")
openai = AsyncOpenAI()

async def run_turn(user_text: str, session_id: str):
    msg = await db.log(
        agent="my-agent", event="user_message",
        data={"text": user_text}, session_id=session_id,
    )
    parent_id = msg.event_id

    response = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": user_text}],
    )
    text = response.choices[0].message.content or ""
    await db.log(
        agent="my-agent", event="assistant_response",
        data={"text": text}, parent_id=parent_id, session_id=session_id,
    )
    return text

asyncio.run(run_turn("Hello", "sess_001"))

All Python methods

db.log()Log any event. parent_id links cause → effect; idempotency_key makes retries safe; supersedes retires an outdated fact.

db.log_batch(events)Up to 500 events in one request, with per-event results.

db.why(event_id)Walk the causal chain from an event to its root.

db.search(query)Search agent history in plain English: keyword out of the box, hybrid keyword + vector with embeddings.

db.verify_chain(agent)Prove no event was edited, removed or reordered (per-agent hash chain).

db.at(agent, timestamp)Reconstruct logged state at a past timestamp.

db.query(agent)List recent events, optionally filtered by event type.

db.context_for(agent, task)Memory block formatted for system prompt injection.

db.baseline(agent)Behavioral baseline and drift score vs historical sessions.

db.memory_diff(session_id)Summary of what happened in one session.

db.forget(key, value)GDPR erasure by metadata filter; leaves a content-free tombstone in the audit chain.

db.export_events(agent)Stream every event as NDJSON; import_events() loads it into another workspace.

db.iter_events(agent)Every event, newest first, cursor-paged.

db.agents()List all agents for your tenant.

Troubleshooting

SyntaxError: await outside async function

Wrap code in async def main() and asyncio.run(main()).

401 Invalid API key

Copy the full key from Dashboard → Agents (no spaces). Use ZIZKADB_API_KEY or AGENTDB_API_KEY in your env.

403 Agent mismatch

Your key is scoped to one agent (e.g. my-bot) but code logged to a different agent name. Use the same name in db.log(), or create a tenant-wide key in Settings.

Dashboard empty but SDK works

Check you are viewing the same agent name your code logs to. Click Test agent on the agent page. Settings test event goes to dashboard-connection-test, not your app agent.

Search returns nothing

Configure embeddings in Settings. Log a few events first so there is history to search.

baseline shows warming_up

Log more sessions with session_id on each event. Drift needs volume.