Skip to main content
Axon SQL is available through the object-oriented SDK surface in both TypeScript and Python. The examples on this page use RunloopSDK in TypeScript and AsyncRunloopSDK in Python.

​
What This Is

Every Axon includes its own private, embedded SQLite database for structured state. Use it to store configuration, queues, indexes, checkpoints, and relational data that belongs to a single Axon. The database is accessed through the SDK:
  • axon.sql.query() — execute a single SQL statement
  • axon.sql.batch() — execute multiple statements atomically in one transaction

​
When to Use SQL vs Events

Use casePrefer
Structured local state, indexes, queues, configurationAxon SQL
Append-only messaging, fan-out, external coordinationAxon event stream
Bridging events to an agent process in a devboxBroker
Use the SQL database when you need querying, transactions, or secondary indexes. Use the event stream or Broker when you need messaging or coordination with external runtimes.

​
Semantics and Guarantees

Axon SQL semantics
  • One database per Axon — each Axon has its own isolated SQLite instance.
  • Private — the database is not visible from other Axons or external connections.
  • Durable — data persists for the lifetime of the Axon.
  • query() — executes exactly one SQL statement.
  • batch() — runs all statements atomically in a single transaction. If any statement fails, all writes are rolled back.
  • Serialized — concurrent queries against the same Axon are serialized server-side.

​
Getting Started

This walkthrough creates an Axon, initializes a schema, writes rows, reads them back, and runs a transaction — everything you need to start using Axon SQL. 
import json
from runloop_api_client import AsyncRunloopSDK

runloop = AsyncRunloopSDK()

# 1. Create an Axon
axon = await runloop.axon.create(name="task-tracker")

# 2. Initialize schema
await axon.sql.batch(
    statements=[
        {"sql": "CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')))"},
        {"sql": "CREATE INDEX IF NOT EXISTS idx_tasks_done ON tasks(done)"},
    ],
)

# 3. Insert rows
await axon.sql.query(
    sql="INSERT INTO tasks (title) VALUES (?)",
    params=["Review the latest PR"],
)
await axon.sql.query(
    sql="INSERT INTO tasks (title) VALUES (?)",
    params=["Update CI pipeline"],
)

# 4. Query rows
result = await axon.sql.query(sql="SELECT * FROM tasks WHERE done = 0")
print(f"Columns: {[col.name for col in result.columns]}")
for row in result.rows:
    print(row)
print(f"Query took {result.meta.duration_ms}ms")

# 5. Run an atomic transaction
await axon.sql.batch(
    statements=[
        {"sql": "UPDATE tasks SET done = 1 WHERE id = ?", "params": [1]},
        {"sql": "INSERT INTO tasks (title) VALUES (?)", "params": ["Deploy to staging"]},
    ],
)

​
Create Schema

Use batch() to set up tables and indexes together. IF NOT EXISTS makes schema creation idempotent so it is safe to run on every startup. 
await axon.sql.batch(
    statements=[
        {"sql": "CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')))"},
        {"sql": "CREATE INDEX IF NOT EXISTS idx_tasks_done ON tasks(done)"},
        {"sql": "CREATE INDEX IF NOT EXISTS idx_tasks_created ON tasks(created_at)"},
    ],
)
Schema tips:
  • Use INTEGER PRIMARY KEY for auto-incrementing IDs (SQLite aliases this to rowid).
  • Store timestamps as ISO 8601 text via datetime('now') or as milliseconds in an INTEGER column.
  • Add indexes for columns you filter or sort on frequently.
  • Wrap schema setup in batch() so all tables and indexes are created atomically.

​
Insert and Query Data

query() executes a single SQL statement and returns a SqlQueryResultView. 
await axon.sql.query(
    sql="INSERT INTO tasks (title) VALUES (?)",
    params=["Review the latest PR"],
)

result = await axon.sql.query(sql="SELECT * FROM tasks WHERE done = 0")
print(f"Columns: {[col.name for col in result.columns]}")
print(f"Rows: {result.rows}")
print(f"Query took {result.meta.duration_ms}ms")

​
Use Parameters Safely

Use ?-style positional placeholders and pass values via the params array. This prevents SQL injection and lets the database optimize repeated queries. 
result = await axon.sql.query(
    sql="SELECT * FROM tasks WHERE done = ? AND title LIKE ?",
    params=[0, "%PR%"],
)
Always use parameterized queries for user-provided values. Never interpolate strings directly into SQL.

​
Run Atomic Transactions

batch() runs multiple statements in a single transaction. If any statement fails, the entire transaction is rolled back — no partial writes. 
result = await axon.sql.batch(
    statements=[
        {"sql": "UPDATE tasks SET done = 1 WHERE id = ?", "params": [1]},
        {"sql": "INSERT INTO tasks (title) VALUES (?)", "params": ["Follow-up review"]},
        {"sql": "SELECT * FROM tasks"},
    ],
)

for i, step in enumerate(result.results):
    if step.error:
        print(f"Statement {i} failed: {step.error.message}")
    elif step.success:
        print(f"Statement {i}: {step.success.meta.changes} changes, {len(step.success.rows)} rows")

