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

> Learn how openai-agents-python handles session persistence with SQLite, Redis, and custom backends. Explore the abstract async protocol and five core methods for seamless storage swapping.

- Repository: [OpenAI/openai-agents-python](https://github.com/openai/openai-agents-python)
- Tags: deep-dive
- Published: 2026-04-17

---

**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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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.

```python
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`](https://github.com/openai/openai-agents-python/blob/main/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.

```python
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`](https://github.com/openai/openai-agents-python/blob/main/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:

```python
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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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.