feat: replace kwargs with invocation_state in agent APIs (#966)

* feat: replace kwargs with invocation_state in agent APIs

* fix: handle **kwargs in stream_async.

* feat: add a unit test for the change

* Update src/strands/agent/agent.py

Co-authored-by: Nick Clegg <[email protected]>

* tool - executors - concurrent - remove no-op gather (#954)

* feat(telemetry): updated traces to match OTEL v1.37 semantic conventions (#952)

* event loop - handle model execution (#958)

* feat: implement concurrent message reading for session managers (#897)

Replace sequential message loading with async concurrent reading in both
S3SessionManager and FileSessionManager to improve performance for long
conversations. Uses asyncio.gather() with run_in_executor() to read
multiple messages simultaneously while maintaining proper ordering.

Resolves: #874

Co-authored-by: Vamil Gandhi <[email protected]>

* hooks - before tool call event - cancel tool (#964)

* fix(telemetry): removed double serialization for events (#977)

* fix(litellm): map LiteLLM context-window errors to ContextWindowOverflowException (#994)

* feat: add more tests and adjust invocation_state dic structure

* Apply suggestion from @Unshure

Co-authored-by: Nick Clegg <[email protected]>

* fix: adjust **kwargs in multiagent primitives

---------

Co-authored-by: Nick Clegg <[email protected]>
Co-authored-by: Patrick Gray <[email protected]>
Co-authored-by: poshinchen <[email protected]>
Co-authored-by: Vamil Gandhi <[email protected]>
Co-authored-by: Vamil Gandhi <[email protected]>
Co-authored-by: ratish <[email protected]>

Author: 7 people

src/strands/agent/agent.py

@@ -13,6 +13,7 @@
 import json
 import logging
 import random
+import warnings
 from concurrent.futures import ThreadPoolExecutor
 from typing import (
     Any,
@@ -374,7 +375,9 @@ def tool_names(self) -> list[str]:
         all_tools = self.tool_registry.get_all_tools_config()
         return list(all_tools.keys())
 
-    def __call__(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
+    def __call__(
+        self, prompt: AgentInput = None, *, invocation_state: dict[str, Any] | None = None, **kwargs: Any
+    ) -> AgentResult:
         """Process a natural language prompt through the agent's event loop.
 
         This method implements the conversational interface with multiple input patterns:
@@ -389,7 +392,8 @@ def __call__(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
-            **kwargs: Additional parameters to pass through the event loop.
+            invocation_state: Additional parameters to pass through the event loop.
+            **kwargs: Additional parameters to pass through the event loop.[Deprecating]
 
         Returns:
             Result object containing:
@@ -401,13 +405,15 @@ def __call__(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
         """
 
         def execute() -> AgentResult:
-            return asyncio.run(self.invoke_async(prompt, **kwargs))
+            return asyncio.run(self.invoke_async(prompt, invocation_state=invocation_state, **kwargs))
 
         with ThreadPoolExecutor() as executor:
             future = executor.submit(execute)
             return future.result()
 
-    async def invoke_async(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
+    async def invoke_async(
+        self, prompt: AgentInput = None, *, invocation_state: dict[str, Any] | None = None, **kwargs: Any
+    ) -> AgentResult:
         """Process a natural language prompt through the agent's event loop.
 
         This method implements the conversational interface with multiple input patterns:
@@ -422,7 +428,8 @@ async def invoke_async(self, prompt: AgentInput = None, **kwargs: Any) -> AgentR
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
-            **kwargs: Additional parameters to pass through the event loop.
+            invocation_state: Additional parameters to pass through the event loop.
+            **kwargs: Additional parameters to pass through the event loop.[Deprecating]
 
         Returns:
             Result: object containing:
@@ -432,7 +439,7 @@ async def invoke_async(self, prompt: AgentInput = None, **kwargs: Any) -> AgentR
                 - metrics: Performance metrics from the event loop
                 - state: The final state of the event loop
         """
-        events = self.stream_async(prompt, **kwargs)
+        events = self.stream_async(prompt, invocation_state=invocation_state, **kwargs)
         async for event in events:
             _ = event
 
@@ -528,9 +535,7 @@ async def structured_output_async(self, output_model: Type[T], prompt: AgentInpu
                 self.hooks.invoke_callbacks(AfterInvocationEvent(agent=self))
 
     async def stream_async(
-        self,
-        prompt: AgentInput = None,
-        **kwargs: Any,
+        self, prompt: AgentInput = None, *, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> AsyncIterator[Any]:
         """Process a natural language prompt and yield events as an async iterator.
 
@@ -546,7 +551,8 @@ async def stream_async(
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
-            **kwargs: Additional parameters to pass to the event loop.
+            invocation_state: Additional parameters to pass through the event loop.
+            **kwargs: Additional parameters to pass to the event loop.[Deprecating]
 
         Yields:
             An async iterator that yields events. Each event is a dictionary containing
@@ -567,7 +573,19 @@ async def stream_async(
                     yield event["data"]
             ```
         """
-        callback_handler = kwargs.get("callback_handler", self.callback_handler)
+        merged_state = {}
+        if kwargs:
+            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
+            merged_state.update(kwargs)
+            if invocation_state is not None:
+                merged_state["invocation_state"] = invocation_state
+        else:
+            if invocation_state is not None:
+                merged_state = invocation_state
+
+        callback_handler = self.callback_handler
+        if kwargs:
+            callback_handler = kwargs.get("callback_handler", self.callback_handler)
 
         # Process input and get message to add (if any)
         messages = self._convert_prompt_to_messages(prompt)
@@ -576,10 +594,10 @@ async def stream_async(
 
         with trace_api.use_span(self.trace_span):
             try:
-                events = self._run_loop(messages, invocation_state=kwargs)
+                events = self._run_loop(messages, invocation_state=merged_state)
 
                 async for event in events:
-                    event.prepare(invocation_state=kwargs)
+                    event.prepare(invocation_state=merged_state)
 
                     if event.is_callback_event:
                         as_dict = event.as_dict()

src/strands/multiagent/base.py

@@ -4,6 +4,7 @@
 """
 
 import asyncio
+import warnings
 from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
@@ -111,8 +112,12 @@ def __call__(
         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)
+
         def execute() -> MultiAgentResult:
-            return asyncio.run(self.invoke_async(task, invocation_state, **kwargs))
+            return asyncio.run(self.invoke_async(task, invocation_state))
 
         with ThreadPoolExecutor() as executor:
             future = executor.submit(execute)

src/strands/multiagent/graph.py

@@ -572,11 +572,11 @@ async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any])
                 elif isinstance(node.executor, Agent):
                     if self.node_timeout is not None:
                         agent_response = await asyncio.wait_for(
-                            node.executor.invoke_async(node_input, **invocation_state),
+                            node.executor.invoke_async(node_input, invocation_state=invocation_state),
                             timeout=self.node_timeout,
                         )
                     else:
-                        agent_response = await node.executor.invoke_async(node_input, **invocation_state)
+                        agent_response = await node.executor.invoke_async(node_input, invocation_state=invocation_state)
 
                     # Extract metrics from agent response
                     usage = Usage(inputTokens=0, outputTokens=0, totalTokens=0)

src/strands/multiagent/swarm.py

@@ -635,8 +635,7 @@ async def _execute_node(
             # Execute node
             result = None
             node.reset_executor_state()
-            # Unpacking since this is the agent class. Other executors should not unpack
-            result = await node.executor.invoke_async(node_input, **invocation_state)
+            result = await node.executor.invoke_async(node_input, invocation_state=invocation_state)
 
             execution_time = round((time.time() - start_time) * 1000)
 

tests/strands/agent/test_agent.py

@@ -4,6 +4,7 @@
 import os
 import textwrap
 import unittest.mock
+import warnings
 from uuid import uuid4
 
 import pytest
@@ -1877,3 +1878,58 @@ def test_tool(action: str) -> str:
     assert '"action": "test_value"' in tool_call_text
     assert '"agent"' not in tool_call_text
     assert '"extra_param"' not in tool_call_text
+
+
+def test_agent__call__handles_none_invocation_state(mock_model, agent):
+    """Test that agent handles None invocation_state without AttributeError."""
+    mock_model.mock_stream.return_value = [
+        {"contentBlockDelta": {"delta": {"text": "test response"}}},
+        {"contentBlockStop": {}},
+    ]
+
+    # This should not raise AttributeError: 'NoneType' object has no attribute 'get'
+    result = agent("test", invocation_state=None)
+
+    assert result.message["content"][0]["text"] == "test response"
+    assert result.stop_reason == "end_turn"
+
+
+def test_agent__call__invocation_state_with_kwargs_deprecation_warning(agent, mock_event_loop_cycle):
+    """Test that kwargs trigger deprecation warning and are merged correctly with invocation_state."""
+
+    async def check_invocation_state(**kwargs):
+        invocation_state = kwargs["invocation_state"]
+        # Should have nested structure when both invocation_state and kwargs are provided
+        assert invocation_state["invocation_state"] == {"my": "state"}
+        assert invocation_state["other_kwarg"] == "foobar"
+        yield EventLoopStopEvent("stop", {"role": "assistant", "content": [{"text": "Response"}]}, {}, {})
+
+    mock_event_loop_cycle.side_effect = check_invocation_state
+
+    with warnings.catch_warnings(record=True) as captured_warnings:
+        warnings.simplefilter("always")
+        agent("hello!", invocation_state={"my": "state"}, other_kwarg="foobar")
+
+    # Verify deprecation warning was issued
+    assert len(captured_warnings) == 1
+    assert issubclass(captured_warnings[0].category, UserWarning)
+    assert "`**kwargs` parameter is deprecating, use `invocation_state` instead." in str(captured_warnings[0].message)
+
+
+def test_agent__call__invocation_state_only_no_warning(agent, mock_event_loop_cycle):
+    """Test that using only invocation_state does not trigger warning and passes state directly."""
+
+    async def check_invocation_state(**kwargs):
+        invocation_state = kwargs["invocation_state"]
+
+        assert invocation_state["my"] == "state"
+        assert "agent" in invocation_state
+        yield EventLoopStopEvent("stop", {"role": "assistant", "content": [{"text": "Response"}]}, {}, {})
+
+    mock_event_loop_cycle.side_effect = check_invocation_state
+
+    with warnings.catch_warnings(record=True) as captured_warnings:
+        warnings.simplefilter("always")
+        agent("hello!", invocation_state={"my": "state"})
+
+    assert len(captured_warnings) == 0

tests/strands/multiagent/test_base.py

@@ -159,17 +159,18 @@ async def invoke_async(self, task, invocation_state, **kwargs):
             self.invoke_async_called = True
             self.received_task = task
             self.received_kwargs = kwargs
+            self.received_invocation_state = invocation_state
             return MultiAgentResult(
                 status=Status.COMPLETED, results={"test": NodeResult(result=Exception("test"), status=Status.COMPLETED)}
             )
 
     agent = TestMultiAgent()
 
     # Test with string task
-    result = agent("test task", param1="value1", param2="value2")
+    result = agent("test task", param1="value1", param2="value2", invocation_state={"value3": "value4"})
 
     assert agent.invoke_async_called
     assert agent.received_task == "test task"
-    assert agent.received_kwargs == {"param1": "value1", "param2": "value2"}
+    assert agent.received_invocation_state == {"param1": "value1", "param2": "value2", "value3": "value4"}
     assert isinstance(result, MultiAgentResult)
     assert result.status == Status.COMPLETED

tests/strands/multiagent/test_graph.py

@@ -310,7 +310,7 @@ async def test_graph_edge_cases(mock_strands_tracer, mock_use_span):
     result = await graph.invoke_async([{"text": "Original task"}])
 
     # Verify entry node was called with original task
-    entry_agent.invoke_async.assert_called_once_with([{"text": "Original task"}])
+    entry_agent.invoke_async.assert_called_once_with([{"text": "Original task"}], invocation_state={})
     assert result.status == Status.COMPLETED
     mock_strands_tracer.start_multiagent_span.assert_called()
     mock_use_span.assert_called_once()
@@ -906,7 +906,7 @@ def __init__(self, name):
             self._session_manager = None
             self.hooks = HookRegistry()
 
-        async def invoke_async(self, input_data):
+        async def invoke_async(self, input_data, invocation_state=None):
             # Increment execution count in state
             count = self.state.get("execution_count") or 0
             self.state.set("execution_count", count + 1)
@@ -1300,7 +1300,9 @@ async def test_graph_kwargs_passing_agent(mock_strands_tracer, mock_use_span):
     test_invocation_state = {"custom_param": "test_value", "another_param": 42}
     result = await graph.invoke_async("Test kwargs passing", test_invocation_state)
 
-    kwargs_agent.invoke_async.assert_called_once_with([{"text": "Test kwargs passing"}], **test_invocation_state)
+    kwargs_agent.invoke_async.assert_called_once_with(
+        [{"text": "Test kwargs passing"}], invocation_state=test_invocation_state
+    )
     assert result.status == Status.COMPLETED
 
 
@@ -1335,5 +1337,7 @@ def test_graph_kwargs_passing_sync(mock_strands_tracer, mock_use_span):
     test_invocation_state = {"custom_param": "test_value", "another_param": 42}
     result = graph("Test kwargs passing sync", test_invocation_state)
 
-    kwargs_agent.invoke_async.assert_called_once_with([{"text": "Test kwargs passing sync"}], **test_invocation_state)
+    kwargs_agent.invoke_async.assert_called_once_with(
+        [{"text": "Test kwargs passing sync"}], invocation_state=test_invocation_state
+    )
     assert result.status == Status.COMPLETED

tests/strands/multiagent/test_swarm.py

@@ -558,7 +558,7 @@ async def test_swarm_kwargs_passing(mock_strands_tracer, mock_use_span):
     test_kwargs = {"custom_param": "test_value", "another_param": 42}
     result = await swarm.invoke_async("Test kwargs passing", test_kwargs)
 
-    assert kwargs_agent.invoke_async.call_args.kwargs == test_kwargs
+    assert kwargs_agent.invoke_async.call_args.kwargs == {"invocation_state": test_kwargs}
     assert result.status == Status.COMPLETED
 
 
@@ -572,5 +572,5 @@ def test_swarm_kwargs_passing_sync(mock_strands_tracer, mock_use_span):
     test_kwargs = {"custom_param": "test_value", "another_param": 42}
     result = swarm("Test kwargs passing sync", test_kwargs)
 
-    assert kwargs_agent.invoke_async.call_args.kwargs == test_kwargs
+    assert kwargs_agent.invoke_async.call_args.kwargs == {"invocation_state": test_kwargs}
     assert result.status == Status.COMPLETED