DynamoDBSession

DynamoDB-based session storage for openai-agents.

Class Definition

DynamoDBSession

DynamoDBSession(
    session_id,
    dynamodb_client,
    table_name,
    *,
    ttl_seconds=None,
)

Bases: SessionABC

DynamoDB-based session storage for openai-agents.

Stores conversation history in a DynamoDB table. Each session is stored as a single item with the session_id as the partition key.

The table schema should have: - Partition key: session_id (String)

The session data is stored in a conversation_data attribute as a JSON-encoded list, with an optional ttl attribute for automatic expiration.

Parameters:

Name	Type	Description	Default
session_id	str	Unique identifier for this session.	required
dynamodb_client	DynamoDBClient	An aiobotocore DynamoDB client.	required
table_name	str	Name of the DynamoDB table.	required
ttl_seconds	int | None	Time-to-live in seconds. None means no expiration. Requires TTL to be enabled on the table with attribute name ttl.	None

Example

from aiobotocore.session import get_session
from openai_agents_session import DynamoDBSession

session = get_session()
async with session.create_client("dynamodb", region_name="us-east-1") as client:
    db_session = DynamoDBSession(
        session_id="user-123",
        dynamodb_client=client,
        table_name="agent_sessions",
        ttl_seconds=3600,
    )

Source code in src/openai_agents_session/dynamodb.py

def __init__(
    self,
    session_id: str,
    dynamodb_client: DynamoDBClient,
    table_name: str,
    *,
    ttl_seconds: int | None = None,
) -> None:
    self.session_id = session_id
    self._client = dynamodb_client
    self._table_name = table_name
    self._ttl_seconds = ttl_seconds

get_items async

get_items(limit=None)

Retrieve the conversation history for this session.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum number of items to retrieve. If None, returns all items. When specified, returns the most recent limit items.	None

Returns:

Type	Description
list[TResponseInputItem]	List of conversation items in chronological order.

Source code in src/openai_agents_session/dynamodb.py

async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
    """Retrieve the conversation history for this session.

    Args:
        limit: Maximum number of items to retrieve. If None, returns all items.
               When specified, returns the most recent `limit` items.

    Returns:
        List of conversation items in chronological order.
    """
    items = await self._get_raw_items()

    if limit is not None:
        return items[-limit:]
    return items

add_items async

add_items(items)

Add new items to the conversation history.

Parameters:

Name	Type	Description	Default
items	list[TResponseInputItem]	List of items to add to the session.	required

Source code in src/openai_agents_session/dynamodb.py

async def add_items(self, items: list[TResponseInputItem]) -> None:
    """Add new items to the conversation history.

    Args:
        items: List of items to add to the session.
    """
    if not items:
        return

    existing = await self._get_raw_items()
    existing.extend(items)
    await self._put_items(existing)

pop_item async

pop_item()

Remove and return the most recent item from the session.

Returns:

Type	Description
TResponseInputItem | None	The most recent item, or None if the session is empty.

Source code in src/openai_agents_session/dynamodb.py

async def pop_item(self) -> TResponseInputItem | None:
    """Remove and return the most recent item from the session.

    Returns:
        The most recent item, or None if the session is empty.
    """
    items = await self._get_raw_items()

    if not items:
        return None

    item = items.pop()
    await self._put_items(items)
    return item

clear_session async

clear_session()

Clear all items for this session.

Source code in src/openai_agents_session/dynamodb.py

async def clear_session(self) -> None:
    """Clear all items for this session."""
    await self._client.delete_item(
        TableName=self._table_name,
        Key={"session_id": {"S": self.session_id}},
    )

Usage Example

from aiobotocore.session import get_session
from openai_agents_session import DynamoDBSession

# Create client
boto_session = get_session()
async with boto_session.create_client(
    "dynamodb",
    region_name="us-east-1",
) as client:
    # Create session
    session = DynamoDBSession(
        session_id="user-123",
        dynamodb_client=client,
        table_name="agent_sessions",
        ttl_seconds=3600,
    )

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

    # Get items
    items = await session.get_items()

    # Clear session
    await session.clear_session()

Parameters

session_id

• Type: str
• Required: Yes

Unique identifier for the session. Used as the partition key in DynamoDB.

dynamodb_client

• Type: DynamoDBClient
• Required: Yes

An aiobotocore DynamoDB client instance.

table_name

• Type: str
• Required: Yes

Name of the DynamoDB table to use.

ttl_seconds

• Type: int | None
• Default: None

Time-to-live in seconds. If set, items will be automatically deleted after this duration. Requires TTL to be enabled on the table with attribute name ttl.

Methods

get_items

async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]

Retrieve conversation history.

Parameters:

• limit: Maximum number of items to return. Returns the most recent items.

Returns: List of conversation items in chronological order.

add_items

async def add_items(self, items: list[TResponseInputItem]) -> None

Add new items to the conversation history.

Parameters:

• items: List of items to add.

pop_item

async def pop_item(self) -> TResponseInputItem | None

Remove and return the most recent item.

Returns: The removed item, or None if the session is empty.

clear_session

async def clear_session(self) -> None

Delete the entire session from DynamoDB.

Helper Functions

create_table_if_not_exists

create_table_if_not_exists async

create_table_if_not_exists(
    client, table_name, *, enable_ttl=True
)

Create the DynamoDB table if it doesn't exist.

This is a helper function for development/testing. In production, you should create the table through Infrastructure as Code (Terraform, CDK, etc.).

Parameters:

Name	Type	Description	Default
client	DynamoDBClient	DynamoDB client.	required
table_name	str	Name of the table to create.	required
enable_ttl	bool	Whether to enable TTL on the table.	True

Source code in src/openai_agents_session/dynamodb.py

async def create_table_if_not_exists(
    client: DynamoDBClient,
    table_name: str,
    *,
    enable_ttl: bool = True,
) -> None:
    """Create the DynamoDB table if it doesn't exist.

    This is a helper function for development/testing. In production,
    you should create the table through Infrastructure as Code (Terraform, CDK, etc.).

    Args:
        client: DynamoDB client.
        table_name: Name of the table to create.
        enable_ttl: Whether to enable TTL on the table.
    """
    try:
        await client.describe_table(TableName=table_name)
        return  # Table already exists
    except client.exceptions.ResourceNotFoundException:
        pass

    await client.create_table(
        TableName=table_name,
        KeySchema=[
            {"AttributeName": "session_id", "KeyType": "HASH"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "session_id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )

    # Wait for table to be active
    waiter = client.get_waiter("table_exists")
    await waiter.wait(TableName=table_name)

    if enable_ttl:
        await client.update_time_to_live(
            TableName=table_name,
            TimeToLiveSpecification={
                "Enabled": True,
                "AttributeName": "ttl",
            },
        )

DynamoDB Item Structure

Each session is stored as a single item:

Attribute	Type	Description
session_id	String (S)	Partition key
conversation_data	String (S)	JSON-encoded list of items
updated_at	Number (N)	Unix timestamp
ttl	Number (N)	TTL timestamp (optional)

See Also

• DynamoDB Backend Guide
• Quick Start