refactor: Update interfaces to include kwargs to enable backwards compatibility

Author: Murat Kaan Meral

src/strands/agent/agent.py

@@ -134,6 +134,7 @@ def caller(
                 }
 
                 async def acall() -> ToolResult:
+                    # Pass kwargs as invocation_state
                     async for event in run_tool(self._agent, tool_use, kwargs):
                         _ = event
 
@@ -483,7 +484,7 @@ async def stream_async(self, prompt: Union[str, list[ContentBlock]], **kwargs: A
         self._start_agent_trace_span(message)
 
         try:
-            events = self._run_loop(message, kwargs)
+            events = self._run_loop(message, invocation_state=kwargs)
             async for event in events:
                 if "callback" in event:
                     callback_handler(**event["callback"])
@@ -499,33 +500,35 @@ async def stream_async(self, prompt: Union[str, list[ContentBlock]], **kwargs: A
             self._end_agent_trace_span(error=e)
             raise
 
-    async def _run_loop(self, message: Message, kwargs: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]:
+    async def _run_loop(
+        self, message: Message, invocation_state: dict[str, Any]
+    ) -> AsyncGenerator[dict[str, Any], None]:
         """Execute the agent's event loop with the given message and parameters.
 
         Args:
             message: The user message to add to the conversation.
-            kwargs: Additional parameters to pass to the event loop.
+            invocation_state: Additional parameters to pass to the event loop.
 
         Yields:
             Events from the event loop cycle.
         """
         self._hooks.invoke_callbacks(BeforeInvocationEvent(agent=self))
 
         try:
-            yield {"callback": {"init_event_loop": True, **kwargs}}
+            yield {"callback": {"init_event_loop": True, **invocation_state}}
 
             self._append_message(message)
 
             # Execute the event loop cycle with retry logic for context limits
-            events = self._execute_event_loop_cycle(kwargs)
+            events = self._execute_event_loop_cycle(invocation_state)
             async for event in events:
                 yield event
 
         finally:
             self.conversation_manager.apply_management(self)
             self._hooks.invoke_callbacks(AfterInvocationEvent(agent=self))
 
-    async def _execute_event_loop_cycle(self, kwargs: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]:
+    async def _execute_event_loop_cycle(self, invocation_state: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]:
         """Execute the event loop cycle with retry logic for context window limits.
 
         This internal method handles the execution of the event loop cycle and implements
@@ -535,22 +538,22 @@ async def _execute_event_loop_cycle(self, kwargs: dict[str, Any]) -> AsyncGenera
         Yields:
             Events of the loop cycle.
         """
-        # Add `Agent` to kwargs to keep backwards-compatibility
-        kwargs["agent"] = self
+        # Add `Agent` to invocation_state to keep backwards-compatibility
+        invocation_state["agent"] = self
 
         try:
             # Execute the main event loop cycle
             events = event_loop_cycle(
                 agent=self,
-                kwargs=kwargs,
+                invocation_state=invocation_state,
             )
             async for event in events:
                 yield event
 
         except ContextWindowOverflowException as e:
             # Try reducing the context size and retrying
             self.conversation_manager.reduce_context(self, e=e)
-            events = self._execute_event_loop_cycle(kwargs)
+            events = self._execute_event_loop_cycle(invocation_state)
             async for event in events:
                 yield event
 

src/strands/agent/conversation_manager/conversation_manager.py

@@ -1,7 +1,7 @@
 """Abstract interface for conversation history management."""
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 if TYPE_CHECKING:
     from ...agent.agent import Agent
@@ -20,7 +20,7 @@ class ConversationManager(ABC):
 
     @abstractmethod
     # pragma: no cover
-    def apply_management(self, agent: "Agent") -> None:
+    def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """Applies management strategy to the provided agent.
 
         Processes the conversation history to maintain appropriate size by modifying the messages list in-place.
@@ -30,12 +30,13 @@ def apply_management(self, agent: "Agent") -> None:
         Args:
             agent: The agent whose conversation history will be manage.
                 This list is modified in-place.
+            **kwargs: Additional keyword arguments for future extensibility.
         """
         pass
 
     @abstractmethod
     # pragma: no cover
-    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None) -> None:
+    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs: Any) -> None:
         """Called when the model's context window is exceeded.
 
         This method should implement the specific strategy for reducing the window size when a context overflow occurs.
@@ -52,5 +53,6 @@ def reduce_context(self, agent: "Agent", e: Optional[Exception] = None) -> None:
             agent: The agent whose conversation history will be reduced.
                 This list is modified in-place.
             e: The exception that triggered the context reduction, if any.
+            **kwargs: Additional keyword arguments for future extensibility.
         """
         pass

