Skip to content

strands.session.session_repository

Session repository interface for agent session management.

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}

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)

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")