Pre-flight Checklist
Problem or Motivation
When a Strands agent is driven through StrandsAgent, tools that use the standard Strands mechanism for per-request data — ToolContext.invocation_state — receive nothing, because the wrapper calls stream_async(None) / stream_async(user_message) without an invocation_state argument. I'd like a supported way to pass a per-run invocation_state dict into the underlying stream_async call.
A plain Strands agent supports request-scoped data via stream_async(invocation_state=...), which tools then read through ToolContext.invocation_state. This is the documented Strands way to hand tools things like auth tokens, tenant/user ids, a DB session, or request-scoped API clients.
The wrapper faithfully forwards construction-time config (_extract_agent_kwargs), but drops this per-run channel:
# agent.py — both call sites, no invocation_state
agent_stream = strands_agent.stream_async(None)
agent_stream = strands_agent.stream_async(user_message)
So any team that (a) wraps an already-configured Strands agent with ag_ui_strands and (b) has tools reading ToolContext.invocation_state finds those tools get None under AG-UI, even though the same tools work when the agent is invoked directly. Concretely, a tool like:
@tool(context=True)
def get_data(tool_context: ToolContext):
request = tool_context.invocation_state.get("request_context")
return client(token=request.token) # AttributeError: 'NoneType' has no attribute 'token'
works on a direct agent.stream_async(invocation_state={"request_context": req}) but breaks under the wrapper.
Precedent in this repo
This is the symmetric completion of a bridge you already ship. In 2c08be2d ("fix(aws-strands): forward RunAgentInput.context to agent state") the wrapper maps RunAgentInput.context into strands_agent.state:
# agent.py
strands_agent.state.set("agui_context", agui_context)
And the adapter already reads invocation_state internally (a2ui_tool.py, invocation_state.get("agent")). So the concept is live here — per-run AG-UI input is already bridged into what tools can read; this request just extends that to the Strands-native invocation_state channel used by user tools.
Proposed Solution
Option A — invocation_state on StrandsAgentConfig (preferred; mirrors session_manager_provider)
@dataclass
class StrandsAgentConfig:
...
# static dict, or a provider called per run with the RunAgentInput
invocation_state: Optional[
Union[Dict[str, Any], Callable[[RunAgentInput], Dict[str, Any]]]
] = None
Then in run(), resolve it and pass through:
inv_state = self.config.invocation_state
if callable(inv_state):
inv_state = inv_state(input_data)
agent_stream = strands_agent.stream_async(prompt, invocation_state=inv_state or {})
This matches the existing session_manager_provider pattern (a per-run provider taking RunAgentInput), so it fits the current design and is a small, well-scoped change.
Option B — auto-map RunAgentInput.forwarded_props → invocation_state
Forward the client's forwarded_props (or a namespaced sub-key) into invocation_state automatically. Less explicit, but zero-config for hosts that already put per-request data in forwarded_props.
I'd favor A for explicitness and parity with session_manager_provider, but happy to follow whichever channel you consider canonical.
Alternatives Considered
We bind the request in a contextvars.ContextVar around the wrapper's run() and have tools fall back to it when invocation_state is empty. It works for in-task tool execution, but it's implicit and doesn't propagate into thread-offloaded sub-agents (e.g. the A2UI generate_a2ui worker-thread path uses asyncio.run in a thread, where a ContextVar set in the main task isn't visible without copy_context()). A first-class invocation_state passthrough would remove the need for the workaround and unify tool behavior across direct and AG-UI invocation.
Additional Context
No response
Pre-flight Checklist
Problem or Motivation
When a Strands agent is driven through
StrandsAgent, tools that use the standard Strands mechanism for per-request data —ToolContext.invocation_state— receive nothing, because the wrapper callsstream_async(None)/stream_async(user_message)without aninvocation_stateargument. I'd like a supported way to pass a per-runinvocation_statedict into the underlyingstream_asynccall.A plain Strands agent supports request-scoped data via
stream_async(invocation_state=...), which tools then read throughToolContext.invocation_state. This is the documented Strands way to hand tools things like auth tokens, tenant/user ids, a DB session, or request-scoped API clients.The wrapper faithfully forwards construction-time config (
_extract_agent_kwargs), but drops this per-run channel:So any team that (a) wraps an already-configured Strands agent with
ag_ui_strandsand (b) has tools readingToolContext.invocation_statefinds those tools getNoneunder AG-UI, even though the same tools work when the agent is invoked directly. Concretely, a tool like:works on a direct
agent.stream_async(invocation_state={"request_context": req})but breaks under the wrapper.Precedent in this repo
This is the symmetric completion of a bridge you already ship. In
2c08be2d("fix(aws-strands): forward RunAgentInput.context to agent state") the wrapper mapsRunAgentInput.contextintostrands_agent.state:And the adapter already reads
invocation_stateinternally (a2ui_tool.py,invocation_state.get("agent")). So the concept is live here — per-run AG-UI input is already bridged into what tools can read; this request just extends that to the Strands-nativeinvocation_statechannel used by user tools.Proposed Solution
Option A —
invocation_stateonStrandsAgentConfig(preferred; mirrorssession_manager_provider)Then in
run(), resolve it and pass through:This matches the existing
session_manager_providerpattern (a per-run provider takingRunAgentInput), so it fits the current design and is a small, well-scoped change.Option B — auto-map
RunAgentInput.forwarded_props→invocation_stateForward the client's
forwarded_props(or a namespaced sub-key) intoinvocation_stateautomatically. Less explicit, but zero-config for hosts that already put per-request data inforwarded_props.I'd favor A for explicitness and parity with
session_manager_provider, but happy to follow whichever channel you consider canonical.Alternatives Considered
We bind the request in a
contextvars.ContextVararound the wrapper'srun()and have tools fall back to it wheninvocation_stateis empty. It works for in-task tool execution, but it's implicit and doesn't propagate into thread-offloaded sub-agents (e.g. the A2UIgenerate_a2uiworker-thread path usesasyncio.runin a thread, where a ContextVar set in the main task isn't visible withoutcopy_context()). A first-classinvocation_statepassthrough would remove the need for the workaround and unify tool behavior across direct and AG-UI invocation.Additional Context
No response