Sessions

The Agents SDK provides built-in session memory to automatically maintain conversation history across multiple agent runs, eliminating the need to manually handle .to_input_list() between turns.

Sessions stores conversation history for a specific session, allowing agents to maintain context without requiring explicit manual memory management. This is particularly useful for building chat applications or multi-turn conversations where you want the agent to remember previous interactions.

Use sessions when you want the SDK to manage client-side memory for you. Sessions cannot be combined with conversation_id, previous_response_id, or auto_previous_response_id in the same run. If you want OpenAI server-managed continuation instead, choose one of those mechanisms rather than layering a session on top.

Quick start

from agents import Agent, Runner, SQLiteSession

# Create agent
agent = Agent(
    name="Assistant",
    instructions="Reply very concisely.",
)

# Create a session instance with a session ID
session = SQLiteSession("conversation_123")

# First turn
result = await Runner.run(
    agent,
    "What city is the Golden Gate Bridge in?",
    session=session
)
print(result.final_output)  # "San Francisco"

# Second turn - agent automatically remembers previous context
result = await Runner.run(
    agent,
    "What state is it in?",
    session=session
)
print(result.final_output)  # "California"

# Also works with synchronous runner
result = Runner.run_sync(
    agent,
    "What's the population?",
    session=session
)
print(result.final_output)  # "Approximately 39 million"

Resuming interrupted runs with the same session

If a run pauses for approval, resume it with the same session instance (or another session instance that points at the same backing store) so the resumed turn continues the same stored conversation history.

result = await Runner.run(agent, "Delete temporary files that are no longer needed.", session=session)

if result.interruptions:
    state = result.to_state()
    for interruption in result.interruptions:
        state.approve(interruption)
    result = await Runner.run(agent, state, session=session)

Core session behavior

When session memory is enabled:

1. Before each run: The runner automatically retrieves the conversation history for the session and prepends it to the input items.
2. After each run: All new items generated during the run (user input, assistant responses, tool calls, etc.) are automatically stored in the session.
3. Context preservation: Each subsequent run with the same session includes the full conversation history, allowing the agent to maintain context.

This eliminates the need to manually call .to_input_list() and manage conversation state between runs.

Control how history and new input merge

When you pass a session, the runner normally prepares model input as:

1. Session history (retrieved from session.get_items(...))
2. New turn input

Use RunConfig.session_input_callback to customize that merge step before the model call. The callback receives two lists:

• history: The retrieved session history (already normalized into input-item format)
• new_input: The current turn's new input items

Return the final list of input items that should be sent to the model.

The callback receives copies of both lists, so you can safely mutate them. The returned list controls the model input for that turn, but the SDK still persists only items that belong to the new turn. Reordering or filtering old history therefore does not cause old session items to be saved again as fresh input.

from agents import Agent, RunConfig, Runner, SQLiteSession


def keep_recent_history(history, new_input):
    # Keep only the last 10 history items, then append the new turn.
    return history[-10:] + new_input


agent = Agent(name="Assistant")
session = SQLiteSession("conversation_123")

result = await Runner.run(
    agent,
    "Continue from the latest updates only.",
    session=session,
    run_config=RunConfig(session_input_callback=keep_recent_history),
)

Use this when you need custom pruning, reordering, or selective inclusion of history without changing how the session stores items. If you need a later final pass immediately before the model call, use call_model_input_filter from the running agents guide.

Limiting retrieved history

Use SessionSettings to control how much history is fetched before each run.

• SessionSettings(limit=None) (default): retrieve all available session items
• SessionSettings(limit=N): retrieve only the most recent N items

You can apply this per run via RunConfig.session_settings:

from agents import Agent, RunConfig, Runner, SessionSettings, SQLiteSession

agent = Agent(name="Assistant")
session = SQLiteSession("conversation_123")

result = await Runner.run(
    agent,
    "Summarize our recent discussion.",
    session=session,
    run_config=RunConfig(session_settings=SessionSettings(limit=50)),
)

If your session implementation exposes default session settings, RunConfig.session_settings overrides any non-None values for that run. This is useful for long conversations where you want to cap retrieval size without changing the session's default behavior.

Memory operations

Basic operations

Sessions supports several operations for managing conversation history:

from agents import SQLiteSession

session = SQLiteSession("user_123", "conversations.db")

# Get all items in a session
items = await session.get_items()

# Add new items to a session
new_items = [
    {"role": "user", "content": "Hello"},
    {"role": "assistant", "content": "Hi there!"}
]
await session.add_items(new_items)

# Remove and return the most recent item
last_item = await session.pop_item()
print(last_item)  # {"role": "assistant", "content": "Hi there!"}

# Clear all items from a session
await session.clear_session()

Using pop_item for corrections

