# Python quickstart

> Install primitives-sdk and build a group with a database, KV cache, job queue and memory from async Python.

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

This guide builds a small agent backend with the Python SDK: a group with a database, a KV cache, a job queue and memory. It runs against a local engine; switching to the cloud is one line.

## 1. Install

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

The package is `primitives-sdk` and the import is `primitives_sdk`. It's async, built on `httpx`, fully typed, and needs Python 3.10 or later.

## 2. Start an engine

Run the engine locally with the [CLI](/docs/cli):

```bash
primitives dev
```

Or use Primitives Cloud: create an API key in the dashboard and export it.

```bash
export PRIMITIVES_API_KEY="pk_…"
```

## 3. Create a client and a group

```python title="agent.py"
import asyncio, os
from primitives_sdk import Primitives

async def main():
    # No key: http://localhost:7777. With a pk_ key: https://api.useprimitives.com.
    async with Primitives(os.environ.get("PRIMITIVES_API_KEY")) as p:
        group = await p.groups.create(name="agent-42")
        ...

asyncio.run(main())
```

Use the client as an async context manager, or call `await p.aclose()`, so its connection pool is closed. The snippets below run inside `main()`.

## 4. Store data in a database

```python
db = await group.db.create("main")

await db.exec("create table tasks (id integer primary key, title text, status text)")
await db.sql("insert into tasks (title, status) values (?, ?)", ["Write docs", "open"])

result = await db.sql("select id, title from tasks where status = ?", ["open"])
print(result.rows)  # [{'id': 1, 'title': 'Write docs'}]
```

Always pass values as parameters, never by formatting them into the SQL.

## 5. Cache with KV

```python
cache = await group.kv.create("cache")

await cache.set("session:abc", {"step": 1}, ttl="1h")
entry = await cache.get("session:abc")
print(entry.value["step"])  # 1; entry.version is for compare-and-swap

await cache.incr("rate:agent-42", 1, ttl="1m")
```

## 6. Run jobs on a queue

```python
jobs = await group.queue.create("jobs")
emails = jobs.queue("emails")

await emails.send({"to": "ada@example.com"}, dedupe_key="welcome-ada")

async for job in emails.consume(lease="5m"):
    try:
        await send_email(job.body)
        await job.ack()
    except Exception as err:
        await job.retry(in_="1m", error=str(err))
```

`consume` long-polls forever; `break` out of the loop to stop. `in_` has a trailing underscore because `in` is a keyword.

## 7. Give it memory

```python
notes = await group.memory.create("notes")

await notes.remember(kind="semantic", text="User prefers pnpm over npm", tags=["preferences"])

hits = await notes.recall("which package manager?", limit=1)
print(hits[0].memory.text)  # User prefers pnpm over npm
```

## 8. Snapshot before risky changes

```python
await db.snapshot("before-migration")
trial = await db.fork(name="main-trial")

await trial.exec("alter table tasks add column priority integer")
# A fork is an independent copy. Happy? Run the same migration on db. Then delete the fork:
await trial.delete()
```

## Handle errors

```python
from primitives_sdk import PrimitivesError

try:
    await group.db.create("main")
except PrimitivesError as err:
    if err.code != "already_exists":
        raise
```

## Next steps

- [Python reference](/docs/sdk/python): every method and type.
- [Use with a sandbox](/docs/guides/sandboxes): hand a scoped key to an agent.
- [Primitives](/docs/primitives/database): how each type behaves.