Python SDK Reference¶

The uniko Python package is a thin, typed binding over the same Rust engine the facade exposes. You program against the same verbs — observe, recall, answer, forget — from one owning Uniko handle, with native asyncio support and a blocking twin for every call. Ships py.typed and a hand-written stub, so your editor and type checker see the full surface.

import uniko

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

await session.observe(uniko.Turn("alice", "I love hiking in the mountains."))
bundle = await agent.recall("hiking")
for item in bundle.items:
    print(item.score, item.content)

The async / sync contract¶

Every facade verb ships two skins:

• the bare name (recall, observe, answer, …) returns an awaitable you await inside an event loop;
• the *_sync twin (recall_sync, observe_sync, …) blocks the calling thread on the shared tokio runtime and returns the resolved value directly. *_sync releases the GIL across the Rust work, so other Python threads keep running.

uni = await uniko.Uniko.in_memory()
bundle = await agent.recall("rockets")

uni = uniko.Uniko.in_memory_sync()
bundle = agent.recall_sync("rockets")

Construction helpers and builder setters are synchronous — there is no _sync twin for agent(), session(), Turn(...), Scope().sessions(...), or any UnikoBuilder setter. Only the async I/O verbs (and terminal build) get the twin. Throughout this page, tables mark which methods have a _sync twin in the Sync column; assume the async signature shown and append _sync to the name for the blocking form.

Uniko — the instance¶

Uniko owns the store, the model runtime, and the ingest pipeline. Build one, mint agents from it, and shutdown when done. repr(uni) is Uniko(open) or Uniko(shut down).

uni = await uniko.Uniko.in_memory()             # ephemeral
uni = await uniko.Uniko.open("./data/kb")       # persistent

uni = await (uniko.Uniko.builder()
    .path("./data/kb")
    .llm(uniko.LlmSpec.openai("answerer", "gpt-4o-mini"))
    .streaming(True)
    .scope_to_agent()
    .build())

config() returns a plain dict with keys: embedding_model_id, embedding_dimensions, recall_limit, recall_token_budget, recall_min_score.

del session, agent      # MUST drop derived handles first
uni.shutdown_sync()     # else raises ConfigError

UnikoBuilder¶

Chainable, synchronous setters; build is the terminal async step. Returns self from every setter so you can fluently compose, and re-using a builder after build() raises RuntimeError("UnikoBuilder already consumed by build()").

LlmSpec¶

Describes the generation model that powers answer() / answer_in(). All constructors are static and synchronous.

spec = uniko.LlmSpec.openai("answerer", "gpt-4o-mini")
uni = await uniko.Uniko.builder().llm(spec).build()

Agent¶

An Agent binds an identity to the store. It is Arc-backed and clonable, so passing it around is cheap. Read through the agent (recall / answer / query), retrieve by id through agent.data, and plan through agent.goals.

agent = uni.agent("assistant")
bundle = await agent.recall("what does the user prefer?")
answer = await agent.answer("when is the deadline?")   # needs .llm(...)

Properties¶

Recall, answer, query¶

Logic & reasoning¶

assume is synchronous and returns an AssumeBuilder; the awaitable lives on the builder's terminal run/run_sync. See Reasoning with Locy.

Sessions & lifecycle¶

Session¶

A Session is a conversation thread you write through. observe takes a mutable, serialized borrow — feed turns in order, one at a time. Each observe is durable and committed before it returns, so reads see the write immediately.

session = agent.session("conversation-1")
result = await session.observe(uniko.Turn("alice", "I love hiking in the mountains."))
assert result.message_node_id > 0

Properties¶

Write & ingest¶

Forget & delete¶

Data¶

By-id retrieval, reached via agent.data. An unknown id returns None — these methods never raise for a missing id.

Goals¶

Goals & tasks, reached via agent.goals. Reads are scoped to the goals the agent owns and the tasks assigned to it. Transition methods return bool — False when the id does not resolve. Creation methods return the new node id (int).

await agent.session("setup").observe(uniko.Turn("assistant", "kicking off"))
goals = agent.goals
gid = await goals.create("Ship the Python SDK", goal_id="g1",
                         description="PyO3 bindings", metrics={"target_phase": 3})
