ArqonDB Python SDK

Agent memory with KV, vector search, and temporal causal graphs — in one database.

Quick Start

pip install arqondb

High-Level Client

from arqondb import Client

# No auth (local dev without GATEWAY_JWT_SECRET)
db = Client("localhost:9379")

# With auth (auto-login, fetches JWT from gateway management API)
db = Client("localhost:9379", username="admin", password="admin")

# With a pre-obtained token
db = Client("localhost:9379", token="eyJ...")

# KV — keys can be str or bytes
db.put("key", b"value")
print(db.get("key"))          # b"value"
db.delete("key")

Agent State: Causal Graph

from arqondb import Client

db = Client("localhost:9379", username="admin", password="admin")

# Add steps to the causal graph
s1 = db.add_step("agent-1", "Observe", content=b"user clicked buy")
s2 = db.add_step("agent-1", "Think",   content=b"should confirm order")
s3 = db.add_step("agent-1", "Act",     content=b"sent confirmation email")

# Link them causally
db.add_edge(s1, s2, "Triggers")
db.add_edge(s2, s3, "Triggers")

# Traverse the chain
result = db.traverse(s1, direction="forward", max_depth=5)
for step in result.steps:
    print(f"  [{step.step_type}] {step.content}")

# Find similar causal chains by embedding
chains = db.find_similar_chains(query_embedding=[0.1] * 128, k=3, chain_depth=3)

Agent State: Branching (Fork)

# Fork a speculative branch
branch_id = db.fork("experiment-a")

# Write to the branch (isolated from main timeline)
db.branch_put(branch_id, "config:model", b"gpt-4o")
val = db.branch_get(branch_id, "config:model")

# Merge back or discard
db.merge_branch(branch_id)
# db.discard_branch(branch_id)

# List all branches
for b in db.list_branches():
    print(f"  branch {b.id}: {b.label} ({b.status})")

Agent State: Reactive State

# Compare-and-swap for optimistic concurrency
result = db.cas_put("counter", expected_seq=0, new_value=b"1")
if result.success:
    print(f"written, new version: {result.new_seq}")
else:
    print(f"conflict, current version: {result.actual_seq}")

# Watch for changes (streaming)
for event in db.watch_prefix("agent-1", "state:"):
    print(f"  {event.event_type}: {event.key} = {event.value}")

Async Client

import asyncio
from arqondb import AsyncClient

async def main():
    async with AsyncClient("localhost:9379", username="admin", password="admin") as db:
        await db.put("key", b"value")
        print(await db.get("key"))

        step_id = await db.add_step("agent-1", "Observe", content=b"hello")
        step = await db.get_step(step_id)
        print(step)

asyncio.run(main())

Agent Memory (High-Level 3-Primitive API)

from arqondb import AgentMemory

memory = AgentMemory("127.0.0.1:9379")

# Store observations
obs1 = memory.observe("user prefers dark mode", metadata={"source": "settings"})
obs2 = memory.observe("user switched to light mode after update")

# Link them causally
memory.link(obs1, obs2, relation="caused")

# Recall relevant memories
results = memory.recall("what theme does the user prefer?", k=5)
for r in results:
    print(f"  [{r.id}] {r.text} (distance={r.distance:.3f})")

Low-Level Client (bytes keys, direct gRPC)

from arqondb import ArqonDBClient, VectorIndexConfig

with ArqonDBClient("127.0.0.1:9379") as client:
    # KV
    client.put(b"key", b"value")
    print(client.get(b"key"))  # b"value"

    # Vector search
    client.create_vector_index("my_index", VectorIndexConfig(dim=768, metric="cosine"))
    client.vector_put("my_index", 1, [0.1] * 768)
    results = client.vector_search("my_index", [0.1] * 768, k=5)

    # Graph
    client.create_graph_index("my_graph", dim=768, metric="cosine")
    client.graph_add_node("my_graph", node_id=1, properties=b'{"type":"event"}')
    client.graph_add_edge("my_graph", src=1, dst=2, edge_type="caused", valid_from=1700000000)

API Reference

Client (recommended)

AsyncClient

Same API as Client, all methods are async.

Why ArqonDB?

Requirements

• ArqonDB server running (see arqondb.com)
• Python >= 3.9