tessl/pypi-openai-agents

Lightweight framework for building multi-agent workflows with LLMs, supporting handoffs, guardrails, tools, and 100+ LLM providers

Overview

Eval results

Files

Memory and Sessions

Name: tessl/pypi-openai-agents
Author: tessl

Sessions provide automatic conversation history management across multiple agent runs, eliminating the need to manually track and pass conversation history. The SDK includes built-in session implementations for SQLite, Redis, OpenAI Conversations API, and supports custom session backends.

Capabilities

Session Protocol

Base interface for session implementations.

class SessionABC:
    """
    Base class for sessions.

    Implement this protocol to create custom session storage.
    """

    async def get_items() -> list[TResponseInputItem]:
        """
        Get conversation items from session.

        Returns:
        - list[TResponseInputItem]: Conversation history
        """

    async def add_items(items: list[TResponseInputItem]) -> None:
        """
        Add items to session.

        Parameters:
        - items: Items to add to history
        """

    async def clear() -> None:
        """Clear all items from session."""

Session = SessionABC  # Type alias

SQLite Session

File-based or in-memory session storage using SQLite.

class SQLiteSession(SessionABC):
    """
    SQLite-based session storage.

    Stores conversation history in SQLite database file
    or in-memory database.
    """

    def __init__(session_id: str, db_path: str = ":memory:"):
        """
        Initialize SQLite session.

        Parameters:
        - session_id: Unique session identifier
        - db_path: Path to SQLite database file (default: ":memory:" for in-memory)
        """

    async def get_items() -> list[TResponseInputItem]:
        """Get conversation items."""

    async def add_items(items: list[TResponseInputItem]) -> None:
        """Add items to session."""

    async def clear() -> None:
        """Clear session."""

Usage example:

from agents import Agent, Runner, SQLiteSession

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

# File-based session
session = SQLiteSession("user_123", "conversations.db")

# 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 - history automatically loaded
result = await Runner.run(
    agent,
    "What state is it in?",
    session=session
)
print(result.final_output)  # "California"

# In-memory session (default)
memory_session = SQLiteSession("temp_conversation")

OpenAI Conversations Session

Session using OpenAI's Conversations API for storage.

class OpenAIConversationsSession(SessionABC):
    """
    OpenAI Conversations API session.

    Stores conversation history using OpenAI's
    managed conversation storage.
    """

    def __init__(conversation_id: str, client: AsyncOpenAI):
        """
        Initialize OpenAI Conversations session.

        Parameters:
        - conversation_id: OpenAI conversation ID
        - client: AsyncOpenAI client instance
        """

    async def get_items() -> list[TResponseInputItem]:
        """Get conversation items."""

    async def add_items(items: list[TResponseInputItem]) -> None:
        """Add items to session."""

    async def clear() -> None:
        """Clear session."""

Usage example:

from agents import Agent, Runner, OpenAIConversationsSession
from openai import AsyncOpenAI

client = AsyncOpenAI()
agent = Agent(name="Assistant")

# Use OpenAI Conversations API
session = OpenAIConversationsSession("conv_123", client)

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

Session Input Callback

Customize how session history is merged with current input.

SessionInputCallback = Callable[
    [list[TResponseInputItem], list[TResponseInputItem]],
    MaybeAwaitable[list[TResponseInputItem]]
]

Usage example:

from agents import RunConfig

async def merge_session_input(session_items, current_items):
    """
    Custom logic for merging session history with current input.

    Parameters:
    - session_items: Items from session
    - current_items: Current input items

    Returns:
    - list: Merged items
    """
    # Keep last 10 messages from session
    recent_session = session_items[-10:]
    return recent_session + current_items

config = RunConfig(
    session_input_callback=merge_session_input
)

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

Advanced Session Implementations

Extended session implementations in agents.extensions.memory.

Redis Session

Distributed session storage using Redis.

# In agents.extensions.memory
class RedisSession(SessionABC):
    """
    Redis-based session storage.

    Provides scalable, distributed conversation storage
    suitable for production deployments.

    Requires: pip install 'openai-agents[redis]'
    """

    def __init__(session_id: str, redis_client):
        """
        Initialize Redis session.

        Parameters:
        - session_id: Unique session identifier
        - redis_client: Redis client instance
        """

    @classmethod
    def from_url(session_id: str, url: str):
        """
        Create session from Redis URL.

        Parameters:
        - session_id: Session identifier
        - url: Redis connection URL

        Returns:
        - RedisSession: Configured session
        """

Usage example:

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

# From URL
session = RedisSession.from_url("user_123", "redis://localhost:6379/0")

# Or with client
import redis.asyncio as redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)
session = RedisSession("user_123", redis_client)

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

SQLAlchemy Session

Database-agnostic session using SQLAlchemy ORM.

# In agents.extensions.memory
class SQLAlchemySession(SessionABC):
    """
    SQLAlchemy-based session storage.

    Supports any database backend compatible with SQLAlchemy
    (PostgreSQL, MySQL, SQLite, etc.).

    Requires: pip install sqlalchemy
    """

    def __init__(session_id: str, engine):
        """
        Initialize SQLAlchemy session.

        Parameters:
        - session_id: Session identifier
        - engine: SQLAlchemy engine
        """

Usage example:

