# Python reference

> The primitives-sdk package (import primitives_sdk): an async client for every primitive.

Source: https://useprimitives.com/docs/sdk/python

The official Python client is published as `primitives-sdk` and imported as `primitives_sdk`. It's async, built on `httpx`, fully typed, and needs Python 3.10 or later.

```bash
pip install primitives-sdk
```

## Client

```python title="main.py"
import asyncio, os
from primitives_sdk 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())
```

`Primitives(token=None, *, base_url=None, timeout=30.0, max_retries=2, transport=None)`. `timeout` is in seconds; `transport` takes an `httpx.AsyncBaseTransport`, for tests. With a `pk_` key the client defaults to `https://api.useprimitives.com`; otherwise to `PRIMITIVES_URL` or `http://localhost:7777`. It retries `429` responses, and `5xx` on `GET`, `PUT`, `DELETE` and `HEAD`, with exponential backoff that honours `Retry-After`. Use it as an async context manager, or call `await p.aclose()`, so the connection pool is closed.

## Naming conventions

The Python SDK mirrors the [TypeScript SDK](/docs/sdk/typescript), with these differences:

| TypeScript | Python | Why |
| --- | --- | --- |
| `job.retry({ in: '1h' })` | `job.retry(in_="1h")` | `in` is a keyword |
| `fs.forkDir({ from, to })` | `fs.fork_dir(source, dest=None)` | `from` is a keyword |
| `fs.move(from, to)` | `fs.move(source, dest)` | |
| `db.fork({ name, group, snapshot })` | `db.fork(name=None, *, group=None, snapshot=None)` | |
| `p.primitives.create({ name, type, group })` | `p.primitives.create(name=, type=, group=)` | |
| `readText`, `dedupeKey`, `branchHandle` | `read_text`, `dedupe_key`, `branch_handle` | snake_case |
| `kv.getMany`, `setMany`, `deleteMany`, `ifVersion` | `kv.get_many`, `set_many`, `delete_many`, `if_version` | snake_case |
| `kv.entries()`, `storage.objects()` | `async for e in kv.iter()`, `async for o in storage.iter()` | |
| `storage.getText`, `revokeSignedUrls`, `expiresIn` | `storage.get_text`, `revoke_signed_urls`, `expires_in` | snake_case |
| Options objects | Keyword arguments | |
| ``db.sql`...${x}` `` | `db.sql("... ?", [x])` | Positional or named parameters |
| `db.export()` → `Uint8Array` | `db.export()` → `bytes` | Also on KV and storage |
| `p.groups.all()` | `async for m in p.groups.list()` | |
| `p.primitives.all(filter)` | `await p.primitives.list(type=, group=)` | Python's `list` returns every page |
| `db.transaction([...])` | `async with db.transaction() as tx:` | Statements are sent as one batch on exit |
| `storage.get(key, { range: { start, end } })` | `storage.get(key, range=(start, end))` or `suffix=n` | |
| `revokeSignedUrls()` → `{ epoch }` | `revoke_signed_urls()` → `int` | |

Return values are frozen dataclasses such as `Manifest`, `GroupInfo`, `PrimitiveInfo`, `SnapshotInfo`, `PrimitiveStats`, `QueryResult`, `FileInfo`, `JobData`, `MemoryRecord`, `RecallHit`, `KvEntry`, `ObjectInfo` and `SignedUrl`. Paged results from `q.jobs()`, `notes.list()` and the audit methods are dicts with `data` and `next_cursor`. Durations are strings (`"5m"`), integer milliseconds, or `datetime.timedelta`.

## Groups

A group is a collection of primitives reached with one key. It starts empty.

```python
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": ...}; workspace 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
```

## Audit

The workspace audit trail, newest first, filtered by any of `group`, `type`, `primitive`, `action` (exact or a prefix like `"db."`), `actor`, `since` and `until` (ms or a `datetime`). Workspace keys only, against the cloud (or BYOC): `primitives dev` has no audit trail. See [Audit log](/docs/concepts/audit).

```python
from datetime import datetime, timedelta, timezone

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)
```

## Primitives

Every primitive has an immutable `id`. Create, list and open primitives by id from the client:

```python
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")
```

`group.db`, `group.fs`, `group.queue`, `group.memory`, `group.vault`, `group.kv` and `group.storage` are shortcuts that work by name inside one group:

```python
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)
```

Every handle has the same lifecycle methods:

```python
info = await orders.info()            # PrimitiveInfo: id, name, type, group, created_at, 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 trial.delete()
```

`delete`, `rename`, `pause` and `resume` need a workspace key. A data method that doesn't fit the primitive's type raises `PrimitivesError(code="wrong_type")`.

## Database

```python
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")
cols = await trial.table("tasks")          # same as db.table("tasks", branch="try-migration")
await trial.promote()  # or force=True, or await trial.discard()

sqlite_file = await db.export()  # bytes
```

`sql`, `query` and `exec` are the same method: one statement per call. `transaction()` collects statements and sends them as one atomic batch when the block exits without an exception; the results are on `tx.results`. Pass `bytes` to bind a blob. `tables(branch=None)` and `table(name, branch=None)` return dicts (`table` has `name`, `type`, `columns` and `sql`); pass `branch` to inspect a branch instead of main.

## Filesystem

```python
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)
```

`fork_dir` copies one directory inside the filesystem; `fork` copies the whole filesystem into a new primitive.

## Queue

```python
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})
```

## Memory

```python
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", metadata={"source": "chat"})
page = await notes.list(kind="semantic", owner="user-7", limit=20)   # {"data": [...], "next_cursor": ...}
await notes.consolidate(threshold=0.9)
await notes.forget(newer.id)
```

`supersede` takes `text`, `kind`, `tags` and `metadata`; omitted ones carry over from the old memory, and `owner` always does. `tags=[]` clears the tags.

## Vault

```python
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)
```

`list()` and `grants()` return every secret and grant, following pages for you.

`fetch` returns a `ProxiedResponse` with `status`, `headers`, `body`, `body_encoding` and `truncated`. The secret's value is redacted from it. `unset(name)` deletes one secret and `delete()` deletes the vault. `set`, `unset`, `grant` and `revoke` need a workspace key.

## KV

```python
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)
```

`KvEntry` has `key`, `value`, `encoding`, `version`, `expires_at`, `created_at` and `updated_at`. Pass `bytes` as a value to store raw bytes. See [KV](/docs/primitives/kv).

## Storage

```python
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, or None if missing: 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
```

`ObjectInfo` has `key`, `size`, `etag`, `content_type`, `metadata`, `created_at` and `updated_at`. `presign` defaults to `method="GET"` and `expires_in="15m"`, and takes `if_none_match="*"` for a create-only upload URL. `get` takes `range=(start, end)` (inclusive, `end` may be `None`) or `suffix=n`. `delete(key)` deletes an object, while `delete()` with no key deletes the store; the same goes for `KvStore`. Both have `export()`, returning the store as SQLite `bytes`. See [Storage](/docs/primitives/storage).

## Errors

```python
from primitives_sdk import PrimitivesError

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

`PrimitivesError` has `status`, `code`, `message` and `request_id`. The error codes are listed in the [REST API reference](/docs/api/errors).