Skip to content

strands.session.file_session_manager

File-based session manager for local filesystem 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

FileSessionManager

Bases: RepositorySessionManager, SessionRepository

File-based session manager for local filesystem 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/file_session_manager.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
class FileSessionManager(RepositorySessionManager, SessionRepository):
    """File-based session manager for local filesystem 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,
        storage_dir: Optional[str] = None,
        **kwargs: Any,
    ):
        """Initialize FileSession with filesystem storage.

        Args:
            session_id: ID for the session.
                ID is not allowed to contain path separators (e.g., a/b).
            storage_dir: Directory for local filesystem storage (defaults to temp dir).
            **kwargs: Additional keyword arguments for future extensibility.
        """
        self.storage_dir = storage_dir or os.path.join(tempfile.gettempdir(), "strands/sessions")
        os.makedirs(self.storage_dir, exist_ok=True)

        super().__init__(session_id=session_id, session_repository=self)

    def _get_session_path(self, session_id: str) -> str:
        """Get session directory path.

        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 os.path.join(self.storage_dir, f"{SESSION_PREFIX}{session_id}")

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

        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 os.path.join(session_path, "agents", f"{AGENT_PREFIX}{agent_id}")

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

        Args:
            session_id: ID of the session
            agent_id: ID of the agent
            message_id: Index of the message
        Returns:
            The filename 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 os.path.join(agent_path, "messages", f"{MESSAGE_PREFIX}{message_id}.json")

    def _read_file(self, path: str) -> dict[str, Any]:
        """Read JSON file."""
        try:
            with open(path, "r", encoding="utf-8") as f:
                return cast(dict[str, Any], json.load(f))
        except json.JSONDecodeError as e:
            raise SessionException(f"Invalid JSON in file {path}: {str(e)}") from e

    def _write_file(self, path: str, data: dict[str, Any]) -> None:
        """Write JSON file."""
        os.makedirs(os.path.dirname(path), exist_ok=True)
        # This automic write ensure the completeness of session files in both single agent/ multi agents
        tmp = f"{path}.tmp"
        with open(tmp, "w", encoding="utf-8", newline="\n") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        os.replace(tmp, path)

    def create_session(self, session: Session, **kwargs: Any) -> Session:
        """Create a new session."""
        session_dir = self._get_session_path(session.session_id)
        if os.path.exists(session_dir):
            raise SessionException(f"Session {session.session_id} already exists")

        # Create directory structure
        os.makedirs(session_dir, exist_ok=True)
        os.makedirs(os.path.join(session_dir, "agents"), exist_ok=True)
        os.makedirs(os.path.join(session_dir, "multi_agents"), exist_ok=True)

        # Write session file
        session_file = os.path.join(session_dir, "session.json")
        session_dict = session.to_dict()
        self._write_file(session_file, session_dict)

        return session

    def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
        """Read session data."""
        session_file = os.path.join(self._get_session_path(session_id), "session.json")
        if not os.path.exists(session_file):
            return None

        session_data = self._read_file(session_file)
        return Session.from_dict(session_data)

    def delete_session(self, session_id: str, **kwargs: Any) -> None:
        """Delete session and all associated data."""
        session_dir = self._get_session_path(session_id)
        if not os.path.exists(session_dir):
            raise SessionException(f"Session {session_id} does not exist")

        shutil.rmtree(session_dir)

    def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
        """Create a new agent in the session."""
        agent_id = session_agent.agent_id

        agent_dir = self._get_agent_path(session_id, agent_id)
        os.makedirs(agent_dir, exist_ok=True)
        os.makedirs(os.path.join(agent_dir, "messages"), exist_ok=True)

        agent_file = os.path.join(agent_dir, "agent.json")
        session_data = session_agent.to_dict()
        self._write_file(agent_file, session_data)

    def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[SessionAgent]:
        """Read agent data."""
        agent_file = os.path.join(self._get_agent_path(session_id, agent_id), "agent.json")
        if not os.path.exists(agent_file):
            return None

        agent_data = self._read_file(agent_file)
        return SessionAgent.from_dict(agent_data)

    def update_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
        """Update agent data."""
        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")

        session_agent.created_at = previous_agent.created_at
        agent_file = os.path.join(self._get_agent_path(session_id, agent_id), "agent.json")
        self._write_file(agent_file, session_agent.to_dict())

    def create_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
        """Create a new message for the agent."""
        message_file = self._get_message_path(
            session_id,
            agent_id,
            session_message.message_id,
        )
        session_dict = session_message.to_dict()
        self._write_file(message_file, session_dict)

    def read_message(self, session_id: str, agent_id: str, message_id: int, **kwargs: Any) -> Optional[SessionMessage]:
        """Read message data."""
        message_path = self._get_message_path(session_id, agent_id, message_id)
        if not os.path.exists(message_path):
            return None
        message_data = self._read_file(message_path)
        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."""
        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 the original created_at timestamp
        session_message.created_at = previous_message.created_at
        message_file = self._get_message_path(session_id, agent_id, message_id)
        self._write_file(message_file, 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."""
        messages_dir = os.path.join(self._get_agent_path(session_id, agent_id), "messages")
        if not os.path.exists(messages_dir):
            raise SessionException(f"Messages directory missing from agent: {agent_id} in session {session_id}")

        # Read all message files, and record the index
        message_index_files: list[tuple[int, str]] = []
        for filename in os.listdir(messages_dir):
            if filename.startswith(MESSAGE_PREFIX) and filename.endswith(".json"):
                # Extract index from message_<index>.json format
                index = int(filename[len(MESSAGE_PREFIX) : -5])  # Remove prefix and .json suffix
                message_index_files.append((index, filename))

        # Sort by index and extract just the filenames
        message_files = [f for _, f in sorted(message_index_files)]

        # Apply pagination to filenames
        if limit is not None:
            message_files = message_files[offset : offset + limit]
        else:
            message_files = message_files[offset:]

        # Load only the message files
        messages: list[SessionMessage] = []
        for filename in message_files:
            file_path = os.path.join(messages_dir, filename)
            message_data = self._read_file(file_path)
            messages.append(SessionMessage.from_dict(message_data))

        return messages

    def _get_multi_agent_path(self, session_id: str, multi_agent_id: str) -> str:
        """Get multi-agent state file path."""
        session_path = self._get_session_path(session_id)
        multi_agent_id = _identifier.validate(multi_agent_id, _identifier.Identifier.AGENT)
        return os.path.join(session_path, "multi_agents", f"{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 the session."""
        multi_agent_id = multi_agent.id
        multi_agent_dir = self._get_multi_agent_path(session_id, multi_agent_id)
        os.makedirs(multi_agent_dir, exist_ok=True)

        multi_agent_file = os.path.join(multi_agent_dir, "multi_agent.json")
        session_data = multi_agent.serialize_state()
        self._write_file(multi_agent_file, 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 filesystem."""
        multi_agent_file = os.path.join(self._get_multi_agent_path(session_id, multi_agent_id), "multi_agent.json")
        if not os.path.exists(multi_agent_file):
            return None
        return self._read_file(multi_agent_file)

    def update_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
        """Update multi-agent state from filesystem."""
        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_file = os.path.join(self._get_multi_agent_path(session_id, multi_agent.id), "multi_agent.json")
        self._write_file(multi_agent_file, multi_agent_state)

__init__(session_id, storage_dir=None, **kwargs)

Initialize FileSession with filesystem 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
storage_dir Optional[str]

Directory for local filesystem storage (defaults to temp dir).

 None
**kwargs Any

Additional keyword arguments for future extensibility.

 {}
Source code in strands/session/file_session_manager.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    session_id: str,
    storage_dir: Optional[str] = None,
    **kwargs: Any,
):
    """Initialize FileSession with filesystem storage.

    Args:
        session_id: ID for the session.
            ID is not allowed to contain path separators (e.g., a/b).
        storage_dir: Directory for local filesystem storage (defaults to temp dir).
        **kwargs: Additional keyword arguments for future extensibility.
    """
    self.storage_dir = storage_dir or os.path.join(tempfile.gettempdir(), "strands/sessions")
    os.makedirs(self.storage_dir, exist_ok=True)

    super().__init__(session_id=session_id, session_repository=self)

create_agent(session_id, session_agent, **kwargs)

Create a new agent in the session.

Source code in strands/session/file_session_manager.py 
160
161
162
163
164
165
166
167
168
169
170
def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
    """Create a new agent in the session."""
    agent_id = session_agent.agent_id

    agent_dir = self._get_agent_path(session_id, agent_id)
    os.makedirs(agent_dir, exist_ok=True)
    os.makedirs(os.path.join(agent_dir, "messages"), exist_ok=True)

    agent_file = os.path.join(agent_dir, "agent.json")
    session_data = session_agent.to_dict()
    self._write_file(agent_file, session_data)

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

Create a new message for the agent.

Source code in strands/session/file_session_manager.py 
192
193
194
195
196
197
198
199
200
def create_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
    """Create a new message for the agent."""
    message_file = self._get_message_path(
        session_id,
        agent_id,
        session_message.message_id,
    )
    session_dict = session_message.to_dict()
    self._write_file(message_file, session_dict)

create_multi_agent(session_id, multi_agent, **kwargs)

Create a new multiagent state in the session.

Source code in strands/session/file_session_manager.py 
262
263
264
265
266
267
268
269
270
def create_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
    """Create a new multiagent state in the session."""
    multi_agent_id = multi_agent.id
    multi_agent_dir = self._get_multi_agent_path(session_id, multi_agent_id)
    os.makedirs(multi_agent_dir, exist_ok=True)

    multi_agent_file = os.path.join(multi_agent_dir, "multi_agent.json")
    session_data = multi_agent.serialize_state()
    self._write_file(multi_agent_file, session_data)

create_session(session, **kwargs)

Create a new session.

Source code in strands/session/file_session_manager.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def create_session(self, session: Session, **kwargs: Any) -> Session:
    """Create a new session."""
    session_dir = self._get_session_path(session.session_id)
    if os.path.exists(session_dir):
        raise SessionException(f"Session {session.session_id} already exists")

    # Create directory structure
    os.makedirs(session_dir, exist_ok=True)
    os.makedirs(os.path.join(session_dir, "agents"), exist_ok=True)
    os.makedirs(os.path.join(session_dir, "multi_agents"), exist_ok=True)

    # Write session file
    session_file = os.path.join(session_dir, "session.json")
    session_dict = session.to_dict()
    self._write_file(session_file, session_dict)

    return session

delete_session(session_id, **kwargs)

Delete session and all associated data.

Source code in strands/session/file_session_manager.py 
152
153
154
155
156
157
158
def delete_session(self, session_id: str, **kwargs: Any) -> None:
    """Delete session and all associated data."""
    session_dir = self._get_session_path(session_id)
    if not os.path.exists(session_dir):
        raise SessionException(f"Session {session_id} does not exist")

    shutil.rmtree(session_dir)

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

List messages for an agent with pagination.

Source code in strands/session/file_session_manager.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
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."""
    messages_dir = os.path.join(self._get_agent_path(session_id, agent_id), "messages")
    if not os.path.exists(messages_dir):
        raise SessionException(f"Messages directory missing from agent: {agent_id} in session {session_id}")

    # Read all message files, and record the index
    message_index_files: list[tuple[int, str]] = []
    for filename in os.listdir(messages_dir):
        if filename.startswith(MESSAGE_PREFIX) and filename.endswith(".json"):
            # Extract index from message_<index>.json format
            index = int(filename[len(MESSAGE_PREFIX) : -5])  # Remove prefix and .json suffix
            message_index_files.append((index, filename))

    # Sort by index and extract just the filenames
    message_files = [f for _, f in sorted(message_index_files)]

    # Apply pagination to filenames
    if limit is not None:
        message_files = message_files[offset : offset + limit]
    else:
        message_files = message_files[offset:]

    # Load only the message files
    messages: list[SessionMessage] = []
    for filename in message_files:
        file_path = os.path.join(messages_dir, filename)
        message_data = self._read_file(file_path)
        messages.append(SessionMessage.from_dict(message_data))

    return messages

read_agent(session_id, agent_id, **kwargs)

Read agent data.

Source code in strands/session/file_session_manager.py 
172
173
174
175
176
177
178
179
def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[SessionAgent]:
    """Read agent data."""
    agent_file = os.path.join(self._get_agent_path(session_id, agent_id), "agent.json")
    if not os.path.exists(agent_file):
        return None

    agent_data = self._read_file(agent_file)
    return SessionAgent.from_dict(agent_data)

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

Read message data.

Source code in strands/session/file_session_manager.py 
202
203
204
205
206
207
208
def read_message(self, session_id: str, agent_id: str, message_id: int, **kwargs: Any) -> Optional[SessionMessage]:
    """Read message data."""
    message_path = self._get_message_path(session_id, agent_id, message_id)
    if not os.path.exists(message_path):
        return None
    message_data = self._read_file(message_path)
    return SessionMessage.from_dict(message_data)

read_multi_agent(session_id, multi_agent_id, **kwargs)

Read multi-agent state from filesystem.

Source code in strands/session/file_session_manager.py 
272
273
274
275
276
277
def read_multi_agent(self, session_id: str, multi_agent_id: str, **kwargs: Any) -> Optional[dict[str, Any]]:
    """Read multi-agent state from filesystem."""
    multi_agent_file = os.path.join(self._get_multi_agent_path(session_id, multi_agent_id), "multi_agent.json")
    if not os.path.exists(multi_agent_file):
        return None
    return self._read_file(multi_agent_file)

read_session(session_id, **kwargs)

Read session data.

Source code in strands/session/file_session_manager.py 
143
144
145
146
147
148
149
150
def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
    """Read session data."""
    session_file = os.path.join(self._get_session_path(session_id), "session.json")
    if not os.path.exists(session_file):
        return None

    session_data = self._read_file(session_file)
    return Session.from_dict(session_data)

update_agent(session_id, session_agent, **kwargs)

Update agent data.

Source code in strands/session/file_session_manager.py 
181
182
183
184
185
186
187
188
189
190
def update_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: Any) -> None:
    """Update agent data."""
    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")

    session_agent.created_at = previous_agent.created_at
    agent_file = os.path.join(self._get_agent_path(session_id, agent_id), "agent.json")
    self._write_file(agent_file, session_agent.to_dict())

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

Update message data.

Source code in strands/session/file_session_manager.py 
210
211
212
213
214
215
216
217
218
219
220
def update_message(self, session_id: str, agent_id: str, session_message: SessionMessage, **kwargs: Any) -> None:
    """Update message data."""
    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 the original created_at timestamp
    session_message.created_at = previous_message.created_at
    message_file = self._get_message_path(session_id, agent_id, message_id)
    self._write_file(message_file, session_message.to_dict())

update_multi_agent(session_id, multi_agent, **kwargs)

Update multi-agent state from filesystem.

Source code in strands/session/file_session_manager.py 
279
280
281
282
283
284
285
286
287
def update_multi_agent(self, session_id: str, multi_agent: "MultiAgentBase", **kwargs: Any) -> None:
    """Update multi-agent state from filesystem."""
    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_file = os.path.join(self._get_multi_agent_path(session_id, multi_agent.id), "multi_agent.json")
    self._write_file(multi_agent_file, multi_agent_state)

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 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
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 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
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 
252
253
254
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 
189
190
191
192
193
194
195
196
197
198
199
200
201
@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 
248
249
250
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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
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 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
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 
240
241
242
243
244
245
246
247
248
249
250
251
252
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 
81
82
83
84
85
86
87
88
89
90
91
92
93
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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
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 
326
327
328
329
330
331
332
333
334
335
336
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 
231
232
233
234
235
236
237
238
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)

Session dataclass

Session data model.

Source code in strands/types/session.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@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 
200
201
202
203
@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 
205
206
207
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@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 
126
127
128
129
130
131
132
133
134
135
136
137
138
@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 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@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 
165
166
167
168
@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 
179
180
181
182
183
184
185
186
187
188
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 
174
175
176
177
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 
170
171
172
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 
74
75
76
77
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 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@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 
 96
 97
 98
 99
100
@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 
76
77
78
79
80
81
82
83
84
@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 
102
103
104
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 
86
87
88
89
90
91
92
93
94
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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
23
24
25
@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 
35
36
37
@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 
56
57
58
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 
15
16
17
@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 
50
51
52
53
54
@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 
27
28
29
@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 
39
40
41
@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 
60
61
62
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 
19
20
21
@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 
31
32
33
@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 
43
44
45
46
47
48
@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 
64
65
66
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")