strands.session.s3_session_manager

S3-based session manager for cloud storage.

AGENT_PREFIX = 'agent_' module-attribute

MESSAGE_PREFIX = 'message_' module-attribute

MULTI_AGENT_PREFIX = 'multi_agent_' module-attribute

SESSION_PREFIX = 'session_' module-attribute

logger = logging.getLogger(__name__) module-attribute

MultiAgentBase

Bases: ABC

Base class for multi-agent helpers.

This class integrates with existing Strands Agent instances and provides multi-agent orchestration capabilities.

Attributes:

Name	Type	Description
id	str	Unique MultiAgent id for session management,etc.

Source code in strands/multiagent/base.py

class MultiAgentBase(ABC):
    """Base class for multi-agent helpers.

    This class integrates with existing Strands Agent instances and provides
    multi-agent orchestration capabilities.

    Attributes:
        id: Unique MultiAgent id for session management,etc.
    """

    id: str

    @abstractmethod
    async def invoke_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> MultiAgentResult:
        """Invoke asynchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.
        """
        raise NotImplementedError("invoke_async not implemented")

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """Stream events during multi-agent execution.

        Default implementation executes invoke_async and yields the result as a single event.
        Subclasses can override this method to provide true streaming capabilities.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.

        Yields:
            Dictionary events containing multi-agent execution information including:
            - Multi-agent coordination events (node start/complete, handoffs)
            - Forwarded single-agent events with node context
            - Final result event
        """
        # Default implementation for backward compatibility
        # Execute invoke_async and yield the result as a single event
        result = await self.invoke_async(task, invocation_state, **kwargs)
        yield {"result": result}

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> MultiAgentResult:
        """Invoke synchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.
        """
        if invocation_state is None:
            invocation_state = {}

        if kwargs:
            invocation_state.update(kwargs)
            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

        return run_async(lambda: self.invoke_async(task, invocation_state))

    def serialize_state(self) -> dict[str, Any]:
        """Return a JSON-serializable snapshot of the orchestrator state."""
        raise NotImplementedError

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore orchestrator state from a session dict."""
        raise NotImplementedError

    def _parse_trace_attributes(
        self, attributes: Mapping[str, AttributeValue] | None = None
    ) -> dict[str, AttributeValue]:
        trace_attributes: dict[str, AttributeValue] = {}
        if attributes:
            for k, v in attributes.items():
                if isinstance(v, (str, int, float, bool)) or (
                    isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
                ):
                    trace_attributes[k] = v
        return trace_attributes

__call__(task, invocation_state=None, **kwargs)

Invoke synchronously.

Parameters:

Name	Type	Description	Default
task	MultiAgentInput	The task to execute	required
invocation_state	dict[str, Any] | None	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	None
**kwargs	Any	Additional keyword arguments passed to underlying agents.	{}

Source code in strands/multiagent/base.py

def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> MultiAgentResult:
    """Invoke synchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.
    """
    if invocation_state is None:
        invocation_state = {}

    if kwargs:
        invocation_state.update(kwargs)
        warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

    return run_async(lambda: self.invoke_async(task, invocation_state))

deserialize_state(payload)

Restore orchestrator state from a session dict.

Source code in strands/multiagent/base.py

def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore orchestrator state from a session dict."""
    raise NotImplementedError

invoke_async(task, invocation_state=None, **kwargs) abstractmethod async

Invoke asynchronously.

Parameters:

Name	Type	Description	Default
task	MultiAgentInput	The task to execute	required
invocation_state	dict[str, Any] | None	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	None
**kwargs	Any	Additional keyword arguments passed to underlying agents.	{}

Source code in strands/multiagent/base.py

@abstractmethod
async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> MultiAgentResult:
    """Invoke asynchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.
    """
    raise NotImplementedError("invoke_async not implemented")

serialize_state()

Return a JSON-serializable snapshot of the orchestrator state.

Source code in strands/multiagent/base.py

def serialize_state(self) -> dict[str, Any]:
    """Return a JSON-serializable snapshot of the orchestrator state."""
    raise NotImplementedError

stream_async(task, invocation_state=None, **kwargs) async

Stream events during multi-agent execution.

Default implementation executes invoke_async and yields the result as a single event. Subclasses can override this method to provide true streaming capabilities.

Parameters:

Name	Type	Description	Default
task	MultiAgentInput	The task to execute	required
invocation_state	dict[str, Any] | None	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	None
**kwargs	Any	Additional keyword arguments passed to underlying agents.	{}

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	Dictionary events containing multi-agent execution information including:
AsyncIterator[dict[str, Any]]	Multi-agent coordination events (node start/complete, handoffs)
AsyncIterator[dict[str, Any]]	Forwarded single-agent events with node context
AsyncIterator[dict[str, Any]]	Final result event

Source code in strands/multiagent/base.py

