aws-transform-agent-toolkit: document streaming and make it the default

aws-transform-agent-toolkit/POWER.md

@@ -333,6 +333,7 @@ Example workflow:
   - Skills are reusable capabilities that expand what an agent can do. The skill registry is a central repository where developers can choose from or contribute to a bank of skills to use with their agents.
 - Troubleshooting agent deployment or runtime issues → `steering/troubleshooting.md`
 - Adding an agent to an existing workflow → `steering/workflow-integration.md`
+- Streaming a turn live, or serving several users of one orchestrator concurrently → `steering/streaming.md`
 - **Deploying agents through Kiro (recommended workflow)** → `steering/deploy-agent-workflow.md`
 
 ## Next Steps

aws-transform-agent-toolkit/steering/agent-registration.md

@@ -404,8 +404,17 @@ The `capabilities` object contains boolean flags and a required `extensions` arr
 | `webAppV2` | boolean | Yes | Web app v2 support flag |
 | `legacyRestartable` | boolean | Yes | Legacy restartable support flag |
 | `chatFileUpload` | boolean | No | Whether users can upload files directly via chat | 
+| `streaming` | boolean | No | Whether AWS Transform streams turns to this agent live (see note) |
 | `extensions` | array | Yes | Required extensions (see below) |
 
+> **`streaming` must match the server.** Set `capabilities.streaming: true` only when the agent's
+> server actually runs the streaming path, i.e. `AgentRuntimeServer` is given a `StreamingConfig`
+> (the scaffolding default — the `StreamingConfig()` in the agent's entry point; see `streaming.md`). If the card says
+> `true` but the server has no `StreamingConfig`, the platform streams to an agent that answers
+> `message/stream` with HTTP 404, so the turn fails. If the card is `false` (or the flag is absent),
+> the platform uses the buffered path even when the server could stream. New agents are scaffolded
+> with both on; a buffered-only agent sets both off. See `streaming.md`.
+
 ##### Required Extensions
 
 Three extensions are **required** in `capabilities.extensions`:
@@ -602,6 +611,7 @@ The minimum structure that passes boto3 and server-side validation (works for bo
     "capabilities": {
       "restartable": true,
       "a2aSupported": true,
+      "streaming": true,
       "legacyDashboard": false,
       "legacyTaskLink": false,
       "webAppV2": true,
@@ -659,6 +669,7 @@ The minimum structure that passes boto3 and server-side validation (works for bo
     "capabilities": {
       "restartable": true,
       "a2aSupported": true,
+      "streaming": true,
       "legacyDashboard": false,
       "legacyTaskLink": false,
       "webAppV2": true,
@@ -720,6 +731,7 @@ An orchestrator card showing subagent dependencies, connectors, and owner metada
     "capabilities": {
       "restartable": true,
       "a2aSupported": true,
+      "streaming": true,
       "legacyDashboard": true,
       "legacyTaskLink": false,
       "webAppV2": true,

aws-transform-agent-toolkit/steering/orchestrator-patterns.md

@@ -117,7 +117,8 @@ Use the simplified `AgentRuntimeServer` with a custom agent factory:
 import argparse
 import logging
 from agent_builder_sdk.server.agent_runtime_server import AgentRuntimeServer
-from agent_builder_sdk.agent_factory import create_default_orchestrator
+from agent_builder_sdk.agent_factory import create_default_async_orchestrator
+from agent_builder_sdk.streaming import StreamingConfig
 from agent_builder_sdk.utils import get_prompt_with_name
 
 # Configure logging
@@ -151,13 +152,17 @@ def main():
     parser = create_parser()
     args = parser.parse_args()
 
-    # Create agent factory with default configuration
-    def agent_factory(mcp_client, storage_dir=None):
-        return create_default_orchestrator(
+    # Create agent factory with default configuration. Streaming needs an async orchestrator and a
+    # factory that forwards conversation_repository and memory_manager, so the pooled per-user
+    # agents share one history and memory store (the server validates this at startup).
+    def agent_factory(mcp_client, storage_dir=None, **kwargs):
+        return create_default_async_orchestrator(
             mcp_client=mcp_client,
             storage_dir=storage_dir,
             system_prompt=get_prompt_with_name("test_orchestrator_prompt"),
             model_id="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
+            conversation_repository=kwargs.get("conversation_repository"),
+            memory_manager=kwargs.get("memory_manager"),
         )
 
     logger.info("Starting Agent Runtime Server...")
@@ -168,6 +173,12 @@ def main():
         binary_location=args.binary_location,
         storage_dir=args.storage_dir,
         delayed_timeout=3600,
+        # Streaming is the default for new agents. The agent card must set
+        # capabilities.streaming: true to match (see agent-registration.md), and the
+        # factory must forward conversation_repository and memory_manager. To ship a
+        # buffered-only agent, drop this line AND set the card flag false — the two must agree.
+        # See streaming.md for the concurrency contract this turns on.
+        streaming_config=StreamingConfig(),
     )
 
     # This will set up everything and run the server
@@ -268,14 +279,22 @@ Both require a full rebuild → new runtime → new version → re-register cycl
 Orchestrators MUST use `AgentRuntimeServer` (NOT `StatelessAgentRuntimeServer`) with `delayed_timeout=3600`. The stateless server has a hard 28s asyncio timeout that kills long-running subagent coordination. `AgentRuntimeServer` has queue support built-in — it acks after 28s and continues processing in the background for up to `delayed_timeout` seconds.
 
 ```python
-from agent_builder_sdk import AgentRuntimeServer
+from agent_builder_sdk.server.agent_runtime_server import AgentRuntimeServer
 
 server = AgentRuntimeServer(agent_factory=agent_factory, delayed_timeout=3600)
 server.start()
 ```
 
 > **WARNING:** Do NOT pass `queue=True` — that parameter does not exist and will crash with `TypeError: __init__() got an unexpected keyword argument 'queue'`. Queue behavior is always enabled internally.
 
+### Streaming is the default
+
+New orchestrators are scaffolded with streaming on: pass `streaming_config=StreamingConfig()` to
+`AgentRuntimeServer` and set `capabilities.streaming: true` on the agent card. The two must agree —
+see the caveat in `streaming.md` and `agent-registration.md`. Streaming also requires the factory to
+forward `conversation_repository` and `memory_manager`, and brings a concurrency contract (no
+per-request state in module/class globals). Read `streaming.md` before customizing the agent.
+
 ### mcp_clients Must Be Plural (List)
 
 ```python
@@ -369,4 +388,5 @@ Use `model_id="us.anthropic.claude-sonnet-4-5-20250929-v1:0"` — bare model IDs
 - Build subagents: see `subagent-patterns.md`
 - Deploy: see `deploy-agent-workflow.md`
 - Register: see `agent-registration.md`
+- Stream turns live / serve several users concurrently: see `streaming.md`
 - Troubleshoot: see `troubleshooting.md`

aws-transform-agent-toolkit/steering/streaming.md

@@ -0,0 +1,120 @@
+---
+inclusion: auto
+name: streaming
+description: "Concurrent streaming for AWS Transform orchestrators and subagents: on by default for scaffolded agents, an SDK-level switch (StreamingConfig, the AG-UI/A2A stream, the subagent relay tools)"
+---
+
+# Streaming
+
+AWS Transform agents answer in two ways. In a **buffered** turn the agent runs to completion and the
+final answer is persisted and delivered. With **streaming** the agent's turn is emitted live as a
+stream of events (reasoning, tool calls, text) while it runs, and several users of the same
+orchestrator can have their turns run concurrently.
+
+At the SDK level streaming is off until you pass a `StreamingConfig`; this power scaffolds new agents
+with it on (below). It stays a switch: you can turn it off for a simple request/response agent, and
+then nothing in this file applies and the agent behaves exactly as a buffered agent.
+
+## When to turn it off
+
+Keep streaming on (the default) when users should see the answer appear as it is produced, or when
+one orchestrator serves several users at once. Turn it off only for a simple request/response agent
+that wants neither — the buffered path needs none of the concurrency rules below. If you turn it
+off, also set `capabilities.streaming: false` on the agent card (see the caveat below); the two must
+agree.
+
+## Enabling it
+
+Pass a `StreamingConfig` to `AgentRuntimeServer`. With no `StreamingConfig` the server does not
+stream, and a `message/stream` request returns HTTP 404 / JSON-RPC `-32601`, as before streaming
+existed.
+
+```python
+from agent_builder_sdk.server.agent_runtime_server import AgentRuntimeServer
+from agent_builder_sdk.streaming import StreamingConfig
+
+server = AgentRuntimeServer(
+    agent_factory=agent_factory,
+    delayed_timeout=3600,
+    streaming_config=StreamingConfig(),   # opt in
+)
+server.start()
+```
+
+A scaffolded agent enables streaming in its own entry point (the `StreamingConfig()` above), so the
+Dockerfile `ENTRYPOINT` needs no change — do not add a `--concurrent-streaming` flag there. That
+flag exists only on the SDK's sample `simple_cli_agent_core.py` (which also takes
+`--emit-reasoning/--no-emit-reasoning` and `--emit-tool-calls/--no-emit-tool-calls`); it applies
+only if your `ENTRYPOINT` runs that sample module directly instead of your own entry point.
+
+**The agent card flag must match the server.** AWS Transform decides whether to stream to an agent
+from its registered agent card's `capabilities.streaming`. Set it `true` only when the server
+actually runs the streaming path (a `StreamingConfig` is passed, as above). The two must agree:
+
+- card `true` + server has `StreamingConfig` → streaming works (the scaffolding default).
+- card `false` (or absent) + no `StreamingConfig` → buffered, as before (a valid buffered-only agent).
+- card `true` but server has **no** `StreamingConfig` → the platform streams to an agent that
+  answers `message/stream` with HTTP 404, so the turn fails. Do not ship this combination.
+- card `false` but server has `StreamingConfig` → the platform never streams; users only get
+  buffered answers even though the server could stream.
+
+See `agent-registration.md` for where this flag lives in the card.
+
+## How it works
+
+Strands runs one invocation at a time per `Agent`, so with streaming on the server keeps a pool with
+one agent per source (a user, or another agent that streams to yours). Agents are built on first use
+and evicted after an idle TTL.
+
+- Each pooled agent builds its **own** model client from your factory. The MCP client, the
+  conversation repository, and the memory manager are **shared**, so a user's buffered and streamed
+  turns see the same history.
+- A synchronous factory is run on a worker thread, one build at a time, so building a new user's
+  agent does not stall live streams.
+
+## Requirements for your factory
+
+Your agent factory must accept and forward `conversation_repository` and `memory_manager`, so the
+pooled agents share one history and memory store. The server checks this at startup.
+
+```python
+def agent_factory(mcp_client, storage_dir, **kwargs):
+    return create_default_async_orchestrator(
+        mcp_client=mcp_client,
+        storage_dir=storage_dir,
+        system_prompt=...,
+        conversation_repository=kwargs.get("conversation_repository"),
+        memory_manager=kwargs.get("memory_manager"),
+    )
+```
+
+## Concurrency contract (important)
+
+With streaming on, several users' turns run concurrently on one event loop. Tools and hooks MUST NOT
+keep per-request state in module-level or class-level mutable globals — that state would leak between
+users. Use a `contextvars.ContextVar`, or the per-request `ToolContext` / `invocation_state`,
+instead.
+
+## Subagents
+
+A subagent can stream too, but only an async subagent is `StreamingCapable`: build it with
+`create_default_async_subagent` (the sample `simple_subagent_cli.py` does this under
+`--concurrent-streaming`). The subagent's own server must also be given a `StreamingConfig` and its
+agent card must set `capabilities.streaming: true`; otherwise the agentic API's ConverseStream
+rejects the relay with `ValidationException: Streaming is not supported for the requested agent`. An
+orchestrator reaches a streaming subagent through one of two opt-in tools, added via `custom_tools`:
+
+- `send_message_to_subagent_and_relay` — streams the subagent's answer to the user **as the
+  orchestrator's own final answer**, and ends the orchestrator's turn. Use it when the subagent's
+  answer is exactly what the user should get and the orchestrator has nothing to add.
+- `stream_message_to_subagent` — returns the subagent's answer (its last model cycle) to the
+  orchestrator without streaming it to the user, so the orchestrator can reason over it and keep
+  working. Like `send_message_to_subagent`, but streamed under the hood.
+
+## Interrupted turns
+
+A streamed turn that is cut short (client disconnect, the turn's time budget, a crash) is closed out
+when the conversation is next loaded: the unanswered user message gets a placeholder reply so the
+next turn answers its own prompt, and a turn ended by a terminating tool keeps that tool's response
+as the reply. You do not need to handle this yourself; it keeps the stored history valid for the
+next turn.