axio-context-sqlite

SQLite-backed persistent context store for axio.

Installation

pip install axio-context-sqlite

Usage

connect(db_path)

Open (or create) a SQLite database at db_path and initialise the schema. Returns an aiosqlite.Connection. The caller is responsible for closing it.

async def main():
    conn = await connect("~/.axio/chat.db")
    # ... use the connection ...
    await conn.close()

The helper enables WAL journal mode and a 5-second busy timeout so concurrent readers and a single writer can safely share the same database file.

SQLiteContextStore(conn, session_id, project=None, db_name="axio_context")

Create a context store bound to one session.

Open a connection with connect(), then create a SQLiteContextStore bound to a session. The caller owns the connection and is responsible for closing it.

import asyncio
import tempfile
import pathlib
from axio_context_sqlite import connect, SQLiteContextStore
from axio.messages import Message
from axio.blocks import TextBlock


async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        store = SQLiteContextStore(conn, session_id="my-session")
        await store.append(Message(role="user", content=[TextBlock(text="Hello")]))
        history = await store.get_history()
        assert len(history) == 1
    finally:
        await conn.close()


asyncio.run(main())

SQLiteContextStore implements the axio.context.ContextStore ABC and persists conversation history across process restarts. Multiple sessions can coexist in the same database file, isolated by session_id and project.

Storage and compression

Message content is stored as serialized JSON. Payloads larger than 512 bytes are automatically compressed with gzip (compresslevel 6) and stored base64-encoded with a gzip: prefix. Smaller payloads are stored as-is with a plain: prefix. Decompression happens transparently on read - callers never see the encoded form.

SQLite performance settings

Every connection opened by connect() is configured with:

• PRAGMA journal_mode=WAL - enables concurrent readers alongside one writer
• PRAGMA busy_timeout=5000 - waits up to 5 seconds before raising a lock error
• PRAGMA synchronous=NORMAL - balances durability and write throughput

Agent integration

import asyncio
import tempfile
import pathlib
from axio.agent import Agent
from axio_context_sqlite import connect, SQLiteContextStore


async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        ctx = SQLiteContextStore(conn, session_id="main")
        agent = Agent(system="You are helpful.", tools=[], transport=transport)
        result = await agent.run("Hello!", ctx)
        assert result == "Hi!"
    finally:
        await conn.close()


asyncio.run(main())

Listing sessions

import asyncio
import tempfile
import pathlib
from axio_context_sqlite import connect, SQLiteContextStore
from axio.messages import Message
from axio.blocks import TextBlock


async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        store = SQLiteContextStore(conn, session_id="main", project="/myproject")
        await store.append(Message(role="user", content=[TextBlock(text="hi")]))
        sessions = await store.list_sessions()
        for s in sessions:
            print(s.session_id, s.preview, s.message_count)
        assert len(sessions) == 1
    finally:
        await conn.close()


asyncio.run(main())

Token accounting

add_context_tokens(input_tokens, output_tokens) atomically increments the stored token counts for the current session and project using a SQL UPSERT (INSERT ... ON CONFLICT DO UPDATE SET ... = ... + excluded....). This is safe to call concurrently from multiple coroutines without an application-level lock.

set_context_tokens() replaces the counts unconditionally, and get_context_tokens() returns a (input_tokens, output_tokens) tuple.

Forking

fork() copies the current session's messages into a new session - useful for branching conversations without affecting the original:

import asyncio
import tempfile
import pathlib
from axio_context_sqlite import connect, SQLiteContextStore
from axio.messages import Message
from axio.blocks import TextBlock


async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        store = SQLiteContextStore(conn, session_id="main")
        await store.append(Message(role="user", content=[TextBlock(text="original")]))
        branch = await store.fork()
        assert branch.session_id != store.session_id
        assert len(await branch.get_history()) == 1
    finally:
        await conn.close()


asyncio.run(main())

License

MIT