async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during multi-agent execution.

    Default implementation executes invoke_async and yields the result as a single event.
    Subclasses can override this method to provide true streaming capabilities.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.

    Yields:
        Dictionary events containing multi-agent execution information including:
        - Multi-agent coordination events (node start/complete, handoffs)
        - Forwarded single-agent events with node context
        - Final result event
    """
    # Default implementation for backward compatibility
    # Execute invoke_async and yield the result as a single event
    result = await self.invoke_async(task, invocation_state, **kwargs)
    yield {"result": result}

RepositorySessionManager

Bases: SessionManager

Session manager for persisting agents in a SessionRepository.

Source code in strands/session/repository_session_manager.py

class RepositorySessionManager(SessionManager):
    """Session manager for persisting agents in a SessionRepository."""

    def __init__(
        self,
        session_id: str,
        session_repository: SessionRepository,
        **kwargs: Any,
    ):
        """Initialize the RepositorySessionManager.

        If no session with the specified session_id exists yet, it will be created
        in the session_repository.

        Args:
            session_id: ID to use for the session. A new session with this id will be created if it does
                not exist in the repository yet
            session_repository: Underlying session repository to use to store the sessions state.
            **kwargs: Additional keyword arguments for future extensibility.

        """
        self.session_repository = session_repository
        self.session_id = session_id
        session = session_repository.read_session(session_id)
        # Create a session if it does not exist yet
        if session is None:
            logger.debug("session_id=<%s> | session not found, creating new session", self.session_id)
            session = Session(session_id=session_id, session_type=SessionType.AGENT)
            session_repository.create_session(session)

        self.session = session

        # Keep track of the latest message of each agent in case we need to redact it.
        self._latest_agent_message: dict[str, Optional[SessionMessage]] = {}

    def append_message(self, message: Message, agent: "Agent", **kwargs: Any) -> None:
        """Append a message to the agent's session.

        Args:
            message: Message to add to the agent in the session
            agent: Agent to append the message to
            **kwargs: Additional keyword arguments for future extensibility.
        """
        # Calculate the next index (0 if this is the first message, otherwise increment the previous index)
        latest_agent_message = self._latest_agent_message[agent.agent_id]
        if latest_agent_message:
            next_index = latest_agent_message.message_id + 1
        else:
            next_index = 0

        session_message = SessionMessage.from_message(message, next_index)
        self._latest_agent_message[agent.agent_id] = session_message
        self.session_repository.create_message(self.session_id, agent.agent_id, session_message)

    def redact_latest_message(self, redact_message: Message, agent: "Agent", **kwargs: Any) -> None:
        """Redact the latest message appended to the session.

        Args:
            redact_message: New message to use that contains the redact content
            agent: Agent to apply the message redaction to
            **kwargs: Additional keyword arguments for future extensibility.
        """
        latest_agent_message = self._latest_agent_message[agent.agent_id]
        if latest_agent_message is None:
            raise SessionException("No message to redact.")
        latest_agent_message.redact_message = redact_message
        return self.session_repository.update_message(self.session_id, agent.agent_id, latest_agent_message)

    def sync_agent(self, agent: "Agent", **kwargs: Any) -> None:
        """Serialize and update the agent into the session repository.

        Args:
            agent: Agent to sync to the session.
            **kwargs: Additional keyword arguments for future extensibility.
        """
        self.session_repository.update_agent(
            self.session_id,
            SessionAgent.from_agent(agent),
        )

    def initialize(self, agent: "Agent", **kwargs: Any) -> None:
        """Initialize an agent with a session.

        Args:
            agent: Agent to initialize from the session
            **kwargs: Additional keyword arguments for future extensibility.
        """
        if agent.agent_id in self._latest_agent_message:
            raise SessionException("The `agent_id` of an agent must be unique in a session.")
        self._latest_agent_message[agent.agent_id] = None

        session_agent = self.session_repository.read_agent(self.session_id, agent.agent_id)

        if session_agent is None:
            logger.debug(
                "agent_id=<%s> | session_id=<%s> | creating agent",
                agent.agent_id,
                self.session_id,
            )

            session_agent = SessionAgent.from_agent(agent)
            self.session_repository.create_agent(self.session_id, session_agent)
            # Initialize messages with sequential indices
            session_message = None
            for i, message in enumerate(agent.messages):
                session_message = SessionMessage.from_message(message, i)
                self.session_repository.create_message(self.session_id, agent.agent_id, session_message)
            self._latest_agent_message[agent.agent_id] = session_message
        else:
            logger.debug(
                "agent_id=<%s> | session_id=<%s> | restoring agent",
                agent.agent_id,
                self.session_id,
            )
            agent.state = AgentState(session_agent.state)

            session_agent.initialize_internal_state(agent)

            # Restore the conversation manager to its previous state, and get the optional prepend messages
            prepend_messages = agent.conversation_manager.restore_from_session(session_agent.conversation_manager_state)

            if prepend_messages is None:
                prepend_messages = []

            # List the messages currently in the session, using an offset of the messages previously removed
            # by the conversation manager.
            session_messages = self.session_repository.list_messages(
                session_id=self.session_id,
                agent_id=agent.agent_id,
                offset=agent.conversation_manager.removed_message_count,
            )
            if len(session_messages) > 0:
                self._latest_agent_message[agent.agent_id] = session_messages[-1]

            # Restore the agents messages array including the optional prepend messages
            agent.messages = prepend_messages + [session_message.to_message() for session_message in session_messages]

            # Fix broken session histories: https://github.com/strands-agents/sdk-python/issues/859
            agent.messages = self._fix_broken_tool_use(agent.messages)

    def _fix_broken_tool_use(self, messages: list[Message]) -> list[Message]:
        """Fix broken tool use/result pairs in message history.

        This method handles two issues:
        1. Orphaned toolUse messages without corresponding toolResult.
           Before 1.15.0, strands had a bug where they persisted sessions with a potentially broken messages array.
           This method retroactively fixes that issue by adding a tool_result outside of session management.
           After 1.15.0, this bug is no longer present.
        2. Orphaned toolResult messages without corresponding toolUse (e.g., when pagination truncates messages)

        Args:
            messages: The list of messages to fix
            agent_id: The agent ID for fetching previous messages
            removed_message_count: Number of messages removed by the conversation manager

        Returns:
            Fixed list of messages with proper tool use/result pairs
        """
        # First, check if the oldest message has orphaned toolResult (no preceding toolUse) and remove it.
        if messages:
            first_message = messages[0]
            if first_message["role"] == "user" and any("toolResult" in content for content in first_message["content"]):
                logger.warning(
                    "Session message history starts with orphaned toolResult with no preceding toolUse. "
                    "This typically happens when messages are truncated due to pagination limits. "
                    "Removing orphaned toolResult message to maintain valid conversation structure."
                )
                messages.pop(0)

        # Then check for orphaned toolUse messages
        for index, message in enumerate(messages):
            # Check all but the latest message in the messages array
            # The latest message being orphaned is handled in the agent class
            if index + 1 < len(messages):
                if any("toolUse" in content for content in message["content"]):
                    tool_use_ids = [
                        content["toolUse"]["toolUseId"] for content in message["content"] if "toolUse" in content
                    ]

                    # Check if there are more messages after the current toolUse message
                    tool_result_ids = [
                        content["toolResult"]["toolUseId"]
                        for content in messages[index + 1]["content"]
                        if "toolResult" in content
                    ]

                    missing_tool_use_ids = list(set(tool_use_ids) - set(tool_result_ids))
                    # If there are missing tool use ids, that means the messages history is broken
                    if missing_tool_use_ids:
                        logger.warning(
                            "Session message history has an orphaned toolUse with no toolResult. "
                            "Adding toolResult content blocks to create valid conversation."
                        )
                        # Create the missing toolResult content blocks
                        missing_content_blocks = generate_missing_tool_result_content(missing_tool_use_ids)

                        if tool_result_ids:
                            # If there were any toolResult ids, that means only some of the content blocks are missing
                            messages[index + 1]["content"].extend(missing_content_blocks)
                        else:
                            # The message following the toolUse was not a toolResult, so lets insert it
                            messages.insert(index + 1, {"role": "user", "content": missing_content_blocks})
        return messages

    def sync_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
        """Serialize and update the multi-agent state into the session repository.

        Args:
            source: Multi-agent source object to sync to the session.
            **kwargs: Additional keyword arguments for future extensibility.
        """
        self.session_repository.update_multi_agent(self.session_id, source)

    def initialize_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
        """Initialize multi-agent state from the session repository.

        Args:
            source: Multi-agent source object to restore state into
            **kwargs: Additional keyword arguments for future extensibility.
        """
        state = self.session_repository.read_multi_agent(self.session_id, source.id, **kwargs)
        if state is None:
            self.session_repository.create_multi_agent(self.session_id, source, **kwargs)
        else:
            logger.debug("session_id=<%s> | restoring multi-agent state", self.session_id)
            source.deserialize_state(state)

    def initialize_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
        """Initialize a bidirectional agent with a session.

        Args:
            agent: BidiAgent to initialize from the session
            **kwargs: Additional keyword arguments for future extensibility.
        """
        if agent.agent_id in self._latest_agent_message:
            raise SessionException("The `agent_id` of an agent must be unique in a session.")
        self._latest_agent_message[agent.agent_id] = None

        session_agent = self.session_repository.read_agent(self.session_id, agent.agent_id)

        if session_agent is None:
            logger.debug(
                "agent_id=<%s> | session_id=<%s> | creating bidi agent",
                agent.agent_id,
                self.session_id,
            )

            session_agent = SessionAgent.from_bidi_agent(agent)
            self.session_repository.create_agent(self.session_id, session_agent)
            # Initialize messages with sequential indices
            session_message = None
            for i, message in enumerate(agent.messages):
                session_message = SessionMessage.from_message(message, i)
                self.session_repository.create_message(self.session_id, agent.agent_id, session_message)
            self._latest_agent_message[agent.agent_id] = session_message
        else:
            logger.debug(
                "agent_id=<%s> | session_id=<%s> | restoring bidi agent",
                agent.agent_id,
                self.session_id,
            )
            agent.state = AgentState(session_agent.state)

            session_agent.initialize_bidi_internal_state(agent)

            # BidiAgent has no conversation_manager, so no prepend_messages or removed_message_count
            session_messages = self.session_repository.list_messages(
                session_id=self.session_id,
                agent_id=agent.agent_id,
                offset=0,
            )
            if len(session_messages) > 0:
                self._latest_agent_message[agent.agent_id] = session_messages[-1]

            # Restore the agents messages array
            agent.messages = [session_message.to_message() for session_message in session_messages]

            # Fix broken session histories: https://github.com/strands-agents/sdk-python/issues/859
            agent.messages = self._fix_broken_tool_use(agent.messages)

    def append_bidi_message(self, message: Message, agent: "BidiAgent", **kwargs: Any) -> None:
        """Append a message to the bidirectional agent's session.

        Args:
            message: Message to add to the agent in the session
            agent: BidiAgent to append the message to
            **kwargs: Additional keyword arguments for future extensibility.
        """
        # Calculate the next index (0 if this is the first message, otherwise increment the previous index)
        latest_agent_message = self._latest_agent_message[agent.agent_id]
        if latest_agent_message:
            next_index = latest_agent_message.message_id + 1
        else:
            next_index = 0

        session_message = SessionMessage.from_message(message, next_index)
        self._latest_agent_message[agent.agent_id] = session_message
        self.session_repository.create_message(self.session_id, agent.agent_id, session_message)

    def sync_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
        """Serialize and update the bidirectional agent into the session repository.

        Args:
            agent: BidiAgent to sync to the session.
            **kwargs: Additional keyword arguments for future extensibility.
        """
        self.session_repository.update_agent(
            self.session_id,
            SessionAgent.from_bidi_agent(agent),
        )

__init__(session_id, session_repository, **kwargs)

Initialize the RepositorySessionManager.

If no session with the specified session_id exists yet, it will be created in the session_repository.

Parameters:

Name	Type	Description	Default
session_id	str	ID to use for the session. A new session with this id will be created if it does not exist in the repository yet	required
session_repository	SessionRepository	Underlying session repository to use to store the sessions state.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def __init__(
    self,
    session_id: str,
    session_repository: SessionRepository,
    **kwargs: Any,
):
    """Initialize the RepositorySessionManager.

    If no session with the specified session_id exists yet, it will be created
    in the session_repository.

    Args:
        session_id: ID to use for the session. A new session with this id will be created if it does
            not exist in the repository yet
        session_repository: Underlying session repository to use to store the sessions state.
        **kwargs: Additional keyword arguments for future extensibility.

    """
    self.session_repository = session_repository
    self.session_id = session_id
    session = session_repository.read_session(session_id)
    # Create a session if it does not exist yet
    if session is None:
        logger.debug("session_id=<%s> | session not found, creating new session", self.session_id)
        session = Session(session_id=session_id, session_type=SessionType.AGENT)
        session_repository.create_session(session)

    self.session = session

    # Keep track of the latest message of each agent in case we need to redact it.
    self._latest_agent_message: dict[str, Optional[SessionMessage]] = {}

append_bidi_message(message, agent, **kwargs)

Append a message to the bidirectional agent's session.

Parameters:

Name	Type	Description	Default
message	Message	Message to add to the agent in the session	required
agent	BidiAgent	BidiAgent to append the message to	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def append_bidi_message(self, message: Message, agent: "BidiAgent", **kwargs: Any) -> None:
    """Append a message to the bidirectional agent's session.

    Args:
        message: Message to add to the agent in the session
        agent: BidiAgent to append the message to
        **kwargs: Additional keyword arguments for future extensibility.
    """
    # Calculate the next index (0 if this is the first message, otherwise increment the previous index)
    latest_agent_message = self._latest_agent_message[agent.agent_id]
    if latest_agent_message:
        next_index = latest_agent_message.message_id + 1
    else:
        next_index = 0

    session_message = SessionMessage.from_message(message, next_index)
    self._latest_agent_message[agent.agent_id] = session_message
    self.session_repository.create_message(self.session_id, agent.agent_id, session_message)

append_message(message, agent, **kwargs)

Append a message to the agent's session.

Parameters:

Name	Type	Description	Default
message	Message	Message to add to the agent in the session	required
agent	Agent	Agent to append the message to	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def append_message(self, message: Message, agent: "Agent", **kwargs: Any) -> None:
    """Append a message to the agent's session.

    Args:
        message: Message to add to the agent in the session
        agent: Agent to append the message to
        **kwargs: Additional keyword arguments for future extensibility.
    """
    # Calculate the next index (0 if this is the first message, otherwise increment the previous index)
    latest_agent_message = self._latest_agent_message[agent.agent_id]
    if latest_agent_message:
        next_index = latest_agent_message.message_id + 1
    else:
        next_index = 0

    session_message = SessionMessage.from_message(message, next_index)
    self._latest_agent_message[agent.agent_id] = session_message
    self.session_repository.create_message(self.session_id, agent.agent_id, session_message)

initialize(agent, **kwargs)

Initialize an agent with a session.

Parameters:

Name	Type	Description	Default
agent	Agent	Agent to initialize from the session	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def initialize(self, agent: "Agent", **kwargs: Any) -> None:
    """Initialize an agent with a session.

    Args:
        agent: Agent to initialize from the session
        **kwargs: Additional keyword arguments for future extensibility.
    """
    if agent.agent_id in self._latest_agent_message:
        raise SessionException("The `agent_id` of an agent must be unique in a session.")
    self._latest_agent_message[agent.agent_id] = None

    session_agent = self.session_repository.read_agent(self.session_id, agent.agent_id)

    if session_agent is None:
        logger.debug(
            "agent_id=<%s> | session_id=<%s> | creating agent",
            agent.agent_id,
            self.session_id,
        )

        session_agent = SessionAgent.from_agent(agent)
        self.session_repository.create_agent(self.session_id, session_agent)
        # Initialize messages with sequential indices
        session_message = None
        for i, message in enumerate(agent.messages):
            session_message = SessionMessage.from_message(message, i)
            self.session_repository.create_message(self.session_id, agent.agent_id, session_message)
        self._latest_agent_message[agent.agent_id] = session_message
    else:
        logger.debug(
            "agent_id=<%s> | session_id=<%s> | restoring agent",
            agent.agent_id,
            self.session_id,
        )
        agent.state = AgentState(session_agent.state)

        session_agent.initialize_internal_state(agent)

        # Restore the conversation manager to its previous state, and get the optional prepend messages
        prepend_messages = agent.conversation_manager.restore_from_session(session_agent.conversation_manager_state)

        if prepend_messages is None:
            prepend_messages = []

        # List the messages currently in the session, using an offset of the messages previously removed
        # by the conversation manager.
        session_messages = self.session_repository.list_messages(
            session_id=self.session_id,
            agent_id=agent.agent_id,
            offset=agent.conversation_manager.removed_message_count,
        )
        if len(session_messages) > 0:
            self._latest_agent_message[agent.agent_id] = session_messages[-1]

        # Restore the agents messages array including the optional prepend messages
        agent.messages = prepend_messages + [session_message.to_message() for session_message in session_messages]

        # Fix broken session histories: https://github.com/strands-agents/sdk-python/issues/859
        agent.messages = self._fix_broken_tool_use(agent.messages)

initialize_bidi_agent(agent, **kwargs)

Initialize a bidirectional agent with a session.

Parameters:

Name	Type	Description	Default
agent	BidiAgent	BidiAgent to initialize from the session	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def initialize_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
    """Initialize a bidirectional agent with a session.

    Args:
        agent: BidiAgent to initialize from the session
        **kwargs: Additional keyword arguments for future extensibility.
    """
    if agent.agent_id in self._latest_agent_message:
        raise SessionException("The `agent_id` of an agent must be unique in a session.")
    self._latest_agent_message[agent.agent_id] = None

    session_agent = self.session_repository.read_agent(self.session_id, agent.agent_id)

    if session_agent is None:
        logger.debug(
            "agent_id=<%s> | session_id=<%s> | creating bidi agent",
            agent.agent_id,
            self.session_id,
        )

        session_agent = SessionAgent.from_bidi_agent(agent)
        self.session_repository.create_agent(self.session_id, session_agent)
        # Initialize messages with sequential indices
        session_message = None
        for i, message in enumerate(agent.messages):
            session_message = SessionMessage.from_message(message, i)
            self.session_repository.create_message(self.session_id, agent.agent_id, session_message)
        self._latest_agent_message[agent.agent_id] = session_message
    else:
        logger.debug(
            "agent_id=<%s> | session_id=<%s> | restoring bidi agent",
            agent.agent_id,
            self.session_id,
        )
        agent.state = AgentState(session_agent.state)

        session_agent.initialize_bidi_internal_state(agent)

        # BidiAgent has no conversation_manager, so no prepend_messages or removed_message_count
        session_messages = self.session_repository.list_messages(
            session_id=self.session_id,
            agent_id=agent.agent_id,
            offset=0,
        )
        if len(session_messages) > 0:
            self._latest_agent_message[agent.agent_id] = session_messages[-1]

        # Restore the agents messages array
        agent.messages = [session_message.to_message() for session_message in session_messages]

        # Fix broken session histories: https://github.com/strands-agents/sdk-python/issues/859
        agent.messages = self._fix_broken_tool_use(agent.messages)

initialize_multi_agent(source, **kwargs)

Initialize multi-agent state from the session repository.

Parameters:

Name	Type	Description	Default
source	MultiAgentBase	Multi-agent source object to restore state into	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def initialize_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
    """Initialize multi-agent state from the session repository.

    Args:
        source: Multi-agent source object to restore state into
        **kwargs: Additional keyword arguments for future extensibility.
    """
    state = self.session_repository.read_multi_agent(self.session_id, source.id, **kwargs)
    if state is None:
        self.session_repository.create_multi_agent(self.session_id, source, **kwargs)
    else:
        logger.debug("session_id=<%s> | restoring multi-agent state", self.session_id)
        source.deserialize_state(state)

redact_latest_message(redact_message, agent, **kwargs)

Redact the latest message appended to the session.

Parameters:

Name	Type	Description	Default
redact_message	Message	New message to use that contains the redact content	required
agent	Agent	Agent to apply the message redaction to	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def redact_latest_message(self, redact_message: Message, agent: "Agent", **kwargs: Any) -> None:
    """Redact the latest message appended to the session.

    Args:
        redact_message: New message to use that contains the redact content
        agent: Agent to apply the message redaction to
        **kwargs: Additional keyword arguments for future extensibility.
    """
    latest_agent_message = self._latest_agent_message[agent.agent_id]
    if latest_agent_message is None:
        raise SessionException("No message to redact.")
    latest_agent_message.redact_message = redact_message
    return self.session_repository.update_message(self.session_id, agent.agent_id, latest_agent_message)

sync_agent(agent, **kwargs)

Serialize and update the agent into the session repository.

Parameters:

Name	Type	Description	Default
agent	Agent	Agent to sync to the session.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def sync_agent(self, agent: "Agent", **kwargs: Any) -> None:
    """Serialize and update the agent into the session repository.

    Args:
        agent: Agent to sync to the session.
        **kwargs: Additional keyword arguments for future extensibility.
    """
    self.session_repository.update_agent(
        self.session_id,
        SessionAgent.from_agent(agent),
    )

sync_bidi_agent(agent, **kwargs)

Serialize and update the bidirectional agent into the session repository.

Parameters:

Name	Type	Description	Default
agent	BidiAgent	BidiAgent to sync to the session.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def sync_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
    """Serialize and update the bidirectional agent into the session repository.

    Args:
        agent: BidiAgent to sync to the session.
        **kwargs: Additional keyword arguments for future extensibility.
    """
    self.session_repository.update_agent(
        self.session_id,
        SessionAgent.from_bidi_agent(agent),
    )

sync_multi_agent(source, **kwargs)

Serialize and update the multi-agent state into the session repository.

Parameters:

Name	Type	Description	Default
source	MultiAgentBase	Multi-agent source object to sync to the session.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/repository_session_manager.py

def sync_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
    """Serialize and update the multi-agent state into the session repository.

    Args:
        source: Multi-agent source object to sync to the session.
        **kwargs: Additional keyword arguments for future extensibility.
    """
    self.session_repository.update_multi_agent(self.session_id, source)

S3SessionManager

Bases: RepositorySessionManager, SessionRepository

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

Source code in strands/session/s3_session_manager.py

class S3SessionManager(RepositorySessionManager, SessionRepository):
    """S3-based session manager for cloud storage.

    Creates the following filesystem structure for the session storage:
    ```bash
    /<sessions_dir>/
    └── session_<session_id>/
        ├── session.json                # Session metadata
        └── agents/
            └── agent_<agent_id>/
                ├── agent.json          # Agent metadata
                └── messages/
                    ├── message_<id1>.json
                    └── message_<id2>.json
    ```
    """

    def __init__(
        self,
        session_id: str,
        bucket: str,
        prefix: str = "",
        boto_session: Optional[boto3.Session] = None,
        boto_client_config: Optional[BotocoreConfig] = None,
        region_name: Optional[str] = None,
        **kwargs: Any,
    ):
        """Initialize S3SessionManager with S3 storage.

        Args:
            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.
        """
        self.bucket = bucket
        self.prefix = prefix

        session = boto_session or boto3.Session(region_name=region_name)

        # Add strands-agents to the request user agent
        if boto_client_config:
            existing_user_agent = getattr(boto_client_config, "user_agent_extra", None)
            # Append 'strands-agents' to existing user_agent_extra or set it if not present
            if existing_user_agent:
                new_user_agent = f"{existing_user_agent} strands-agents"
            else:
                new_user_agent = "strands-agents"
            client_config = boto_client_config.merge(BotocoreConfig(user_agent_extra=new_user_agent))
        else:
            client_config = BotocoreConfig(user_agent_extra="strands-agents")

        self.client = session.client(service_name="s3", config=client_config)
        super().__init__(session_id=session_id, session_repository=self)

    def _get_session_path(self, session_id: str) -> str:
        """Get session S3 prefix.

        Args:
            session_id: ID for the session.

        Raises:
            ValueError: If session id contains a path separator.
        """
        session_id = _identifier.validate(session_id, _identifier.Identifier.SESSION)
        return f"{self.prefix}/{SESSION_PREFIX}{session_id}/"

    def _get_agent_path(self, session_id: str, agent_id: str) -> str:
        """Get agent S3 prefix.

        Args:
            session_id: ID for the session.
            agent_id: ID for the agent.

        Raises:
            ValueError: If session id or agent id contains a path separator.
        """
        session_path = self._get_session_path(session_id)
        agent_id = _identifier.validate(agent_id, _identifier.Identifier.AGENT)
        return f"{session_path}agents/{AGENT_PREFIX}{agent_id}/"

    def _get_message_path(self, session_id: str, agent_id: str, message_id: int) -> str:
        """Get message S3 key.

        Args:
            session_id: ID of the session
            agent_id: ID of the agent
            message_id: Index of the message

        Returns:
            The key for the message

        Raises:
            ValueError: If message_id is not an integer.
        """
        if not isinstance(message_id, int):
            raise ValueError(f"message_id=<{message_id}> | message id must be an integer")

        agent_path = self._get_agent_path(session_id, agent_id)
        return f"{agent_path}messages/{MESSAGE_PREFIX}{message_id}.json"

    def _read_s3_object(self, key: str) -> Optional[Dict[str, Any]]:
        """Read JSON object from S3."""
        try:
            response = self.client.get_object(Bucket=self.bucket, Key=key)
            content = response["Body"].read().decode("utf-8")
            return cast(dict[str, Any], json.loads(content))
        except ClientError as e:
            if e.response["Error"]["Code"] == "NoSuchKey":
                return None
            else:
                raise SessionException(f"S3 error reading {key}: {e}") from e
        except json.JSONDecodeError as e:
            raise SessionException(f"Invalid JSON in S3 object {key}: {e}") from e

    def _write_s3_object(self, key: str, data: Dict[str, Any]) -> None:
        """Write JSON object to S3."""
        try:
            content = json.dumps(data, indent=2, ensure_ascii=False)
            self.client.put_object(
                Bucket=self.bucket, Key=key, Body=content.encode("utf-8"), ContentType="application/json"
            )
        except ClientError as e:
            raise SessionException(f"Failed to write S3 object {key}: {e}") from e

    def create_session(self, session: Session, **kwargs: Any) -> Session:
        """Create a new session in S3."""
        session_key = f"{self._get_session_path(session.session_id)}session.json"

        # Check if session already exists
        try:
            self.client.head_object(Bucket=self.bucket, Key=session_key)
            raise SessionException(f"Session {session.session_id} already exists")
        except ClientError as e:
            if e.response["Error"]["Code"] != "404":
                raise SessionException(f"S3 error checking session existence: {e}") from e

        # Write session object
        session_dict = session.to_dict()
        self._write_s3_object(session_key, session_dict)
        return session

    def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
        """Read session data from S3."""
        session_key = f"{self._get_session_path(session_id)}session.json"
        session_data = self._read_s3_object(session_key)
        if session_data is None:
            return None
        return Session.from_dict(session_data)

    def delete_session(self, session_id: str, **kwargs: Any) -> None:
        """Delete session and all associated data from S3."""
        session_prefix = self._get_session_path(session_id)
        try:
            paginator = self.client.get_paginator("list_objects_v2")
            pages = paginator.paginate(Bucket=self.bucket, Prefix=session_prefix)

            objects_to_delete = []
            for page in pages:
                if "Contents" in page:
                    objects_to_delete.extend([{"Key": obj["Key"]} for obj in page["Contents"]])

            if not objects_to_delete:
                raise SessionException(f"Session {session_id} does not exist")

            # Delete objects in batches
            for i in range(0, len(objects_to_delete), 1000):
                batch = objects_to_delete[i : i + 1000]
                self.client.delete_objects(Bucket=self.bucket, Delete={"Objects": batch})

        except ClientError as e:
            raise SessionException(f"S3 error deleting session {session_id}: {e}") from e

    def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
        """Create a new agent in S3."""
        agent_id = session_agent.agent_id
        agent_dict = session_agent.to_dict()
        agent_key = f"{self._get_agent_path(session_id, agent_id)}agent.json"
        self._write_s3_object(agent_key, agent_dict)

    def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[SessionAgent]:
        """Read agent data from S3."""
        agent_key = f"{self._get_agent_path(session_id, agent_id)}agent.json"
        agent_data = self._read_s3_object(agent_key)
        if agent_data is None:
            return None
        return SessionAgent.from_dict(agent_data)

    def update_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
        """Update agent data in S3."""
        agent_id = session_agent.agent_id
        previous_agent = self.read_agent(session_id=session_id, agent_id=agent_id)
        if previous_agent is None:
            raise SessionException(f"Agent {agent_id} in session {session_id} does not exist")

        # Preserve creation timestamp
        session_agent.created_at = previous_agent.created_at
        agent_key = f"{self._get_agent_path(session_id, agent_id)}agent.json"
        self._write_s3_object(agent_key, session_agent.to_dict())

    def create_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
        """Create a new message in S3."""
        message_id = session_message.message_id
        message_dict = session_message.to_dict()
        message_key = self._get_message_path(session_id, agent_id, message_id)
        self._write_s3_object(message_key, message_dict)

    def read_message(self, session_id: str, agent_id: str, message_id: int, **kwargs: Any) -> Optional[SessionMessage]:
        """Read message data from S3."""
        message_key = self._get_message_path(session_id, agent_id, message_id)
        message_data = self._read_s3_object(message_key)
        if message_data is None:
            return None
        return SessionMessage.from_dict(message_data)

    def update_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
        """Update message data in S3."""
        message_id = session_message.message_id
        previous_message = self.read_message(session_id=session_id, agent_id=agent_id, message_id=message_id)
        if previous_message is None:
            raise SessionException(f"Message {message_id} does not exist")

        # Preserve creation timestamp
        session_message.created_at = previous_message.created_at
        message_key = self._get_message_path(session_id, agent_id, message_id)
        self._write_s3_object(message_key, session_message.to_dict())

    def list_messages(
        self, session_id: str, agent_id: str, limit: Optional[int] = None, offset: int = 0, **kwargs: Any
    ) -> List[SessionMessage]:
        """List messages for an agent with pagination from S3."""
        messages_prefix = f"{self._get_agent_path(session_id, agent_id)}messages/"
        try:
            paginator = self.client.get_paginator("list_objects_v2")
            pages = paginator.paginate(Bucket=self.bucket, Prefix=messages_prefix)

            # Collect all message keys and extract their indices
            message_index_keys: list[tuple[int, str]] = []
            for page in pages:
                if "Contents" in page:
                    for obj in page["Contents"]:
                        key = obj["Key"]
                        if key.endswith(".json") and MESSAGE_PREFIX in key:
                            # Extract the filename part from the full S3 key
                            filename = key.split("/")[-1]
                            # Extract index from message_<index>.json format
                            index = int(filename[len(MESSAGE_PREFIX) : -5])  # Remove prefix and .json suffix
                            message_index_keys.append((index, key))

            # Sort by index and extract just the keys
            message_keys = [k for _, k in sorted(message_index_keys)]

            # Apply pagination to keys before loading content
            if limit is not None:
                message_keys = message_keys[offset : offset + limit]
            else:
                message_keys = message_keys[offset:]

            # Load only the required message objects
            messages: List[SessionMessage] = []
            for key in message_keys:
                message_data = self._read_s3_object(key)
                if message_data:
                    messages.append(SessionMessage.from_dict(message_data))

            return messages

        except ClientError as e:
            raise SessionException(f"S3 error reading messages: {e}") from e

    def _get_multi_agent_path(self, session_id: str, multi_agent_id: str) -> str:
        """Get multi-agent S3 prefix."""
        session_path = self._get_session_path(session_id)
        multi_agent_id = _identifier.validate(multi_agent_id, _identifier.Identifier.AGENT)
        return f"{session_path}multi_agents/{MULTI_AGENT_PREFIX}{multi_agent_id}/"

    def create_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
        """Create a new multiagent state in S3."""
        multi_agent_id = multi_agent.id
        multi_agent_key = f"{self._get_multi_agent_path(session_id, multi_agent_id)}multi_agent.json"
        session_data = multi_agent.serialize_state()
        self._write_s3_object(multi_agent_key, session_data)

    def read_multi_agent(self, session_id: str, multi_agent_id: str, **kwargs: Any) -> Optional[dict[str, Any]]:
        """Read multi-agent state from S3."""
        multi_agent_key = f"{self._get_multi_agent_path(session_id, multi_agent_id)}multi_agent.json"
        return self._read_s3_object(multi_agent_key)

    def update_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
        """Update multi-agent state in S3."""
        multi_agent_state = multi_agent.serialize_state()
        previous_multi_agent_state = self.read_multi_agent(session_id=session_id, multi_agent_id=multi_agent.id)
        if previous_multi_agent_state is None:
            raise SessionException(f"MultiAgent state {multi_agent.id} in session {session_id} does not exist")

        multi_agent_key = f"{self._get_multi_agent_path(session_id, multi_agent.id)}multi_agent.json"
        self._write_s3_object(multi_agent_key, multi_agent_state)

__init__(session_id, bucket, prefix='', boto_session=None, boto_client_config=None, region_name=None, **kwargs)

Initialize S3SessionManager with S3 storage.

Parameters:

Name	Type	Description	Default
session_id	str	ID for the session ID is not allowed to contain path separators (e.g., a/b).	required
bucket	str	S3 bucket name (required)	required
prefix	str	S3 key prefix for storage organization	''
boto_session	Optional[Session]	Optional boto3 session	None
boto_client_config	Optional[Config]	Optional boto3 client configuration	None
region_name	Optional[str]	AWS region for S3 storage	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Source code in strands/session/s3_session_manager.py

def __init__(
    self,
    session_id: str,
    bucket: str,
    prefix: str = "",
    boto_session: Optional[boto3.Session] = None,
    boto_client_config: Optional[BotocoreConfig] = None,
    region_name: Optional[str] = None,
    **kwargs: Any,
):
    """Initialize S3SessionManager with S3 storage.

    Args:
        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.
    """
    self.bucket = bucket
    self.prefix = prefix

    session = boto_session or boto3.Session(region_name=region_name)

    # Add strands-agents to the request user agent
    if boto_client_config:
        existing_user_agent = getattr(boto_client_config, "user_agent_extra", None)
        # Append 'strands-agents' to existing user_agent_extra or set it if not present
        if existing_user_agent:
            new_user_agent = f"{existing_user_agent} strands-agents"
        else:
            new_user_agent = "strands-agents"
        client_config = boto_client_config.merge(BotocoreConfig(user_agent_extra=new_user_agent))
    else:
        client_config = BotocoreConfig(user_agent_extra="strands-agents")

    self.client = session.client(service_name="s3", config=client_config)
    super().__init__(session_id=session_id, session_repository=self)

create_agent(session_id, session_agent, **kwargs)

Create a new agent in S3.

Source code in strands/session/s3_session_manager.py

def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
    """Create a new agent in S3."""
    agent_id = session_agent.agent_id
    agent_dict = session_agent.to_dict()
    agent_key = f"{self._get_agent_path(session_id, agent_id)}agent.json"
    self._write_s3_object(agent_key, agent_dict)

create_message(session_id, agent_id, session_message, **kwargs)

Create a new message in S3.

Source code in strands/session/s3_session_manager.py

def create_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
    """Create a new message in S3."""
    message_id = session_message.message_id
    message_dict = session_message.to_dict()
    message_key = self._get_message_path(session_id, agent_id, message_id)
    self._write_s3_object(message_key, message_dict)

create_multi_agent(session_id, multi_agent, **kwargs)

Create a new multiagent state in S3.

Source code in strands/session/s3_session_manager.py

def create_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
    """Create a new multiagent state in S3."""
    multi_agent_id = multi_agent.id
    multi_agent_key = f"{self._get_multi_agent_path(session_id, multi_agent_id)}multi_agent.json"
    session_data = multi_agent.serialize_state()
    self._write_s3_object(multi_agent_key, session_data)

create_session(session, **kwargs)

Create a new session in S3.

Source code in strands/session/s3_session_manager.py

def create_session(self, session: Session, **kwargs: Any) -> Session:
    """Create a new session in S3."""
    session_key = f"{self._get_session_path(session.session_id)}session.json"

    # Check if session already exists
    try:
        self.client.head_object(Bucket=self.bucket, Key=session_key)
        raise SessionException(f"Session {session.session_id} already exists")
    except ClientError as e:
        if e.response["Error"]["Code"] != "404":
            raise SessionException(f"S3 error checking session existence: {e}") from e

    # Write session object
    session_dict = session.to_dict()
    self._write_s3_object(session_key, session_dict)
    return session

delete_session(session_id, **kwargs)

Delete session and all associated data from S3.

Source code in strands/session/s3_session_manager.py

def delete_session(self, session_id: str, **kwargs: Any) -> None:
    """Delete session and all associated data from S3."""
    session_prefix = self._get_session_path(session_id)
    try:
        paginator = self.client.get_paginator("list_objects_v2")
        pages = paginator.paginate(Bucket=self.bucket, Prefix=session_prefix)

        objects_to_delete = []
        for page in pages:
            if "Contents" in page:
                objects_to_delete.extend([{"Key": obj["Key"]} for obj in page["Contents"]])

        if not objects_to_delete:
            raise SessionException(f"Session {session_id} does not exist")

        # Delete objects in batches
        for i in range(0, len(objects_to_delete), 1000):
            batch = objects_to_delete[i : i + 1000]
            self.client.delete_objects(Bucket=self.bucket, Delete={"Objects": batch})

    except ClientError as e:
        raise SessionException(f"S3 error deleting session {session_id}: {e}") from e

list_messages(session_id, agent_id, limit=None, offset=0, **kwargs)

List messages for an agent with pagination from S3.

Source code in strands/session/s3_session_manager.py

def list_messages(
    self, session_id: str, agent_id: str, limit: Optional[int] = None, offset: int = 0, **kwargs: Any
) -> List[SessionMessage]:
    """List messages for an agent with pagination from S3."""
    messages_prefix = f"{self._get_agent_path(session_id, agent_id)}messages/"
    try:
        paginator = self.client.get_paginator("list_objects_v2")
        pages = paginator.paginate(Bucket=self.bucket, Prefix=messages_prefix)

        # Collect all message keys and extract their indices
        message_index_keys: list[tuple[int, str]] = []
        for page in pages:
            if "Contents" in page:
                for obj in page["Contents"]:
                    key = obj["Key"]
                    if key.endswith(".json") and MESSAGE_PREFIX in key:
                        # Extract the filename part from the full S3 key
                        filename = key.split("/")[-1]
                        # Extract index from message_<index>.json format
                        index = int(filename[len(MESSAGE_PREFIX) : -5])  # Remove prefix and .json suffix
                        message_index_keys.append((index, key))

        # Sort by index and extract just the keys
        message_keys = [k for _, k in sorted(message_index_keys)]

        # Apply pagination to keys before loading content
        if limit is not None:
            message_keys = message_keys[offset : offset + limit]
        else:
            message_keys = message_keys[offset:]

        # Load only the required message objects
        messages: List[SessionMessage] = []
        for key in message_keys:
            message_data = self._read_s3_object(key)
            if message_data:
                messages.append(SessionMessage.from_dict(message_data))

        return messages

    except ClientError as e:
        raise SessionException(f"S3 error reading messages: {e}") from e

read_agent(session_id, agent_id, **kwargs)

Read agent data from S3.

Source code in strands/session/s3_session_manager.py

def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[SessionAgent]:
    """Read agent data from S3."""
    agent_key = f"{self._get_agent_path(session_id, agent_id)}agent.json"
    agent_data = self._read_s3_object(agent_key)
    if agent_data is None:
        return None
    return SessionAgent.from_dict(agent_data)

read_message(session_id, agent_id, message_id, **kwargs)

Read message data from S3.

Source code in strands/session/s3_session_manager.py

def read_message(self, session_id: str, agent_id: str, message_id: int, **kwargs: Any) -> Optional[SessionMessage]:
    """Read message data from S3."""
    message_key = self._get_message_path(session_id, agent_id, message_id)
    message_data = self._read_s3_object(message_key)
    if message_data is None:
        return None
    return SessionMessage.from_dict(message_data)

read_multi_agent(session_id, multi_agent_id, **kwargs)

Read multi-agent state from S3.

Source code in strands/session/s3_session_manager.py

def read_multi_agent(self, session_id: str, multi_agent_id: str, **kwargs: Any) -> Optional[dict[str, Any]]:
    """Read multi-agent state from S3."""
    multi_agent_key = f"{self._get_multi_agent_path(session_id, multi_agent_id)}multi_agent.json"
    return self._read_s3_object(multi_agent_key)

read_session(session_id, **kwargs)

Read session data from S3.

Source code in strands/session/s3_session_manager.py

def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
    """Read session data from S3."""
    session_key = f"{self._get_session_path(session_id)}session.json"
    session_data = self._read_s3_object(session_key)
    if session_data is None:
        return None
    return Session.from_dict(session_data)

update_agent(session_id, session_agent, **kwargs)

Update agent data in S3.

Source code in strands/session/s3_session_manager.py

def update_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
    """Update agent data in S3."""
    agent_id = session_agent.agent_id
    previous_agent = self.read_agent(session_id=session_id, agent_id=agent_id)
    if previous_agent is None:
        raise SessionException(f"Agent {agent_id} in session {session_id} does not exist")

    # Preserve creation timestamp
    session_agent.created_at = previous_agent.created_at
    agent_key = f"{self._get_agent_path(session_id, agent_id)}agent.json"
    self._write_s3_object(agent_key, session_agent.to_dict())

update_message(session_id, agent_id, session_message, **kwargs)

Update message data in S3.

Source code in strands/session/s3_session_manager.py

def update_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
    """Update message data in S3."""
    message_id = session_message.message_id
    previous_message = self.read_message(session_id=session_id, agent_id=agent_id, message_id=message_id)
    if previous_message is None:
        raise SessionException(f"Message {message_id} does not exist")

    # Preserve creation timestamp
    session_message.created_at = previous_message.created_at
    message_key = self._get_message_path(session_id, agent_id, message_id)
    self._write_s3_object(message_key, session_message.to_dict())

update_multi_agent(session_id, multi_agent, **kwargs)

Update multi-agent state in S3.

Source code in strands/session/s3_session_manager.py

def update_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
    """Update multi-agent state in S3."""
    multi_agent_state = multi_agent.serialize_state()
    previous_multi_agent_state = self.read_multi_agent(session_id=session_id, multi_agent_id=multi_agent.id)
    if previous_multi_agent_state is None:
        raise SessionException(f"MultiAgent state {multi_agent.id} in session {session_id} does not exist")

    multi_agent_key = f"{self._get_multi_agent_path(session_id, multi_agent.id)}multi_agent.json"
    self._write_s3_object(multi_agent_key, multi_agent_state)

Session dataclass

Session data model.

Source code in strands/types/session.py

@dataclass
class Session:
    """Session data model."""

    session_id: str
    session_type: SessionType
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    @classmethod
    def from_dict(cls, env: dict[str, Any]) -> "Session":
        """Initialize a Session from a dictionary, ignoring keys that are not class parameters."""
        return cls(**{k: v for k, v in env.items() if k in inspect.signature(cls).parameters})

    def to_dict(self) -> dict[str, Any]:
        """Convert the Session to a dictionary representation."""
        return asdict(self)

from_dict(env) classmethod

Initialize a Session from a dictionary, ignoring keys that are not class parameters.

Source code in strands/types/session.py

@classmethod
def from_dict(cls, env: dict[str, Any]) -> "Session":
    """Initialize a Session from a dictionary, ignoring keys that are not class parameters."""
    return cls(**{k: v for k, v in env.items() if k in inspect.signature(cls).parameters})

to_dict()

Convert the Session to a dictionary representation.

Source code in strands/types/session.py

def to_dict(self) -> dict[str, Any]:
    """Convert the Session to a dictionary representation."""
    return asdict(self)

SessionAgent dataclass

Agent that belongs to a Session.

Attributes:

Name	Type	Description
agent_id	str	Unique id for the agent.
state	dict[str, Any]	User managed state.
conversation_manager_state	dict[str, Any]	State for conversation management.
created_at	str	Created at time.
updated_at	str	Updated at time.

Source code in strands/types/session.py

@dataclass
class SessionAgent:
    """Agent that belongs to a Session.

    Attributes:
        agent_id: Unique id for the agent.
        state: User managed state.
        conversation_manager_state: State for conversation management.
        created_at: Created at time.
        updated_at: Updated at time.
    """

    agent_id: str
    state: dict[str, Any]
    conversation_manager_state: dict[str, Any]
    _internal_state: dict[str, Any] = field(default_factory=dict)  # Strands managed state
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    @classmethod
    def from_agent(cls, agent: "Agent") -> "SessionAgent":
        """Convert an Agent to a SessionAgent."""
        if agent.agent_id is None:
            raise ValueError("agent_id needs to be defined.")
        return cls(
            agent_id=agent.agent_id,
            conversation_manager_state=agent.conversation_manager.get_state(),
            state=agent.state.get(),
            _internal_state={
                "interrupt_state": agent._interrupt_state.to_dict(),
            },
        )

    @classmethod
    def from_bidi_agent(cls, agent: "BidiAgent") -> "SessionAgent":
        """Convert a BidiAgent to a SessionAgent.

        Args:
            agent: BidiAgent to convert

        Returns:
            SessionAgent with empty conversation_manager_state (BidiAgent doesn't use conversation manager)
        """
        if agent.agent_id is None:
            raise ValueError("agent_id needs to be defined.")

        # BidiAgent doesn't have _interrupt_state yet, so we use empty dict for internal state
        internal_state = {}
        if hasattr(agent, "_interrupt_state"):
            internal_state["interrupt_state"] = agent._interrupt_state.to_dict()

        return cls(
            agent_id=agent.agent_id,
            conversation_manager_state={},  # BidiAgent has no conversation_manager
            state=agent.state.get(),
            _internal_state=internal_state,
        )

    @classmethod
    def from_dict(cls, env: dict[str, Any]) -> "SessionAgent":
        """Initialize a SessionAgent from a dictionary, ignoring keys that are not class parameters."""
        return cls(**{k: v for k, v in env.items() if k in inspect.signature(cls).parameters})

    def to_dict(self) -> dict[str, Any]:
        """Convert the SessionAgent to a dictionary representation."""
        return asdict(self)

    def initialize_internal_state(self, agent: "Agent") -> None:
        """Initialize internal state of agent."""
        if "interrupt_state" in self._internal_state:
            agent._interrupt_state = _InterruptState.from_dict(self._internal_state["interrupt_state"])

    def initialize_bidi_internal_state(self, agent: "BidiAgent") -> None:
        """Initialize internal state of BidiAgent.

        Args:
            agent: BidiAgent to initialize internal state for
        """
        # BidiAgent doesn't have _interrupt_state yet, so we skip interrupt state restoration
        # When BidiAgent adds _interrupt_state support, this will automatically work
        if "interrupt_state" in self._internal_state and hasattr(agent, "_interrupt_state"):
            agent._interrupt_state = _InterruptState.from_dict(self._internal_state["interrupt_state"])

from_agent(agent) classmethod

Convert an Agent to a SessionAgent.

Source code in strands/types/session.py

@classmethod
def from_agent(cls, agent: "Agent") -> "SessionAgent":
    """Convert an Agent to a SessionAgent."""
    if agent.agent_id is None:
        raise ValueError("agent_id needs to be defined.")
    return cls(
        agent_id=agent.agent_id,
        conversation_manager_state=agent.conversation_manager.get_state(),
        state=agent.state.get(),
        _internal_state={
            "interrupt_state": agent._interrupt_state.to_dict(),
        },
    )

from_bidi_agent(agent) classmethod

Convert a BidiAgent to a SessionAgent.

Parameters:

Name	Type	Description	Default
agent	BidiAgent	BidiAgent to convert	required

Returns:

Type	Description
SessionAgent	SessionAgent with empty conversation_manager_state (BidiAgent doesn't use conversation manager)

Source code in strands/types/session.py

@classmethod
def from_bidi_agent(cls, agent: "BidiAgent") -> "SessionAgent":
    """Convert a BidiAgent to a SessionAgent.

    Args:
        agent: BidiAgent to convert

    Returns:
        SessionAgent with empty conversation_manager_state (BidiAgent doesn't use conversation manager)
    """
    if agent.agent_id is None:
        raise ValueError("agent_id needs to be defined.")

    # BidiAgent doesn't have _interrupt_state yet, so we use empty dict for internal state
    internal_state = {}
    if hasattr(agent, "_interrupt_state"):
        internal_state["interrupt_state"] = agent._interrupt_state.to_dict()

    return cls(
        agent_id=agent.agent_id,
        conversation_manager_state={},  # BidiAgent has no conversation_manager
        state=agent.state.get(),
        _internal_state=internal_state,
    )

from_dict(env) classmethod

Initialize a SessionAgent from a dictionary, ignoring keys that are not class parameters.

Source code in strands/types/session.py

@classmethod
def from_dict(cls, env: dict[str, Any]) -> "SessionAgent":
    """Initialize a SessionAgent from a dictionary, ignoring keys that are not class parameters."""
    return cls(**{k: v for k, v in env.items() if k in inspect.signature(cls).parameters})

initialize_bidi_internal_state(agent)

Initialize internal state of BidiAgent.

Parameters:

Name	Type	Description	Default
agent	BidiAgent	BidiAgent to initialize internal state for	required

Source code in strands/types/session.py

def initialize_bidi_internal_state(self, agent: "BidiAgent") -> None:
    """Initialize internal state of BidiAgent.

    Args:
        agent: BidiAgent to initialize internal state for
    """
    # BidiAgent doesn't have _interrupt_state yet, so we skip interrupt state restoration
    # When BidiAgent adds _interrupt_state support, this will automatically work
    if "interrupt_state" in self._internal_state and hasattr(agent, "_interrupt_state"):
        agent._interrupt_state = _InterruptState.from_dict(self._internal_state["interrupt_state"])

initialize_internal_state(agent)

Initialize internal state of agent.

Source code in strands/types/session.py

def initialize_internal_state(self, agent: "Agent") -> None:
    """Initialize internal state of agent."""
    if "interrupt_state" in self._internal_state:
        agent._interrupt_state = _InterruptState.from_dict(self._internal_state["interrupt_state"])

to_dict()

Convert the SessionAgent to a dictionary representation.

Source code in strands/types/session.py

def to_dict(self) -> dict[str, Any]:
    """Convert the SessionAgent to a dictionary representation."""
    return asdict(self)

SessionException

Bases: Exception

Exception raised when session operations fail.

Source code in strands/types/exceptions.py

class SessionException(Exception):
    """Exception raised when session operations fail."""

    pass

SessionMessage dataclass

Message within a SessionAgent.

Attributes:

Name	Type	Description
message	Message	Message content
message_id	int	Index of the message in the conversation history
redact_message	Optional[Message]	If the original message is redacted, this is the new content to use
created_at	str	ISO format timestamp for when this message was created
updated_at	str	ISO format timestamp for when this message was last updated

Source code in strands/types/session.py

@dataclass
class SessionMessage:
    """Message within a SessionAgent.

    Attributes:
        message: Message content
        message_id: Index of the message in the conversation history
        redact_message: If the original message is redacted, this is the new content to use
        created_at: ISO format timestamp for when this message was created
        updated_at: ISO format timestamp for when this message was last updated
    """

    message: Message
    message_id: int
    redact_message: Optional[Message] = None
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    @classmethod
    def from_message(cls, message: Message, index: int) -> "SessionMessage":
        """Convert from a Message, base64 encoding bytes values."""
        return cls(
            message=message,
            message_id=index,
            created_at=datetime.now(timezone.utc).isoformat(),
            updated_at=datetime.now(timezone.utc).isoformat(),
        )

    def to_message(self) -> Message:
        """Convert SessionMessage back to a Message, decoding any bytes values.

        If the message was redacted, return the redact content instead.
        """
        if self.redact_message is not None:
            return self.redact_message
        else:
            return self.message

    @classmethod
    def from_dict(cls, env: dict[str, Any]) -> "SessionMessage":
        """Initialize a SessionMessage from a dictionary, ignoring keys that are not class parameters."""
        extracted_relevant_parameters = {k: v for k, v in env.items() if k in inspect.signature(cls).parameters}
        return cls(**decode_bytes_values(extracted_relevant_parameters))

    def to_dict(self) -> dict[str, Any]:
        """Convert the SessionMessage to a dictionary representation."""
        return encode_bytes_values(asdict(self))  # type: ignore

from_dict(env) classmethod

Initialize a SessionMessage from a dictionary, ignoring keys that are not class parameters.

Source code in strands/types/session.py

@classmethod
def from_dict(cls, env: dict[str, Any]) -> "SessionMessage":
    """Initialize a SessionMessage from a dictionary, ignoring keys that are not class parameters."""
    extracted_relevant_parameters = {k: v for k, v in env.items() if k in inspect.signature(cls).parameters}
    return cls(**decode_bytes_values(extracted_relevant_parameters))

from_message(message, index) classmethod

Convert from a Message, base64 encoding bytes values.

Source code in strands/types/session.py

@classmethod
def from_message(cls, message: Message, index: int) -> "SessionMessage":
    """Convert from a Message, base64 encoding bytes values."""
    return cls(
        message=message,
        message_id=index,
        created_at=datetime.now(timezone.utc).isoformat(),
        updated_at=datetime.now(timezone.utc).isoformat(),
    )

to_dict()

Convert the SessionMessage to a dictionary representation.

Source code in strands/types/session.py

def to_dict(self) -> dict[str, Any]:
    """Convert the SessionMessage to a dictionary representation."""
    return encode_bytes_values(asdict(self))  # type: ignore

to_message()

Convert SessionMessage back to a Message, decoding any bytes values.

If the message was redacted, return the redact content instead.

Source code in strands/types/session.py

def to_message(self) -> Message:
    """Convert SessionMessage back to a Message, decoding any bytes values.

    If the message was redacted, return the redact content instead.
    """
    if self.redact_message is not None:
        return self.redact_message
    else:
        return self.message

SessionRepository

Bases: ABC

Abstract repository for creating, reading, and updating Sessions, AgentSessions, and AgentMessages.

Source code in strands/session/session_repository.py

class SessionRepository(ABC):
    """Abstract repository for creating, reading, and updating Sessions, AgentSessions, and AgentMessages."""

    @abstractmethod
    def create_session(self, session: Session, **kwargs: Any) -> Session:
        """Create a new Session."""

    @abstractmethod
    def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
        """Read a Session."""

    @abstractmethod
    def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
        """Create a new Agent in a Session."""

    @abstractmethod
    def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[SessionAgent]:
        """Read an Agent."""

    @abstractmethod
    def update_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
        """Update an Agent."""

    @abstractmethod
    def create_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
        """Create a new Message for the Agent."""

    @abstractmethod
    def read_message(self, session_id: str, agent_id: str, message_id: int, **kwargs: Any) -> Optional[SessionMessage]:
        """Read a Message."""

    @abstractmethod
    def update_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
        """Update a Message.

        A message is usually only updated when some content is redacted due to a guardrail.
        """

    @abstractmethod
    def list_messages(
        self, session_id: str, agent_id: str, limit: Optional[int] = None, offset: int = 0, **kwargs: Any
    ) -> list[SessionMessage]:
        """List Messages from an Agent with pagination."""

    def create_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
        """Create a new MultiAgent state for the Session."""
        raise NotImplementedError("MultiAgent is not implemented for this repository")

    def read_multi_agent(self, session_id: str, multi_agent_id: str, **kwargs: Any) -> Optional[dict[str, Any]]:
        """Read the MultiAgent state for the Session."""
        raise NotImplementedError("MultiAgent is not implemented for this repository")

    def update_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
        """Update the MultiAgent state for the Session."""
        raise NotImplementedError("MultiAgent is not implemented for this repository")

create_agent(session_id, session_agent, **kwargs) abstractmethod

Create a new Agent in a Session.

Source code in strands/session/session_repository.py

@abstractmethod
def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
    """Create a new Agent in a Session."""

create_message(session_id, agent_id, session_message, **kwargs) abstractmethod

Create a new Message for the Agent.

Source code in strands/session/session_repository.py

@abstractmethod
def create_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
    """Create a new Message for the Agent."""

create_multi_agent(session_id, multi_agent, **kwargs)

Create a new MultiAgent state for the Session.

Source code in strands/session/session_repository.py

def create_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
    """Create a new MultiAgent state for the Session."""
    raise NotImplementedError("MultiAgent is not implemented for this repository")

create_session(session, **kwargs) abstractmethod

Create a new Session.

Source code in strands/session/session_repository.py

@abstractmethod
def create_session(self, session: Session, **kwargs: Any) -> Session:
    """Create a new Session."""

list_messages(session_id, agent_id, limit=None, offset=0, **kwargs) abstractmethod

List Messages from an Agent with pagination.

Source code in strands/session/session_repository.py

@abstractmethod
def list_messages(
    self, session_id: str, agent_id: str, limit: Optional[int] = None, offset: int = 0, **kwargs: Any
) -> list[SessionMessage]:
    """List Messages from an Agent with pagination."""

read_agent(session_id, agent_id, **kwargs) abstractmethod

Read an Agent.

Source code in strands/session/session_repository.py

@abstractmethod
def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[SessionAgent]:
    """Read an Agent."""

read_message(session_id, agent_id, message_id, **kwargs) abstractmethod

Read a Message.

Source code in strands/session/session_repository.py

@abstractmethod
def read_message(self, session_id: str, agent_id: str, message_id: int, **kwargs: Any) -> Optional[SessionMessage]:
    """Read a Message."""

read_multi_agent(session_id, multi_agent_id, **kwargs)

Read the MultiAgent state for the Session.

Source code in strands/session/session_repository.py

def read_multi_agent(self, session_id: str, multi_agent_id: str, **kwargs: Any) -> Optional[dict[str, Any]]:
    """Read the MultiAgent state for the Session."""
    raise NotImplementedError("MultiAgent is not implemented for this repository")

read_session(session_id, **kwargs) abstractmethod

Read a Session.

Source code in strands/session/session_repository.py

@abstractmethod
def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
    """Read a Session."""

update_agent(session_id, session_agent, **kwargs) abstractmethod

Update an Agent.

Source code in strands/session/session_repository.py

@abstractmethod
def update_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
    """Update an Agent."""

update_message(session_id, agent_id, session_message, **kwargs) abstractmethod

Update a Message.

A message is usually only updated when some content is redacted due to a guardrail.

Source code in strands/session/session_repository.py

@abstractmethod
def update_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
    """Update a Message.

    A message is usually only updated when some content is redacted due to a guardrail.
    """

update_multi_agent(session_id, multi_agent, **kwargs)

Update the MultiAgent state for the Session.

Source code in strands/session/session_repository.py

def update_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
    """Update the MultiAgent state for the Session."""
    raise NotImplementedError("MultiAgent is not implemented for this repository")