​
Understand Results

​
Query Result

query() returns a SqlQueryResultView:
FieldTypeDescription
columnsSqlColumnMetaView[]Column names and declared types
rowsarrayResult rows as arrays (empty for non-SELECT statements)
metaSqlResultMetaViewExecution metadata
Each SqlColumnMetaView has name (column name or alias) and type (declared type: TEXT, INTEGER, REAL, BLOB, or empty). SqlResultMetaView contains:
FieldTypeDescription
changesnumberRows modified by INSERT, UPDATE, or DELETE
duration_msnumberExecution time in milliseconds
rows_read_limit_reachedbooleantrue when the result was truncated at the 1,000-row limit
Example response for SELECT id, title, done FROM tasks WHERE done = 0: 
{
  "columns": [
    { "name": "id", "type": "INTEGER" },
    { "name": "title", "type": "TEXT" },
    { "name": "done", "type": "INTEGER" }
  ],
  "rows": [
    [1, "Review the latest PR", 0],
    [2, "Update CI pipeline", 0]
  ],
  "meta": {
    "changes": 0,
    "duration_ms": 1,
    "rows_read_limit_reached": false
  }
}

​
Batch Result

batch() returns a SqlBatchResultView containing a results array with one entry per statement. Each entry is a SqlStepResultView with either:
  • success — a SqlQueryResultView (same structure as a single query result)
  • error — a SqlStepErrorView with a message string

​
Handle Errors

query() raises an exception for invalid SQL or execution errors. Catch it in your application code: 
try:
    await axon.sql.query(sql="SELECT * FROM nonexistent_table")
except Exception as e:
    print(f"Query failed: {e}")
batch() executes atomically — if any statement fails, all writes are rolled back. Each step result includes either success or error, so you can inspect individual failures: 
result = await axon.sql.batch(
    statements=[
        {"sql": "INSERT INTO tasks (title) VALUES (?)", "params": ["Valid insert"]},
        {"sql": "INSERT INTO nonexistent (col) VALUES (?)", "params": ["This fails"]},
    ],
)

for i, step in enumerate(result.results):
    if step.error:
        print(f"Statement {i} failed: {step.error.message}")
Constraint violations (e.g. UNIQUE, NOT NULL, FOREIGN KEY) behave the same way — the failing statement produces an error in the step result and the entire batch is rolled back.

​
Common Patterns

​
Key-Value Store

A simple key-value table for agent state, configuration, or checkpoints: 
await axon.sql.batch(
    statements=[
        {"sql": "CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, value TEXT NOT NULL)"},
        {"sql": "INSERT OR REPLACE INTO kv (key, value) VALUES (?, ?)", "params": ["cursor", "evt_42"]},
        {"sql": "INSERT OR REPLACE INTO kv (key, value) VALUES (?, ?)", "params": ["status", "running"]},
    ],
)

result = await axon.sql.query(
    sql="SELECT value FROM kv WHERE key = ?",
    params=["cursor"],
)

​
Task Queue

A FIFO queue with claim semantics using UPDATE ... LIMIT 1: 
await axon.sql.query(
    sql="CREATE TABLE IF NOT EXISTS queue (id INTEGER PRIMARY KEY, payload TEXT NOT NULL, claimed INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')))",
)

await axon.sql.query(
    sql="INSERT INTO queue (payload) VALUES (?)",
    params=['{"task": "run-tests", "repo": "runloop"}'],
)

claimed = await axon.sql.batch(
    statements=[
        {"sql": "UPDATE queue SET claimed = 1 WHERE id = (SELECT id FROM queue WHERE claimed = 0 ORDER BY id LIMIT 1) RETURNING *"},
    ],
)

​
Schema Versioning

Track schema migrations with a version table so schema setup is safe to re-run: 
await axon.sql.query(
    sql="CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)",
)

result = await axon.sql.query(sql="SELECT MAX(version) as v FROM schema_version")
current_version = result.rows[0][0] if result.rows and result.rows[0][0] is not None else 0

if current_version < 1:
    await axon.sql.batch(
        statements=[
            {"sql": "CREATE TABLE tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0)"},
            {"sql": "INSERT INTO schema_version (version) VALUES (1)"},
        ],
    )

if current_version < 2:
    await axon.sql.batch(
        statements=[
            {"sql": "ALTER TABLE tasks ADD COLUMN created_at TEXT NOT NULL DEFAULT (datetime('now'))"},
            {"sql": "CREATE INDEX idx_tasks_created ON tasks(created_at)"},
            {"sql": "INSERT INTO schema_version (version) VALUES (2)"},
        ],
    )

​
Limits and Caveats

  • The Axon SQL database uses standard SQLite semantics. Supported types are TEXT, INTEGER, REAL, BLOB, and NULL.
  • Booleans are represented as integers (0 / 1).
  • query() executes exactly one SQL statement. Multi-statement strings are not supported in a single query() call — use batch() instead.
  • Axons Overview — Event streams, publishing, and subscribing
  • Broker — Bridging Axons to agents running in devboxes
  • SDKs — TypeScript and Python SDK installation