The pop_item method is particularly useful when you want to undo or modify the last item in a conversation:

from agents import Agent, Runner, SQLiteSession

agent = Agent(name="Assistant")
session = SQLiteSession("correction_example")

# Initial conversation
result = await Runner.run(
    agent,
    "What's 2 + 2?",
    session=session
)
print(f"Agent: {result.final_output}")

# User wants to correct their question
assistant_item = await session.pop_item()  # Remove agent's response
user_item = await session.pop_item()  # Remove user's question

# Ask a corrected question
result = await Runner.run(
    agent,
    "What's 2 + 3?",
    session=session
)
print(f"Agent: {result.final_output}")

Built-in session implementations

The SDK provides several session implementations for different use cases:

Choose a built-in session implementation

Use this table to pick a starting point before reading the detailed examples below.

Session type	Best for	Notes
SQLiteSession	Local development and simple apps	Built-in, lightweight, file-backed or in-memory
AsyncSQLiteSession	Async SQLite with aiosqlite	Extension backend with async driver support
RedisSession	Shared memory across workers/services	Good for low-latency distributed deployments
SQLAlchemySession	Production apps with existing databases	Works with SQLAlchemy-supported databases
DaprSession	Cloud-native deployments with Dapr sidecars	Supports multiple state stores plus TTL and consistency controls
OpenAIConversationsSession	Server-managed storage in OpenAI	OpenAI Conversations API-backed history
OpenAIResponsesCompactionSession	Long conversations with automatic compaction	Wrapper around another session backend
AdvancedSQLiteSession	SQLite plus branching/analytics	Heavier feature set; see dedicated page
EncryptedSession	Encryption + TTL on top of another session	Wrapper; choose an underlying backend first

Some implementations have dedicated pages with additional details; those are linked inline in their subsections.

If you are implementing a Python server for ChatKit, use a chatkit.store.Store implementation for ChatKit's thread and item persistence. Agents SDK sessions such as SQLAlchemySession manage SDK-side conversation history, but they are not a drop-in replacement for ChatKit's store. See the chatkit-python guide on implementing your ChatKit data store.

OpenAI Conversations API sessions

Use OpenAI's Conversations API through OpenAIConversationsSession.

from agents import Agent, Runner, OpenAIConversationsSession

# Create agent
agent = Agent(
    name="Assistant",
    instructions="Reply very concisely.",
)

# Create a new conversation
session = OpenAIConversationsSession()

# Optionally resume a previous conversation by passing a conversation ID
# session = OpenAIConversationsSession(conversation_id="conv_123")

# Start conversation
result = await Runner.run(
    agent,
    "What city is the Golden Gate Bridge in?",
    session=session
)
print(result.final_output)  # "San Francisco"

# Continue the conversation
result = await Runner.run(
    agent,
    "What state is it in?",
    session=session
)
print(result.final_output)  # "California"

OpenAI Responses compaction sessions

Use OpenAIResponsesCompactionSession to compact stored conversation history with the Responses API (responses.compact). It wraps an underlying session and can automatically compact after each turn based on should_trigger_compaction. Do not wrap OpenAIConversationsSession with it; those two features manage history in different ways.

Typical usage (auto-compaction)

from agents import Agent, Runner, SQLiteSession
from agents.memory import OpenAIResponsesCompactionSession

underlying = SQLiteSession("conversation_123")
session = OpenAIResponsesCompactionSession(
    session_id="conversation_123",
    underlying_session=underlying,
)

agent = Agent(name="Assistant")
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)

By default, compaction runs after each turn once the candidate threshold is reached.

compaction_mode="previous_response_id" works best when you are already chaining turns with Responses API response IDs. compaction_mode="input" rebuilds the compaction request from the current session items instead, which is useful when the response chain is unavailable or you want the session contents to be the source of truth. The default "auto" chooses the safest available option.

auto-compaction can block streaming

Compaction clears and rewrites the session history, so the SDK waits for compaction to finish before considering the run complete. In streaming mode, this means run.stream_events() can stay open for a few seconds after the last output token if compaction is heavy.

If you want low-latency streaming or fast turn-taking, disable auto-compaction and call run_compaction() yourself between turns (or during idle time). You can decide when to force compaction based on your own criteria.

from agents import Agent, Runner, SQLiteSession
from agents.memory import OpenAIResponsesCompactionSession

underlying = SQLiteSession("conversation_123")
session = OpenAIResponsesCompactionSession(
    session_id="conversation_123",
    underlying_session=underlying,
    # Disable triggering the auto compaction
    should_trigger_compaction=lambda _: False,
)

agent = Agent(name="Assistant")
result = await Runner.run(agent, "Hello", session=session)

# Decide when to compact (e.g., on idle, every N turns, or size thresholds).
await session.run_compaction({"force": True})

