[FEATURE] Implement native async iterator support

[awsarron]

Problem Statement

The Agent class provides the stream_async function that returns an AsyncIterator:

sdk-python/src/strands/agent/agent.py

Line 356 in 77f5fa7

async def stream_async(self, prompt: str, **kwargs: Any) -> AsyncIterator[Any]:

The current stream_async implementation calls _run_loop and uses threads with a queue to yield streamed data events from the Strands Agents event loop:

sdk-python/src/strands/agent/agent.py

Line 423 in 77f5fa7

def _run_loop(

sdk-python/src/strands/agent/agent.py

Lines 409 to 421 in 77f5fa7

	thread = Thread(target=target_callback, daemon=True)
	thread.start()
	try:
	while True:
	item = await queue.get()
	if item == _stop_event:
	break
	if isinstance(item, BaseException):
	raise item
	yield item
	finally:
	thread.join()

The problem is that this threading implementation doesn't work with Python features like thread local contextvars. Certain Python libraries (e.g. https://github.com/UKGovernmentBEIS/inspect_ai) use thread local context vars and require them to function correctly. The only reasonable solution for correctly supporting concurrency within a single thread is async.

This issue is to propose, discuss, and refine a true async implementation of the Agent.stream_async function and Strands Agent event loop.

Proposed Solution

No response

Use Case

• Thread local contextvars
• A more maintainable implementation of async agent requests
• Allows for future enhancements that enable streamed tool results

Alternatives Solutions

No response

Additional Context

No response

[zastrowm]

Another use case is providing caller control over the lifecycle of an agent request. Right now all events are enqueued eagerly - meaning that if even when callers stop iterating, the agent continues until it hits a natural stopping point. In theory, an async iterator could/would be pull-based, allowing the agent to suspend if the caller stops iterating which should enable more advanced use cases or control.

[pgrayy]

The following is a high level breakdown of the tasks required to achieve true async support:

Part 1: Core Async Implementation (Est: 7 Days)

• Reimplement stream_async as a real async method:
  • Await the event loop instead of using threads
  • Modify event loop to yield events individually as opposed to returning one final payload
  • Await model provider converse calls
  • Support awaiting async tool implementations
  • Support awaiting async callback implementations
  • Support awaiting direct async tool invocations
  • Call the async event loop synchronously from the existing synchronous __call__ method
• Write tests and documentation for the new async capabilities

Part 2: Formalize Stream Events Schema (Est: TBD)

• Formalize schema and ordering of events yielded from stream_async
• Define consistent event structure for all streaming operations
• Document the event schema for developers

Part 3: Model Provider Async Implementation (Est: TBD)

• Convert the underlying clients of each model provider to async:
  • AnthropicModel
  • BedrockModel
  • LiteLLMModel
  • LlamaAPIModel
  • OllamaModel
  • OpenAIModel

Part 4: Tool Async Implementation (Est: TBD)

• Reimplement packaged tools to be async where appropriate (e.g., use_llm)
• Update tool decorator to support async functions

Part 5: Callback Handler Async Implementation (Est: TBD)

• Define async callback handler interface
• Reimplement packaged callback handlers to be async where appropriate

---

Part 1 is underway right now.