---
title: "SQLAlchemy Sessions: Production Database-Backed Agent Memory"
description: "Use SQLAlchemySession with PostgreSQL and asyncpg for production-grade persistent agent memory including connection pooling, auto table creation, and migration strategies."
canonical: https://callsphere.ai/blog/sqlalchemy-sessions-production-database-agent-memory
category: "Learn Agentic AI"
tags: ["OpenAI", "SQLAlchemy", "PostgreSQL", "Database", "Production"]
author: "CallSphere Team"
published: 2026-03-14T00:00:00.000Z
updated: 2026-05-06T01:02:41.721Z
---

# SQLAlchemy Sessions: Production Database-Backed Agent Memory

> Use SQLAlchemySession with PostgreSQL and asyncpg for production-grade persistent agent memory including connection pooling, auto table creation, and migration strategies.

## When You Need a Real Database Behind Your Agent

SQLite is great for prototypes. Redis is great for ephemeral, high-speed access. But many production systems need agent memory stored in a real relational database — one that supports ACID transactions, complex queries against session data, backup and recovery, and integration with your existing data infrastructure.

The OpenAI Agents SDK provides `SQLAlchemySession` for exactly this case. It works with any database SQLAlchemy supports — PostgreSQL, MySQL, MariaDB, and SQLite — through async drivers.

## Installation

Install the SQLAlchemy extension:

```mermaid
flowchart TD
    MSG(["New message"])
    WORKING["Working memory
rolling window"]
    EPISODIC[("Episodic memory
past sessions")]
    SEMANTIC[("Semantic memory
facts and preferences")]
    SUM["Summarizer
compresses old turns"]
    ROUTER{"Retrieve
needed memories"}
    PROMPT["Assembled context"]
    LLM["LLM"]
    UPD["Memory updater
writes new facts"]
    MSG --> WORKING --> ROUTER
    ROUTER -->|Past sessions| EPISODIC
    ROUTER -->|User facts| SEMANTIC
    EPISODIC --> SUM --> PROMPT
    SEMANTIC --> PROMPT
    WORKING --> PROMPT --> LLM --> UPD
    UPD --> EPISODIC
    UPD --> SEMANTIC
    style ROUTER fill:#4f46e5,stroke:#4338ca,color:#fff
    style LLM fill:#f59e0b,stroke:#d97706,color:#1f2937
    style EPISODIC fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style SEMANTIC fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
```

```bash
pip install openai-agents[sqlalchemy]
```

For PostgreSQL with the recommended async driver:

```bash
pip install asyncpg
```

For MySQL:

```bash
pip install aiomysql
```

## SQLAlchemySession.from_url() with PostgreSQL

The fastest way to get started is the `from_url()` factory method:

```python
from agents.extensions.sessions import SQLAlchemySession

# PostgreSQL with asyncpg driver
session = await SQLAlchemySession.from_url(
    "postgresql+asyncpg://user:password@localhost:5432/myapp",
    create_tables=True,
)
```

The `create_tables=True` parameter tells the session to create its required tables on first connection. In production, you will want to manage this through migrations instead.

### Full Connection Example

```python
import asyncio
from agents import Agent, Runner
from agents.extensions.sessions import SQLAlchemySession

agent = Agent(
    name="DatabaseAgent",
    instructions="You are an assistant with persistent memory backed by PostgreSQL.",
)

async def main():
    session = await SQLAlchemySession.from_url(
        "postgresql+asyncpg://postgres:secret@db.internal:5432/agents",
        create_tables=True,
    )

    result = await Runner.run(
        agent,
        "Remember that my project deadline is March 30th.",
        session=session,
        session_id="user-789",
    )
    print(result.final_output)

    result = await Runner.run(
        agent,
        "When is my project deadline?",
        session=session,
        session_id="user-789",
    )
    print(result.final_output)  # References March 30th

asyncio.run(main())
```

## Async Database Drivers

SQLAlchemySession uses SQLAlchemy's async engine, which requires an async-compatible database driver. Here are the recommended drivers by database:

| Database | Driver | Connection URL Prefix |
| --- | --- | --- |
| PostgreSQL | asyncpg | `postgresql+asyncpg://` |
| PostgreSQL | psycopg (async) | `postgresql+psycopg://` |
| MySQL | aiomysql | `mysql+aiomysql://` |
| SQLite | aiosqlite | `sqlite+aiosqlite://` |

### Using asyncpg (Recommended for PostgreSQL)