SQLite sessions

The default, lightweight session implementation using SQLite:

from agents import SQLiteSession

# In-memory database (lost when process ends)
session = SQLiteSession("user_123")

# Persistent file-based database
session = SQLiteSession("user_123", "conversations.db")

# Use the session
result = await Runner.run(
    agent,
    "Hello",
    session=session
)

Async SQLite sessions

Use AsyncSQLiteSession when you want SQLite persistence backed by aiosqlite.

pip install aiosqlite

from agents import Agent, Runner
from agents.extensions.memory import AsyncSQLiteSession

agent = Agent(name="Assistant")
session = AsyncSQLiteSession("user_123", db_path="conversations.db")
result = await Runner.run(agent, "Hello", session=session)

Redis sessions

Use RedisSession for shared session memory across multiple workers or services.

pip install openai-agents[redis]

from agents import Agent, Runner
from agents.extensions.memory import RedisSession

agent = Agent(name="Assistant")
session = RedisSession.from_url(
    "user_123",
    url="redis://localhost:6379/0",
)
result = await Runner.run(agent, "Hello", session=session)

SQLAlchemy sessions

Production-ready Agents SDK session persistence using any SQLAlchemy-supported database:

from agents.extensions.memory import SQLAlchemySession

# Using database URL
session = SQLAlchemySession.from_url(
    "user_123",
    url="postgresql+asyncpg://user:pass@localhost/db",
    create_tables=True
)

# Using existing engine
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
session = SQLAlchemySession("user_123", engine=engine, create_tables=True)

See SQLAlchemy Sessions for detailed documentation.

Dapr sessions

Use DaprSession when you already run Dapr sidecars or want session storage that can move across different state-store backends without changing your agent code.

pip install openai-agents[dapr]

from agents import Agent, Runner
from agents.extensions.memory import DaprSession

agent = Agent(name="Assistant")

async with DaprSession.from_address(
    "user_123",
    state_store_name="statestore",
    dapr_address="localhost:50001",
) as session:
    result = await Runner.run(agent, "Hello", session=session)
    print(result.final_output)

Notes:

• from_address(...) creates and owns the Dapr client for you. If your app already manages one, construct DaprSession(...) directly with dapr_client=....
• Pass ttl=... to let the backing state store expire old session data automatically when the store supports TTL.
• Pass consistency=DAPR_CONSISTENCY_STRONG when you need stronger read-after-write guarantees.
• The Dapr Python SDK also checks the HTTP sidecar endpoint. In local development, start Dapr with --dapr-http-port 3500 as well as the gRPC port used in dapr_address.
• See examples/memory/dapr_session_example.py for a full setup walkthrough, including local components and troubleshooting.

Advanced SQLite sessions

Enhanced SQLite sessions with conversation branching, usage analytics, and structured queries:

from agents.extensions.memory import AdvancedSQLiteSession

# Create with advanced features
session = AdvancedSQLiteSession(
    session_id="user_123",
    db_path="conversations.db",
    create_tables=True
)

# Automatic usage tracking
result = await Runner.run(agent, "Hello", session=session)
await session.store_run_usage(result)  # Track token usage

# Conversation branching
await session.create_branch_from_turn(2)  # Branch from turn 2

See Advanced SQLite Sessions for detailed documentation.

Encrypted sessions

Transparent encryption wrapper for any session implementation:

from agents.extensions.memory import EncryptedSession, SQLAlchemySession

# Create underlying session
underlying_session = SQLAlchemySession.from_url(
    "user_123",
    url="sqlite+aiosqlite:///conversations.db",
    create_tables=True
)

# Wrap with encryption and TTL
session = EncryptedSession(
    session_id="user_123",
    underlying_session=underlying_session,
    encryption_key="your-secret-key",
    ttl=600  # 10 minutes
)

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

See Encrypted Sessions for detailed documentation.

Other session types

There are a few more built-in options. Please refer to examples/memory/ and source code under extensions/memory/.

Operational patterns

Session ID naming

Use meaningful session IDs that help you organize conversations:

• User-based: "user_12345"
• Thread-based: "thread_abc123"
• Context-based: "support_ticket_456"

Memory persistence