await goals.start("g1")
tid = await goals.create_task("write tests", goal_id="g1", task_id="t1", priority=0.8)
ctx = await goals.context("g1")
await goals.complete("g1", {"done": True})

Reads¶

Each read has a _sync twin (all_sync, get_sync, …).

Transitions¶

Each returns Awaitable[bool] (False when the id doesn't resolve) and has a _sync twin.

Creation¶

Both return Awaitable[int] (the new node id) and have a _sync twin. All arguments after title are keyword-only.

create(title, *, goal_id=None, description=None, status=None, metrics=None,
       guardrails=None, deadline: datetime | None = None, parent_goal_id=None)

create_task(title, *, task_id=None, description=None, status=None,
            priority: float | None = None, goal_id=None,
            depends_on_task_id=None, subtask_of_task_id=None)

AssumeBuilder¶

Returned by Agent.assume. Lets you run a hypothetical mutation plus a query against it; the mutation rolls back and never touches the real graph.

rows = await (agent.assume(
    "ASSUME { CREATE (:Fact {fact_id: 'srv-port', subject: 'srv', predicate: 'port', object: '9090'}) }")
    .then_query("MATCH (f:Fact {subject: 'srv'}) RETURN f")
    .run())
assert len(rows) == 1

after = await agent.query("MATCH (f:Fact {subject: 'srv'}) RETURN f")
assert after == []   # hypothetical mutation rolled back

Input & value-object builders¶

All input builders are synchronous constructors with chainable setters.

Turn¶

Turn(sender_id: str, content: str)

A Turn is reusable — feeding it to observe/submit does not consume it.

turn = (uniko.Turn("alice", "see the attached report")
        .addressed_to(["bob"])
        .metadata("channel", "email")
        .attach(uniko.IngestSource.from_path("report.pdf")))

IngestSource¶

Factories build a source; chainable setters refine it.

src = uniko.IngestSource.from_bytes(raw).with_mime("application/pdf").with_id("doc-1")

Scope¶

Scope()

Confines reads to a subset of the graph. Chainable.

scope = uniko.Scope().sessions(["s1"]).since(cutoff)
bundle = await agent.recall_in("budget", scope)

Output value types¶

All output types are frozen — read-only getters, no setters — and implement __repr__.

RecallSource¶

The provenance of a recalled item.

RecallItem¶

One ranked piece of context.

ContextBundle¶

The result of recall / recall_in. Implements __len__.

Answer¶

The result of answer / answer_in.

ObserveResult¶

The result of observe.

DeletionReport¶

Returned by every delete/forget/purge verb.

MessageView¶

Returned by Data.message.

ArtifactView¶

Returned by Data.artifact.

GoalView¶

TaskView¶

GoalContext¶

Returned by Goals.context.

IngestOutcome¶

Returned by Session.ingest.

AbductionResult¶

Returned by Agent.abduce. Implements __len__.

Each modification dict has keys modification, validated: bool, and cost: float.

result = await agent.abduce("ABDUCE reachable WHERE a.kind = 'query'")
for mod in result.modifications:
    print(mod["modification"], mod["validated"], mod["cost"])

Value ↔ Python type mapping¶

Cypher rows (query, query_in, run_rule, AssumeBuilder.run) come back as plain list[dict[str, Any]]. Engine values map to native Python types as follows.

Exceptions¶

The base is UnikoError(Exception). Every instance carries a .kind: str attribute, so you can branch on the kind even when the concrete class is the base.

try:
    await agent.observe(turn)
except uniko.ConflictError:
    ...                          # transient — retry
except uniko.UnikoError as e:
    if e.kind == "storage":
        ...                      # a storage-domain error folded into the base

Dedicated subclasses¶

Each is an issubclass of UnikoError and carries a fixed .kind.

Storage-domain errors¶

These fold into the base UnikoError (they are not dedicated subclasses) but still carry a distinct .kind — branch on it:

"storage" · "search" · "schema" · "pipeline" · "locy" · "embedding" · "internal"

with pytest.raises(uniko.ConfigError) as info:
    await agent.answer("anything")   # no LLM configured
assert info.value.kind == "config"

Next steps¶