How Session Persistence Works with SQLite, Redis, and Custom Backends in openai-agents-python

Session persistence in openai-agents-python relies on an abstract async protocol that the runtime consumes, enabling seamless swapping between SQLite, Redis, or any custom storage backend by implementing five core async methods.

The openai-agents-python SDK stores turn-by-turn conversation history in a session object that persists across agent runs. Instead of enforcing a specific database, the library defines a minimal interface in src/agents/memory/session.py that any storage layer can implement, allowing developers to choose between the built-in SQLite and Redis providers or build their own integration with cloud databases or in-memory caches.

The Session Persistence Protocol

Every session backend must inherit from SessionABC and implement five async methods defined in src/agents/memory/session.py:

• get_items(limit: int | None = None) -> list[TResponseInputItem] – Retrieves conversation history, optionally truncated to the most recent limit items.
• add_items(items: list[TResponseInputItem]) -> None – Appends new turn data to the session.
• pop_item() -> TResponseInputItem | None – Atomically removes and returns the most recent item; used for retry rollbacks.
• clear_session() -> None – Wipes all data for the session ID.
• close() -> None – Cleans up connections (pools, file handles, etc.).

The runtime in src/agents/run_internal/session_persistence.py treats these methods as a black box. It calls prepare_input_with_session() to merge historic items with new user input, save_result_to_session() to persist the model’s response, and rewind_session_items() to pop items when a retry is triggered.

SQLite Session Backend

The SQLite implementation in src/agents/memory/sqlite_session.py provides a file-backed or in-memory session store using Python’s standard sqlite3 module with aiosqlite for async support.

Connection and Schema Management

The class maintains thread-local connections for file-backed databases and a shared connection for :memory: instances. It lazily initializes two tables:

• agent_sessions – Stores metadata (session_id, created_at, updated_at).
• agent_messages – Stores the JSON-serialized items with an auto-incrementing id for ordering.

Read and Write Operations

Retrieval uses SELECT … ORDER BY id ASC for full history, or ORDER BY id DESC LIMIT ? followed by a reversal to maintain chronological order when a limit is specified.

Insertion wraps items in a transaction using INSERT OR IGNORE for the session row followed by INSERT into agent_messages.

Atomic pop leverages SQLite 3.35+ syntax: DELETE … RETURNING removes the newest row and returns its JSON payload in a single atomic operation, preventing race conditions during retries.

from agents.memory.sqlite_session import SQLiteSession
from agents.run import Runner

# Persistent disk-based session

session = SQLiteSession(
    session_id="conversation-123",
    db_path="chat_history.db"
)

result = await Runner.run(agent, "Hello!", session=session)
await session.close()

Redis Session Backend

The Redis implementation in src/agents/extensions/memory/redis_session.py uses an async Redis client (via redis-py) to store conversation history in Redis lists and hashes.

Key Layout and Data Structure

The backend uses a structured key pattern:

• {key_prefix}:{session_id} – A hash storing session metadata (created_at, updated_at).
• {key_prefix}:{session_id}:messages – A Redis list containing JSON-serialized items in insertion order.
• {key_prefix}:{session_id}:counter – An atomic counter for generating unique IDs.

Operations and TTL Management

Reading history uses LRANGE 0 -1 for full retrieval or LRANGE -limit -1 for recent items. The implementation handles both bytes and str payloads for compatibility.

Writing uses a pipeline to atomically:

1. Set metadata hash fields with HSET.
2. Push JSON strings to the list with RPUSH.
3. Update the updated_at timestamp.

If ttl is configured, the EXPIRE command is applied to all keys in the same pipeline.

Popping items for retry rollbacks uses RPOP to remove the most recent entry from the list.

from agents.extensions.memory.redis_session import RedisSession
from agents.run import Runner

# Create from URL with optional TTL

session = RedisSession.from_url(
    session_id="user-456",
    url="redis://localhost:6379/0",
    ttl=86400  # 24 hour expiration

)

await Runner.run(agent, "Explain quantum computing", session=session)
await session.close()

Implementing Custom Session Backends

