From 91aede91b458b9e88b65eade71532972810e4ffa Mon Sep 17 00:00:00 2001 From: Maulik Dang Date: Tue, 27 Oct 2026 22:55:13 -0700 Subject: [PATCH 1/2] aws-transform-agent-toolkit: document streaming and make it the default - New steering/streaming.md: opt-in StreamingConfig, the factory requirement (forward conversation_repository and memory_manager), the concurrency contract (no per-request state in module/class globals), and the two subagent relay tools (send_message_to_subagent_and_relay, stream_message_to_subagent). - Make streaming the scaffolding default: orchestrator-patterns.md Option A passes streaming_config=StreamingConfig(). - Document the capabilities.streaming agent-card flag in agent-registration.md, with a caveat that it must match the server: a card advertising streaming without a StreamingConfig makes message/stream return 404 and the turn fails. - Clarify streaming is enabled in the entry-point code, not via a Dockerfile flag. - Fix a pre-existing broken import: AgentRuntimeServer is not re-exported at the package top level, so use the module path. --- aws-transform-agent-toolkit/POWER.md | 1 + .../steering/agent-registration.md | 10 ++ .../steering/orchestrator-patterns.md | 18 ++- .../steering/streaming.md | 118 ++++++++++++++++++ 4 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 aws-transform-agent-toolkit/steering/streaming.md diff --git a/aws-transform-agent-toolkit/POWER.md b/aws-transform-agent-toolkit/POWER.md index cc5bb311..f51f7a8f 100644 --- a/aws-transform-agent-toolkit/POWER.md +++ b/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 diff --git a/aws-transform-agent-toolkit/steering/agent-registration.md b/aws-transform-agent-toolkit/steering/agent-registration.md index a2cd0667..7e3dad0a 100644 --- a/aws-transform-agent-toolkit/steering/agent-registration.md +++ b/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; `simple_cli_agent_core.py --concurrent-streaming`). 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`: @@ -720,6 +729,7 @@ An orchestrator card showing subagent dependencies, connectors, and owner metada "capabilities": { "restartable": true, "a2aSupported": true, + "streaming": true, "legacyDashboard": true, "legacyTaskLink": false, "webAppV2": true, diff --git a/aws-transform-agent-toolkit/steering/orchestrator-patterns.md b/aws-transform-agent-toolkit/steering/orchestrator-patterns.md index a3579314..c2f2484d 100644 --- a/aws-transform-agent-toolkit/steering/orchestrator-patterns.md +++ b/aws-transform-agent-toolkit/steering/orchestrator-patterns.md @@ -118,6 +118,7 @@ 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.streaming import StreamingConfig from agent_builder_sdk.utils import get_prompt_with_name # Configure logging @@ -168,6 +169,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,7 +275,7 @@ 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() @@ -276,6 +283,14 @@ 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 +384,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` diff --git a/aws-transform-agent-toolkit/steering/streaming.md b/aws-transform-agent-toolkit/steering/streaming.md new file mode 100644 index 00000000..bd543568 --- /dev/null +++ b/aws-transform-agent-toolkit/steering/streaming.md @@ -0,0 +1,118 @@ +--- +inclusion: auto +name: streaming +description: "Opt-in concurrent streaming for AWS Transform orchestrators and subagents (StreamingConfig, the AG-UI/A2A stream, the subagent relay tools)" +--- + +# Streaming + +AWS Transform agents answer in two ways. By default a turn is **buffered**: the agent runs to +completion and the final answer is persisted and delivered. **Streaming** is opt-in: 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. + +New agents are scaffolded with streaming on. 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 the +buffered examples in `orchestrator-patterns.md`. + +## 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`). 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. From cb5bcd6e5958d69fbdaf8dcaa86d9688f3098d84 Mon Sep 17 00:00:00 2001 From: Maulik Dang Date: Wed, 28 Oct 2026 09:45:21 -0700 Subject: [PATCH 2/2] Address review: async orchestrator scaffold, card streaming flags, subagent/framing fixes - orchestrator-patterns.md: the main() scaffold used the sync create_default_orchestrator with a no-**kwargs factory, which fails AgentRuntimeServer's streaming-capability validation at startup. Switch to create_default_async_orchestrator and forward conversation_repository/memory_manager. - agent-registration.md: add "streaming": true to the Minimal and Complete agentCard examples so a scaffold copying them matches the streaming server default; drop the --concurrent-streaming flag from the card note and point at the entry point's StreamingConfig(). - streaming.md: state that a streamed subagent needs its own StreamingConfig and capabilities.streaming: true or ConverseStream rejects the relay; reconcile the frontmatter/intro framing (off at the SDK level, on for scaffolded agents). --- .../steering/agent-registration.md | 4 +++- .../steering/orchestrator-patterns.md | 12 ++++++---- .../steering/streaming.md | 22 ++++++++++--------- 3 files changed, 23 insertions(+), 15 deletions(-) diff --git a/aws-transform-agent-toolkit/steering/agent-registration.md b/aws-transform-agent-toolkit/steering/agent-registration.md index 7e3dad0a..61adc0e2 100644 --- a/aws-transform-agent-toolkit/steering/agent-registration.md +++ b/aws-transform-agent-toolkit/steering/agent-registration.md @@ -409,7 +409,7 @@ The `capabilities` object contains boolean flags and a required `extensions` arr > **`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; `simple_cli_agent_core.py --concurrent-streaming`). If the card says +> (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 @@ -611,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, @@ -668,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, diff --git a/aws-transform-agent-toolkit/steering/orchestrator-patterns.md b/aws-transform-agent-toolkit/steering/orchestrator-patterns.md index c2f2484d..9bb0d6c0 100644 --- a/aws-transform-agent-toolkit/steering/orchestrator-patterns.md +++ b/aws-transform-agent-toolkit/steering/orchestrator-patterns.md @@ -117,7 +117,7 @@ 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 @@ -152,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...") diff --git a/aws-transform-agent-toolkit/steering/streaming.md b/aws-transform-agent-toolkit/steering/streaming.md index bd543568..111e1576 100644 --- a/aws-transform-agent-toolkit/steering/streaming.md +++ b/aws-transform-agent-toolkit/steering/streaming.md @@ -1,19 +1,19 @@ --- inclusion: auto name: streaming -description: "Opt-in concurrent streaming for AWS Transform orchestrators and subagents (StreamingConfig, the AG-UI/A2A stream, the subagent relay tools)" +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. By default a turn is **buffered**: the agent runs to -completion and the final answer is persisted and delivered. **Streaming** is opt-in: 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. +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. -New agents are scaffolded with streaming on. 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 the -buffered examples in `orchestrator-patterns.md`. +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 @@ -99,8 +99,10 @@ instead. 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`). An orchestrator reaches a streaming subagent through one of two opt-in -tools, added via `custom_tools`: +`--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