src/strands/agent/conversation_manager/null_conversation_manager.py

@@ -1,6 +1,6 @@
 """Null implementation of conversation management."""
 
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 if TYPE_CHECKING:
     from ...agent.agent import Agent
@@ -19,20 +19,22 @@ class NullConversationManager(ConversationManager):
     - Situations where the full conversation history should be preserved
     """
 
-    def apply_management(self, _agent: "Agent") -> None:
+    def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """Does nothing to the conversation history.
 
         Args:
-            _agent: The agent whose conversation history will remain unmodified.
+            agent: The agent whose conversation history will remain unmodified.
+            **kwargs: Additional keyword arguments for future extensibility.
         """
         pass
 
-    def reduce_context(self, _agent: "Agent", e: Optional[Exception] = None) -> None:
+    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs: Any) -> None:
         """Does not reduce context and raises an exception.
 
         Args:
-            _agent: The agent whose conversation history will remain unmodified.
+            agent: The agent whose conversation history will remain unmodified.
             e: The exception that triggered the context reduction, if any.
+            **kwargs: Additional keyword arguments for future extensibility.
 
         Raises:
             e: If provided.

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -1,7 +1,7 @@
 """Sliding window conversation history management."""
 
 import logging
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 if TYPE_CHECKING:
     from ...agent.agent import Agent
@@ -55,7 +55,7 @@ def __init__(self, window_size: int = 40, should_truncate_results: bool = True):
         self.window_size = window_size
         self.should_truncate_results = should_truncate_results
 
-    def apply_management(self, agent: "Agent") -> None:
+    def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """Apply the sliding window to the agent's messages array to maintain a manageable history size.
 
         This method is called after every event loop cycle, as the messages array may have been modified with tool
@@ -68,6 +68,7 @@ def apply_management(self, agent: "Agent") -> None:
 
         Args:
             agent: The agent whose messages will be managed.
+            **kwargs: Additional keyword arguments for future extensibility.
                 This list is modified in-place.
         """
         messages = agent.messages
@@ -111,7 +112,7 @@ def _remove_dangling_messages(self, messages: Messages) -> None:
                     if not any("toolResult" in content for content in messages[-1]["content"]):
                         messages.pop()
 
-    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None) -> None:
+    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs: Any) -> None:
         """Trim the oldest messages to reduce the conversation context size.
 
         The method handles special cases where trimming the messages leads to:
@@ -122,6 +123,7 @@ def reduce_context(self, agent: "Agent", e: Optional[Exception] = None) -> None:
             agent: The agent whose messages will be reduce.
                 This list is modified in-place.
             e: The exception that triggered the context reduction, if any.
+            **kwargs: Additional keyword arguments for future extensibility.
 
         Raises:
             ContextWindowOverflowException: If the context cannot be reduced further.

src/strands/agent/conversation_manager/summarizing_conversation_manager.py

@@ -1,7 +1,7 @@
 """Summarizing conversation history management with configurable options."""
 
 import logging
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Any, List, Optional
 
 from ...types.content import Message
 from ...types.exceptions import ContextWindowOverflowException
@@ -78,7 +78,7 @@ def __init__(
         self.summarization_agent = summarization_agent
         self.summarization_system_prompt = summarization_system_prompt
 
-    def apply_management(self, agent: "Agent") -> None:
+    def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """Apply management strategy to conversation history.
 
         For the summarizing conversation manager, no proactive management is performed.
@@ -87,17 +87,19 @@ def apply_management(self, agent: "Agent") -> None:
         Args:
             agent: The agent whose conversation history will be managed.
                 The agent's messages list is modified in-place.
+            **kwargs: Additional keyword arguments for future extensibility.
         """
         # No proactive management - summarization only happens on context overflow
         pass
 
-    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None) -> None:
+    def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs: Any) -> None:
         """Reduce context using summarization.
 
         Args:
             agent: The agent whose conversation history will be reduced.
                 The agent's messages list is modified in-place.
             e: The exception that triggered the context reduction, if any.
+            **kwargs: Additional keyword arguments for future extensibility.
 
         Raises:
             ContextWindowOverflowException: If the context cannot be summarized.