Python SDK¶

The same engine, from Python — no server, no IPC¶

The uniko Python package is a PyO3 binding over the exact same Rust engine that powers the core library. It links into your Python process like SQLite: one in-process handle, no sidecar service, no network hop, no serialization tax on the hot path. You get $0-cost local ingest, graph-native reasoning, and provenance-backed recall — all from import uniko.

The SDK is async-first. Every facade verb returns an asyncio awaitable and runs the real Rust work on a shared tokio runtime. When you need to call from synchronous code, every verb also ships a *_sync twin that blocks the calling thread and releases the GIL across the Rust work — so a blocking call never starves the rest of your interpreter.

When to reach for the Python SDK¶

Reach for it when your agent lives in Python and you want cognitive memory without standing up infrastructure:

• You're building on a Python agent framework and want recall, facts, goals, and reasoning in the same process — not behind an HTTP boundary.
• You want the in-process, embedded model (no Neo4j, no Qdrant, no Postgres) with a Pythonic async surface.
• You need provenance: every recalled item traces back to the message that grounds it, and you want those handles in Python.

If you're writing Rust, use the core library directly — the Python SDK is a thin skin over the same types.

Package and module layout¶

The distribution is named uniko (import uniko). The native extension is an abi3 cdylib at uniko._uniko; the package re-exports its public surface, so you import everything straight from the top-level package:

import uniko

uni = await uniko.Uniko.in_memory()
agent = uni.agent("assistant")

• uniko — the package you import. Re-exports the full public API (30 names).
• uniko._uniko — the compiled native extension. You don't import this directly.
• Ships py.typed and a hand-written type stub (_uniko.pyi), so editors and type checkers see the full signature surface, including the *_sync twins. Requires Python ≥ 3.10.

The public surface is one Uniko handle from which you mint Agents, open Sessions, observe Turns, and read memory back through Data and Goals. Builders (Turn, IngestSource, Scope, LlmSpec, UnikoBuilder, AssumeBuilder) construct inputs; frozen view types (ContextBundle, Answer, MessageView, GoalView, …) carry results back.

Alpha status — read this first¶

The Python SDK is in active development. Here's the honest state of the world:

To build and install from the binding crate:

# from bindings/uniko-py/
maturin develop

A handful of design-doc surfaces did not ship in this alpha — notably a UnikoConfig object (config() returns a plain dict), Python-authored extractors, and visibility scoping on Scope. The API reference documents only what actually ships.

A minimal example¶

The core loop is three calls: open an instance, observe a turn, recall a context bundle. Pick your skin — the async surface and the synchronous *_sync twins are line-for-line equivalent.

import asyncio
import uniko


async def main() -> None:
    uni = await uniko.Uniko.in_memory()
    agent = uni.agent("assistant")
    session = agent.session("conversation-1")

    # Observe a message. Local ONNX extraction runs, then one atomic
    # transaction commits the Message, Entities, Observations, and edges
    # — durable and read-after-write before this returns.
    result = await session.observe(
        uniko.Turn("alice", "I love hiking in the mountains.")
    )
    assert result.message_node_id > 0

    # Recall compiled knowledge. No LLM in this path; each item carries
    # its kind, score, and the sources it traces back to.
    bundle = await agent.recall("hiking")
    for item in bundle.items:
        print(f"[{item.kind}] {item.content}")


asyncio.run(main())

import uniko

uni = uniko.Uniko.in_memory_sync()
agent = uni.agent("assistant")
session = agent.session("conversation-1")

result = session.observe_sync(
    uniko.Turn("alice", "I love hiking in the mountains.")
)
assert result.message_node_id > 0

bundle = agent.recall_sync("hiking")
for item in bundle.items:
    print(f"[{item.kind}] {item.content}")

# Shut down cleanly: drop derived handles first, then shut down the store.
del session, agent
uni.shutdown_sync()

Errors are structured¶

Every failure raises a uniko.UnikoError (or a subclass), and every instance carries a .kind string you can branch on. Subclasses cover the cases you'll catch by type — ConfigError, LlmError, TimeoutError, ConflictError (a transient, retriable SSI abort), and UnsupportedError. Storage-domain failures fold into the base UnikoError with distinct kinds ("storage", "search", "schema", "pipeline", "locy", "embedding", "internal"), so branch on e.kind for those.

try:
    answer = await agent.answer("Where does Alice like to hike?")
except uniko.ConfigError as e:
    # No LLM configured — answer()/answer_in() need an LlmSpec on the builder.
    print(f"configure an LLM first ({e.kind})")

What you can do¶

Next steps¶