Any class inheriting from SessionABC in src/agents/memory/session.py can serve as a backend. The protocol is intentionally minimal to reduce implementation friction.

For example, a custom in-memory session useful for unit testing:

from agents.memory.session import SessionABC
from agents.items import TResponseInputItem

class InMemorySession(SessionABC):
    def __init__(self, session_id: str):
        self.session_id = session_id
        self._store: list[TResponseInputItem] = []

    async def get_items(self, limit: int | None = None):
        if limit is None:
            return list(self._store)
        return self._store[-limit:]

    async def add_items(self, items: list[TResponseInputItem]):
        self._store.extend(items)

    async def pop_item(self):
        return self._store.pop() if self._store else None

    async def clear_session(self):
        self._store.clear()

    async def close(self):
        pass

Common custom implementations include:

• DynamoDB or Azure Table Storage for serverless deployments.
• PostgreSQL with JSONB columns for complex querying.
• File-based JSONL append-only logs for audit trails.

Runtime Integration and Lifecycle

The session persistence layer integrates with the agent runtime in src/agents/run_internal/session_persistence.py through three core operations:

1. Input Preparation (prepare_input_with_session): Retrieves historic items via get_items() and merges them with the current user input. The runtime deduplicates items using _fingerprint_or_repr (lines 778-782) to avoid storing duplicate context.

2. Result Persistence (save_result_to_session): Appends the model's response items to the session via add_items(). For OpenAI Responses API sessions, optional compaction logic may truncate older items to manage token limits.

3. Retry Rollback (rewind_session_items): When a model call fails or requires retry, the runtime calls pop_item() to atomically remove the partially saved turn, then verifies cleanup with wait_for_session_cleanup.

The session object is passed to Runner.run() via the session parameter. The runtime does not manage connection lifecycles beyond calling close() when the runner finishes, making the implementation responsible for connection pooling and cleanup.

Summary

• Session persistence in openai-agents-python follows an abstract async protocol defined in SessionABC with five methods: get_items, add_items, pop_item, clear_session, and close.
• SQLite backend (sqlite_session.py) provides atomic, ACID-compliant storage using DELETE … RETURNING for rollback operations and supports both file-backed and in-memory databases.
• Redis backend (redis_session.py) uses list operations (LRANGE, RPUSH, RPOP) with optional TTL and pipeline transactions for high-performance distributed sessions.
• Custom backends inherit from SessionABC and can integrate with any storage layer (DynamoDB, PostgreSQL, file systems) without modifying agent code.
• The runtime in session_persistence.py handles deduplication, fingerprinting, and retry rollbacks automatically, treating the session as a black-box storage layer.

Frequently Asked Questions

How do I switch between SQLite and Redis in openai-agents-python?

You switch backends by instantiating the appropriate session class and passing it to the Runner. For SQLite, use SQLiteSession(session_id="id", db_path="file.db"); for Redis, use RedisSession.from_url(session_id="id", url="redis://..."). The agent code remains identical regardless of which backend you choose.

What data format is stored in the session backends?

The session stores response input items—serialized representations of conversation turns including user messages, assistant messages, and tool results. The SQLite backend stores these as JSON strings in the agent_messages table, while Redis stores them as JSON-encoded strings in a list. The runtime handles deduplication using _fingerprint_or_repr before persistence.

Can I use a custom session backend for production workloads?

Yes. Any class implementing the SessionABC interface from src/agents/memory/session.py can be used in production. You must ensure your implementation handles concurrency correctly (e.g., using database transactions, Redis pipelines, or atomic operations) and properly manages connection lifecycles in the close() method. The runtime calls close() when the runner finishes, but you are responsible for connection pooling during the run.

Does the runtime automatically handle retries and rollback with session persistence?

Yes. When a model call fails and requires retry, the runtime in src/agents/run_internal/session_persistence.py calls rewind_session_items(), which invokes the session's pop_item() method to atomically remove the partially saved turn. The runtime then verifies cleanup with wait_for_session_cleanup before retrying. This ensures that failed attempts do not corrupt the conversation history.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}