• Use in-memory SQLite (SQLiteSession("session_id")) for temporary conversations
• Use file-based SQLite (SQLiteSession("session_id", "path/to/db.sqlite")) for persistent conversations
• Use async SQLite (AsyncSQLiteSession("session_id", db_path="...")) when you need an aiosqlite-based implementation
• Use Redis-backed sessions (RedisSession.from_url("session_id", url="redis://...")) for shared, low-latency session memory
• Use SQLAlchemy-powered sessions (SQLAlchemySession("session_id", engine=engine, create_tables=True)) for production systems with existing databases supported by SQLAlchemy
• Use Dapr state store sessions (DaprSession.from_address("session_id", state_store_name="statestore", dapr_address="localhost:50001")) for production cloud-native deployments with support for 30+ database backends with built-in telemetry, tracing, and data isolation
• Use OpenAI-hosted storage (OpenAIConversationsSession()) when you prefer to store history in the OpenAI Conversations API
• Use encrypted sessions (EncryptedSession(session_id, underlying_session, encryption_key)) to wrap any session with transparent encryption and TTL-based expiration
• Consider implementing custom session backends for other production systems (for example, Django) for more advanced use cases

Multiple sessions

from agents import Agent, Runner, SQLiteSession

agent = Agent(name="Assistant")

# Different sessions maintain separate conversation histories
session_1 = SQLiteSession("user_123", "conversations.db")
session_2 = SQLiteSession("user_456", "conversations.db")

result1 = await Runner.run(
    agent,
    "Help me with my account",
    session=session_1
)
result2 = await Runner.run(
    agent,
    "What are my charges?",
    session=session_2
)

Session sharing

# Different agents can share the same session
support_agent = Agent(name="Support")
billing_agent = Agent(name="Billing")
session = SQLiteSession("user_123")

# Both agents will see the same conversation history
result1 = await Runner.run(
    support_agent,
    "Help me with my account",
    session=session
)
result2 = await Runner.run(
    billing_agent,
    "What are my charges?",
    session=session
)

Complete example

Here's a complete example showing session memory in action:

import asyncio
from agents import Agent, Runner, SQLiteSession


async def main():
    # Create an agent
    agent = Agent(
        name="Assistant",
        instructions="Reply very concisely.",
    )

    # Create a session instance that will persist across runs
    session = SQLiteSession("conversation_123", "conversation_history.db")

    print("=== Sessions Example ===")
    print("The agent will remember previous messages automatically.\n")

    # First turn
    print("First turn:")
    print("User: What city is the Golden Gate Bridge in?")
    result = await Runner.run(
        agent,
        "What city is the Golden Gate Bridge in?",
        session=session
    )
    print(f"Assistant: {result.final_output}")
    print()

    # Second turn - the agent will remember the previous conversation
    print("Second turn:")
    print("User: What state is it in?")
    result = await Runner.run(
        agent,
        "What state is it in?",
        session=session
    )
    print(f"Assistant: {result.final_output}")
    print()

    # Third turn - continuing the conversation
    print("Third turn:")
    print("User: What's the population of that state?")
    result = await Runner.run(
        agent,
        "What's the population of that state?",
        session=session
    )
    print(f"Assistant: {result.final_output}")
    print()

    print("=== Conversation Complete ===")
    print("Notice how the agent remembered the context from previous turns!")
    print("Sessions automatically handles conversation history.")


if __name__ == "__main__":
    asyncio.run(main())

Custom session implementations

You can implement your own session memory by creating a class that follows the Session protocol:

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

class MyCustomSession(SessionABC):
    """Custom session implementation following the Session protocol."""

    def __init__(self, session_id: str):
        self.session_id = session_id
        # Your initialization here

    async def get_items(self, limit: int | None = None) -> List[TResponseInputItem]:
        """Retrieve conversation history for this session."""
        # Your implementation here
        pass

    async def add_items(self, items: List[TResponseInputItem]) -> None:
        """Store new items for this session."""
        # Your implementation here
        pass

    async def pop_item(self) -> TResponseInputItem | None:
        """Remove and return the most recent item from this session."""
        # Your implementation here
        pass

    async def clear_session(self) -> None:
        """Clear all items for this session."""
        # Your implementation here
        pass

# Use your custom session
agent = Agent(name="Assistant")
result = await Runner.run(
    agent,
    "Hello",
    session=MyCustomSession("my_session")
)

Community session implementations

The community has developed additional session implementations:

Package	Description
openai-django-sessions	Django ORM-based sessions for any Django-supported database (PostgreSQL, MySQL, SQLite, and more)

If you've built a session implementation, please feel free to submit a documentation PR to add it here!

API reference

For detailed API documentation, see:

• Session - Protocol interface
• OpenAIConversationsSession - OpenAI Conversations API implementation
• OpenAIResponsesCompactionSession - Responses API compaction wrapper
• SQLiteSession - Basic SQLite implementation
• AsyncSQLiteSession - Async SQLite implementation based on aiosqlite
• RedisSession - Redis-backed session implementation
• SQLAlchemySession - SQLAlchemy-powered implementation
• DaprSession - Dapr state store implementation
• AdvancedSQLiteSession - Enhanced SQLite with branching and analytics
• EncryptedSession - Encrypted wrapper for any session