diff --git a/aws-transform-agent-toolkit/POWER.md b/aws-transform-agent-toolkit/POWER.md index cc5bb31..f51f7a8 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 a2cd066..61adc0e 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 — 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, diff --git a/aws-transform-agent-toolkit/steering/orchestrator-patterns.md b/aws-transform-agent-toolkit/steering/orchestrator-patterns.md index a357931..9bb0d6c 100644 --- a/aws-transform-agent-toolkit/steering/orchestrator-patterns.md +++ b/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,7 +279,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 +287,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 +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` diff --git a/aws-transform-agent-toolkit/steering/streaming.md b/aws-transform-agent-toolkit/steering/streaming.md new file mode 100644 index 0000000..111e157 --- /dev/null +++ b/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.