Python SDK

pip install useprimitives

import asyncio, os
from primitives import Primitives

async def main():
    async with Primitives() as p:  # PRIMITIVES_URL or http://localhost:7777
        group = await p.groups.create(name="agent-42")
        db = await group.db.create("main")
        result = await db.sql("select 1 as ok")
        print(result.rows)

asyncio.run(main())

group = await p.groups.get("agent-42")
async for m in p.groups.list():
    print(m.name)

agent = p.group("agent-42")  # handle without a request, e.g. with a key scoped to agent-42
info = await group.info()    # GroupInfo, with .primitives
everything = await group.primitives.list()           # [PrimitiveInfo]
databases = await group.primitives.list(type="db")
page = await group.audit(limit=20, action="db.")  # {"data": [AuditEntry], "next_cursor": ...}; account keys only
await group.destroy()                # an empty group; conflict (409) if it has primitives
await group.destroy(force=True)      # the group and every primitive in it

page = await p.audit(group="agent-42", type="db", since=datetime.now(timezone.utc) - timedelta(days=1))
async for entry in p.audit_all(action="vault."):
    print(entry.ts, entry.actor, entry.action, entry.primitive_id)

orders = await p.primitives.create(name="orders", type="db", group="agent-42")
print(orders.id)                           # "01jbz3k8m5q2w7x9v4n6t0r1ce"

same = p.db(orders.id)                     # typed handle, no request; also p.fs, p.queue, p.memory, p.vault, p.kv, p.storage
anything = p.primitive(orders.id)          # untyped handle with the lifecycle methods
info = await p.primitives.get(orders.id)   # PrimitiveInfo
dbs = await p.primitives.list(type="db", group="agent-42")

orders = await group.db.create("orders")  # create in this group; already_exists if taken
cache = group.db("cache")                 # resolved by name in the group on first use
for i in await group.db.list():           # [PrimitiveInfo]
    print(i.id, i.name, i.size_bytes)

info = await orders.info()            # PrimitiveInfo: id, name, type, group, paused, size_bytes, forked_from
await orders.snapshot("before-deploy")
for snap in await orders.snapshots(): # newest first
    print(snap.name, snap.created_at)
await orders.restore("before-deploy") # a snapshot name or id

trial = await orders.fork(name="orders-try")                 # a Database handle for the copy
staging = await orders.fork(group="staging")                 # into another group; new id; not_found if it doesn't exist
old = await orders.fork(name="orders-old", snapshot="before-deploy")

stats = await orders.stats()          # PrimitiveStats: id, name, type, size_bytes, items, paused
trail = await orders.audit(action="db.exec", limit=20)  # this primitive's own audit entries
await orders.pause()                  # data calls now raise PrimitivesError(code="paused")
await orders.resume()
await orders.rename("orders-v1")     # same id
await orders.move_to("agent-43")      # same id, another group
await trial.delete()

db = group.db("main")
result = await db.sql("select id, title from tasks where status = ?", ["open"])
for row in result.rows:
    print(row["title"])

await db.batch([
    "create table if not exists log (msg text)",
    ("insert into log (msg) values (?)", ["started"]),
])

async with db.transaction() as tx:
    await tx.sql("update tasks set status = ? where id = ?", ["done", 7])
    await tx.sql("insert into log (msg) values (?)", ["closed 7"])

trial = await db.branch("try-migration")
await trial.exec("alter table tasks add column priority integer")
await trial.promote()  # or force=True, or await trial.discard()

sqlite_file = await db.export()  # bytes

fs = group.fs("workspace")
await fs.write("/repo/plan.md", plan)
text = await fs.read_text("/repo/plan.md")
data = await fs.read("/repo/logo.png")          # bytes
files = await fs.glob("/repo/src/**/*.py")
hits = await fs.search("lease expired", path="/repo")

fork = await fs.fork_dir("/repo")               # {"path": ..., "files": ...}
await fs.move(fork["path"], "/scratch/attempt-1")
await fs.remove("/scratch", recursive=True)

store = group.queue("jobs")     # a queue primitive holds named queues and topics
q = store.queue("reviews")
await q.send({"pr": 1234}, delay="5m", dedupe_key="pr-1234")

async for job in q.consume(lease="10m"):
    await job.heartbeat(extend="10m")
    await review_pull_request(job.body)
    await job.ack()  # or job.retry(in_="1h"), job.fail("reason"), job.dead("reason")

dead = await q.jobs(status="dead")
purged = await q.purge("done", older_than="7d")
counts = await store.queues()    # per-queue counts
await store.topics.publish("pr-opened", {"pr": 1234})

notes = group.memory("notes")
await notes.remember(kind="semantic", text="User prefers pnpm over npm", tags=["preferences"])
hits = await notes.recall("package manager?", kinds=["semantic", "episodic"], limit=5)
newer = await notes.supersede(hits[0].memory.id, text="User now prefers bun")
await notes.consolidate(threshold=0.9)
await notes.forget(newer.id)

prod = group.vault("prod")
await prod.set("GITHUB_TOKEN", os.environ["GITHUB_TOKEN"])
await prod.grant(
    "GITHUB_TOKEN",
    allow={"hosts": ["api.github.com"], "methods": ["GET", "POST"]},
    ttl="15m",
)
res = await prod.fetch("https://api.github.com/user", auth={"bearer": "GITHUB_TOKEN"})
print(res.status, res.body)

cache = group.kv("cache")
await cache.set("session:abc", {"step": 3}, ttl="1h")       # without ttl, any expiry is cleared
entry = await cache.get("session:abc")                       # KvEntry, or None
await cache.set("session:abc", {"step": 4}, if_version=entry.version)  # precondition_failed (412) if it changed

counter = await cache.incr("rate:agent-42", 1, ttl="1m")
await cache.set_many([{"key": "a", "value": 1}, {"key": "b", "value": 2}])  # atomic by default; atomic=False stops at the first failure
values = await cache.get_many(["a", "b"])
removed = await cache.delete_many(["a", "b"])

async for e in cache.iter(prefix="session:", values=True):
    print(e.key, e.value, e.expires_at)

artifacts = group.storage("artifacts")
info = await artifacts.put("reports/q3.pdf", pdf_bytes, content_type="application/pdf", metadata={"run": "r-1234"})
await artifacts.put("locks/run-1234", b"", if_none_match="*")   # create only
text = await artifacts.get_text("notes/summary.md")
meta = await artifacts.head("reports/q3.pdf")                    # ObjectInfo
obj = await artifacts.get("reports/q3.pdf", range=(0, 1023))  # StorageObject: body, size, etag, content_type, metadata, last_modified, range
await artifacts.copy("reports/q3.pdf", "reports/latest.pdf")   # copy(source, dest)
await artifacts.delete("reports/q3.pdf", if_match=meta.etag)

async for obj in artifacts.iter(prefix="reports/"):
    print(obj.key, obj.size, obj.etag)

link = await artifacts.presign("reports/latest.pdf", expires_in="1h")  # SignedUrl: url, method, headers, expires_at
upload = await artifacts.presign("uploads/a.png", method="PUT", content_type="image/png", max_size=5_000_000)
epoch = await artifacts.revoke_signed_urls()                     # int; invalidates every URL issued so far

from primitives import PrimitivesError

try:
    await trial.promote()
except PrimitivesError as err:
    if err.code == "branch_conflict":
        await trial.discard()
    else:
        raise