from agents import Agent, Runner
from agents.extensions.memory import SQLAlchemySession
from sqlalchemy import create_engine

# PostgreSQL
engine = create_engine("postgresql://user:pass@localhost/dbname")
session = SQLAlchemySession("user_123", engine)

# MySQL
engine = create_engine("mysql://user:pass@localhost/dbname")
session = SQLAlchemySession("user_456", engine)

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

Advanced SQLite Session

Enhanced SQLite session with additional features.

# In agents.extensions.memory
class AdvancedSQLiteSession(SessionABC):
    """
    Advanced SQLite session with additional features.

    Provides enhanced querying, metadata storage,
    and performance optimizations.
    """

    def __init__(session_id: str, db_path: str, options: dict = None):
        """
        Initialize advanced SQLite session.

        Parameters:
        - session_id: Session identifier
        - db_path: Database file path
        - options: Additional configuration options
        """

Dapr Session

Session storage using Dapr state management.

# In agents.extensions.memory
class DaprSession(SessionABC):
    """
    Dapr state store session.

    Integrates with Dapr for cloud-native, portable
    state management across multiple backends.

    Requires: pip install dapr
    """

    def __init__(session_id: str, store_name: str, dapr_client):
        """
        Initialize Dapr session.

        Parameters:
        - session_id: Session identifier
        - store_name: Dapr state store name
        - dapr_client: Dapr client instance
        """

Usage example:

from agents import Agent, Runner
from agents.extensions.memory import DaprSession
from dapr.clients import DaprClient

with DaprClient() as dapr_client:
    session = DaprSession("user_123", "statestore", dapr_client)
    result = await Runner.run(agent, "Hello", session=session)

Encrypted Session Wrapper

Wrapper for encrypting session data at rest.

# In agents.extensions.memory
class EncryptSession:
    """
    Encrypted session wrapper.

    Wraps any session implementation with encryption
    for sensitive conversation data.
    """

    def __init__(inner_session: Session, encryption_key: bytes):
        """
        Initialize encrypted session wrapper.

        Parameters:
        - inner_session: Session to wrap
        - encryption_key: Encryption key for data
        """

Usage example:

from agents import SQLiteSession
from agents.extensions.memory import EncryptSession

# Wrap any session with encryption
base_session = SQLiteSession("user_123", "conversations.db")
encryption_key = b"your-32-byte-encryption-key-here"
encrypted_session = EncryptSession(base_session, encryption_key)

result = await Runner.run(agent, "Sensitive data", session=encrypted_session)

Session Usage Patterns

Basic Session Usage

from agents import Agent, Runner, SQLiteSession

agent = Agent(name="Assistant", instructions="Be helpful.")
session = SQLiteSession("user_123")

# First conversation
result = await Runner.run(agent, "My name is Alice", session=session)

# Later conversation - remembers context
result = await Runner.run(agent, "What's my name?", session=session)
# Response: "Your name is Alice"

Multiple Sessions

# Different users maintain separate histories
user1_session = SQLiteSession("user_1", "app.db")
user2_session = SQLiteSession("user_2", "app.db")

result1 = await Runner.run(agent, "I like Python", session=user1_session)
result2 = await Runner.run(agent, "I like Java", session=user2_session)

# Each session has independent history

Session with Synchronous Runner

# Sessions work with both async and sync runners
session = SQLiteSession("user_123")

result = Runner.run_sync(agent, "Hello", session=session)
result = Runner.run_sync(agent, "How are you?", session=session)

Clearing Session History

# Clear conversation history
await session.clear()

# Fresh conversation
result = await Runner.run(agent, "Start fresh", session=session)

Session Inspection

# Get current session items
items = await session.get_items()
print(f"Session has {len(items)} messages")

# Manually add items (advanced)
await session.add_items([
    {"type": "user", "content": "Custom message"}
])

Custom Session Implementation

Create a custom session backend by implementing the SessionABC protocol:

from agents.memory import SessionABC
from agents import TResponseInputItem

class MyCustomSession(SessionABC):
    """Custom session using your own storage."""

    def __init__(self, session_id: str):
        self.session_id = session_id
        # Initialize your storage

    async def get_items(self) -> list[TResponseInputItem]:
        """Load from your storage."""
        items = await my_storage.load(self.session_id)
        return items

    async def add_items(self, items: list[TResponseInputItem]) -> None:
        """Save to your storage."""
        await my_storage.append(self.session_id, items)

    async def clear(self) -> None:
        """Clear from your storage."""
        await my_storage.delete(self.session_id)

# Use custom session
session = MyCustomSession("user_123")
result = await Runner.run(agent, "Hello", session=session)

Examples of custom storage backends:

Cloud storage (S3, Azure Blob, GCS)
NoSQL databases (MongoDB, DynamoDB)
In-memory caches (Memcached)
Message queues with persistence
Custom encrypted storage

Best Practices

Session IDs: Use stable, unique identifiers (user IDs, conversation IDs)
Database Choice: SQLite for single-server, Redis/PostgreSQL for distributed
Encryption: Use EncryptSession for sensitive conversations
History Limits: Implement custom session_input_callback to limit history size
Cleanup: Periodically clear old sessions to manage storage
Error Handling: Handle session storage failures gracefully
Testing: Test session persistence across restarts
Monitoring: Track session storage performance and size