strands.session.s3_session_manager

S3-based session manager for cloud storage.

S3SessionManager

class S3SessionManager(RepositorySessionManager, SessionRepository)

Defined in: src/strands/session/s3_session_manager.py:29

S3-based session manager for cloud storage.

Creates the following filesystem structure for the session storage:

/<sessions_dir>/
└── session_<session_id>/
    ├── session.json                # Session metadata
    └── agents/
        └── agent_<agent_id>/
            ├── agent.json          # Agent metadata
            └── messages/
                ├── message_<id1>.json
                └── message_<id2>.json

__init__

def __init__(session_id: str,
             bucket: str,
             prefix: str = "",
             boto_session: boto3.Session | None = None,
             boto_client_config: BotocoreConfig | None = None,
             region_name: str | None = None,
             **kwargs: Any)

Defined in: src/strands/session/s3_session_manager.py:46

Initialize S3SessionManager with S3 storage.

Arguments:

• session_id - ID for the session ID is not allowed to contain path separators (e.g., a/b).
• bucket - S3 bucket name (required)
• prefix - S3 key prefix for storage organization
• boto_session - Optional boto3 session
• boto_client_config - Optional boto3 client configuration
• region_name - AWS region for S3 storage
• **kwargs - Additional keyword arguments for future extensibility.

create_session

def create_session(session: Session, **kwargs: Any) -> Session

Defined in: src/strands/session/s3_session_manager.py:158

Create a new session in S3.

read_session

def read_session(session_id: str, **kwargs: Any) -> Session | None

Defined in: src/strands/session/s3_session_manager.py:175

Read session data from S3.

delete_session

def delete_session(session_id: str, **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:183

Delete session and all associated data from S3.

create_agent

def create_agent(session_id: str, session_agent: SessionAgent,
                 **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:206

Create a new agent in S3.

read_agent

def read_agent(session_id: str, agent_id: str,
               **kwargs: Any) -> SessionAgent | None

Defined in: src/strands/session/s3_session_manager.py:213

Read agent data from S3.

update_agent

def update_agent(session_id: str, session_agent: SessionAgent,
                 **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:221

Update agent data in S3.

create_message

def create_message(session_id: str, agent_id: str,
                   session_message: SessionMessage, **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:233

Create a new message in S3.

read_message

def read_message(session_id: str, agent_id: str, message_id: int,
                 **kwargs: Any) -> SessionMessage | None

Defined in: src/strands/session/s3_session_manager.py:240

Read message data from S3.

update_message

def update_message(session_id: str, agent_id: str,
                   session_message: SessionMessage, **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:248

Update message data in S3.

list_messages

def list_messages(session_id: str,
                  agent_id: str,
                  limit: int | None = None,
                  offset: int = 0,
                  **kwargs: Any) -> list[SessionMessage]

Defined in: src/strands/session/s3_session_manager.py:260

List messages for an agent with pagination from S3.

Arguments:

• session_id - ID of the session
• agent_id - ID of the agent
• limit - Optional limit on number of messages to return
• offset - Optional offset for pagination
• **kwargs - Additional keyword arguments

Returns:

List of SessionMessage objects, sorted by message_id.

Raises:

• SessionException - If S3 error occurs during message retrieval.

create_multi_agent

def create_multi_agent(session_id: str, multi_agent: "MultiAgentBase",
                       **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:351

Create a new multiagent state in S3.

read_multi_agent

def read_multi_agent(session_id: str, multi_agent_id: str,
                     **kwargs: Any) -> dict[str, Any] | None

Defined in: src/strands/session/s3_session_manager.py:358

Read multi-agent state from S3.

update_multi_agent

def update_multi_agent(session_id: str, multi_agent: "MultiAgentBase",
                       **kwargs: Any) -> None

Defined in: src/strands/session/s3_session_manager.py:363

Update multi-agent state in S3.