asyncpg is a high-performance async PostgreSQL driver written in Cython. It is significantly faster than psycopg for most workloads:

```python
# asyncpg — fastest option
session = await SQLAlchemySession.from_url(
    "postgresql+asyncpg://user:pass@host:5432/db",
    create_tables=True,
)
```

### Using aiosqlite for Development

During development, you can use SQLAlchemy-backed SQLite for easy local testing:

```python
# Local dev with SQLite
session = await SQLAlchemySession.from_url(
    "sqlite+aiosqlite:///./dev_sessions.db",
    create_tables=True,
)
```

This lets you develop against the same SQLAlchemySession interface you use in production, with a zero-dependency local database.

## Auto-Setup with create_tables=True

When `create_tables=True` is set, SQLAlchemySession automatically creates the required tables if they do not exist. The schema typically includes:

- A `session_items` table with columns for `session_id`, `item_data` (JSON), `item_order`, and `created_at`
- An index on `session_id` for fast retrieval

```python
# First run — tables are created automatically
session = await SQLAlchemySession.from_url(
    "postgresql+asyncpg://user:pass@host/db",
    create_tables=True,
)
```

This is convenient for prototyping, but in a real production environment you should manage schema through migrations.

## Connection Pooling

SQLAlchemy's async engine supports connection pooling out of the box. For high-throughput agent workloads, tuning the pool is essential.

### Configuring Pool Parameters

```python
from sqlalchemy.ext.asyncio import create_async_engine
from agents.extensions.sessions import SQLAlchemySession

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host:5432/db",
    pool_size=20,           # Base pool connections
    max_overflow=10,        # Additional connections under load
    pool_timeout=30,        # Seconds to wait for a connection
    pool_recycle=3600,      # Recycle connections after 1 hour
    pool_pre_ping=True,     # Verify connections before use
)

session = await SQLAlchemySession.from_engine(engine, create_tables=True)
```

### Pool Sizing Guidelines

A good starting formula: `pool_size = num_workers * 2`. If you have 4 uvicorn workers, start with a pool size of 8. Monitor connection wait times and adjust.

```python
# For a FastAPI app with 4 workers
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host:5432/db",
    pool_size=8,
    max_overflow=4,
    pool_pre_ping=True,
)
```

## Migration Strategies

For production deployments, use Alembic to manage the session table schema instead of `create_tables=True`.

### Step 1: Generate the Initial Migration

Create the table manually in an Alembic migration:

```python
"""create agent session tables

Revision ID: 001_agent_sessions
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB

def upgrade():
    op.create_table(
        "session_items",
        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
        sa.Column("session_id", sa.String(255), nullable=False, index=True),
        sa.Column("item_data", JSONB, nullable=False),
        sa.Column("item_order", sa.Integer, nullable=False),
        sa.Column("created_at", sa.DateTime, server_default=sa.func.now()),
    )
    op.create_index(
        "ix_session_items_session_order",
        "session_items",
        ["session_id", "item_order"],
    )

def downgrade():
    op.drop_table("session_items")
```

### Step 2: Use the Session Without Auto-Creation

```python
# Production — tables managed by Alembic
session = await SQLAlchemySession.from_url(
    "postgresql+asyncpg://user:pass@host:5432/db",
    create_tables=False,  # Explicitly disable
)
```

### Step 3: Future Schema Changes

If the SDK updates its schema requirements, create a new Alembic migration to add columns or indexes rather than relying on `create_tables=True` which only handles initial creation.

## Complete Production Setup

Here is a full FastAPI application with SQLAlchemy-backed agent sessions:

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine
from agents import Agent, Runner
from agents.extensions.sessions import SQLAlchemySession, SessionSettings

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@db:5432/agents",
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

session_backend = None
agent = Agent(
    name="ProdAgent",
    instructions="You are a production assistant with database-backed memory.",
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    global session_backend
    session_backend = await SQLAlchemySession.from_engine(
        engine, create_tables=False
    )
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(session_id: str, message: str):
    settings = SessionSettings(limit=50)
    result = await Runner.run(
        agent,
        message,
        session=session_backend,
        session_id=session_id,
        session_settings=settings,
    )
    return {"response": result.final_output}
```

This gives you connection pooling, proper lifecycle management, session limits, and all the durability guarantees of PostgreSQL.

**Sources:**

- [https://openai.github.io/openai-agents-python/sessions/](https://openai.github.io/openai-agents-python/sessions/)
- [https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)

---

Source: https://callsphere.ai/blog/sqlalchemy-sessions-production-database-agent-memory