diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3c72438..150ce01 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,7 +26,7 @@ jobs: steps: - uses: actions/checkout@v7 - - uses: astral-sh/setup-uv@v9.0.0 + - uses: astral-sh/setup-uv@v10.0.1 with: enable-cache: true @@ -59,7 +59,7 @@ jobs: steps: - uses: actions/checkout@v7 - - uses: astral-sh/setup-uv@v9.0.0 + - uses: astral-sh/setup-uv@v10.0.1 with: enable-cache: true python-version: ${{ matrix.python-version }} diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 0f1e906..06478b4 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -40,7 +40,7 @@ jobs: steps: - uses: actions/checkout@v7 - - uses: astral-sh/setup-uv@v9.0.0 + - uses: astral-sh/setup-uv@v10.0.1 with: enable-cache: true diff --git a/.gitignore b/.gitignore index 997303a..d61ede6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ .env .claude .kiro/specs +.kiro/settings .sessions .converter .hypothesis diff --git a/.kiro/skills/library-development/SKILL.md b/.kiro/skills/library-development/SKILL.md index 3267dbb..65f63dc 100644 --- a/.kiro/skills/library-development/SKILL.md +++ b/.kiro/skills/library-development/SKILL.md @@ -35,7 +35,7 @@ load it whenever you are unsure where something goes. 3. **The pipeline flows one way** — text -> dict -> validated schema -> live objects. Never resolve during parsing; never parse during resolution (see The Pipeline). 4. **Explicit over implicit** — no auto-registration, no global singletons, no hidden state. Every object is wired by hand and passed as an argument. 5. **Single responsibility** — each module does one thing; one resolver per config concept, one builder per orchestration mode. -6. **Composition over inheritance** — small functions and focused modules that compose. The only base classes are the strands-facing ones (`MCPServer`, `StreamConverter`, `HookProvider`). +6. **Composition over inheritance** — small functions and focused modules that compose. The only base classes are the strands-facing ones (`MCPServer`, `HookProvider`). 7. **Smallest reasonable change** — don't refactor unrelated code to land a feature. --- diff --git a/.kiro/skills/library-development/references/project-map.md b/.kiro/skills/library-development/references/project-map.md index 2d59341..c4c2e65 100644 --- a/.kiro/skills/library-development/references/project-map.md +++ b/.kiro/skills/library-development/references/project-map.md @@ -45,13 +45,12 @@ src/strands_compose/ ├── tools/ │ ├── loaders.py # resolve_tool_spec(s) — module/file/dir → AgentTool │ ├── extractors.py # extract_last_message · serialize_multiagent_result -│ └── wrappers.py # node_as_tool / node_as_async_tool — wrap a node as a delegate tool +│ └── wrappers.py # multiagent_as_tool — wrap a Swarm/Graph as a delegate tool ├── hooks/ # reusable HookProvider implementations │ ├── event_publisher.py # EventPublisher — strands hook events → StreamEvent (the key one) │ ├── stop_guard.py # StopGuard / MultiAgentStopGuard — external cancel signal │ ├── max_calls_guard.py # MaxToolCallsGuard — tool-call circuit breaker │ └── tool_name_sanitizer.py# ToolNameSanitizer — repair model-mangled tool names -├── converters/ # StreamEvent → protocol chunks (base ABC · openai · raw) ├── renderers/ # terminal output (base ABC · ansi) └── startup/ # opt-in health checks (validator.py) + report (report.py) ``` diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index d14bb0b..6ac059e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,8 +1,7 @@ # https://pre-commit.com # https://pre-commit.com/hooks.html -# Use the active venv Python — avoids path issues on Windows where -# pre-commit looks for python3.12 at ~/.local/bin/python.exe +# Use the active venv Python default_language_version: python: python3 @@ -29,7 +28,7 @@ repos: - id: trailing-whitespace - repo: https://github.com/astral-sh/ruff-pre-commit - rev: 'v0.15.20' + rev: 'v0.16.3' hooks: - id: ruff # lint — commit + push - id: ruff-format # format — commit + push @@ -42,7 +41,7 @@ repos: files: \.(py|yaml|yml|md|toml|json|env)$ - repo: https://github.com/commitizen-tools/commitizen - rev: 'v4.16.4' + rev: 'v4.17.0' hooks: - id: commitizen # validates commit message format stages: [commit-msg] # only on commit-msg event, not push diff --git a/AGENTS.md b/AGENTS.md index e5ee5fa..ed16d5c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,6 +7,33 @@ subclasses. --- +## Response Style + +Answer with a short executive summary. Lead with the answer or decision in the +first line. + +- Direct question -> 1-3 sentences. Yes/no question -> start with yes or no. +- Expand only when asked for details, code, a full review, or a rationale. +- Prefer a compact table or 3-5 bullets over prose sections. +- Report the outcome, not the journey. + +### Do not + +- No multi-section reports, headers, or background unless requested. +- No narrating what was searched, read, or what sub-agents found. +- No restating earlier turns, no recaps of prior decisions. +- No storytelling, no "what settled it" / "where I was wrong" essays when one line will do. +- No praise, filler, or hedging. + +### Still required + +Brevity never overrides correctness. Keep these even when short: + +- State uncertainty plainly, and say what was verified vs assumed. +- Correct a wrong earlier statement if it would change a decision. +- Flag risk before a destructive or hard-to-reverse action. + + ## Read the Skill First — MANDATORY Before touching any code, load the skill for the area you are working in. Skills diff --git a/README.md b/README.md index c20dc37..b0bd095 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,7 @@ Strands Compose is an ecosystem that includes the following packages: |-------|---------|-------------| | **Define the agents** | [**strands-compose**](https://github.com/strands-compose/sdk-python) | Developers | | Run / deploy the agents | [strands-compose-agentcore](https://github.com/strands-compose/bedrock-agentcore) | Developers, operations | -|*Put the agents in front of people | [strands-compose-chat](https://github.com/strands-compose/chat-ui) | **End users** | +|Put the agents in front of people | [strands-compose-chat](https://github.com/strands-compose/chat) | **End users** | --- diff --git a/docs/configuration/Chapter_10.md b/docs/configuration/Chapter_10.md index 5fc2a80..e6bbbb9 100644 --- a/docs/configuration/Chapter_10.md +++ b/docs/configuration/Chapter_10.md @@ -43,7 +43,9 @@ orchestrations: entry: team ``` -**How it works**: strands-compose **forks** a new agent from the `entry_name` agent's blueprint (model, system_prompt, hooks, tools) and adds delegate tools for each connection. The original `coordinator` agent is **never mutated**. Each connection becomes an async tool that the coordinator can call. +**How it works**: strands-compose **forks** a new agent from the `entry_name` agent's blueprint (model, system_prompt, hooks, tools) and adds delegate tools for each connection. The original `coordinator` agent is **never mutated**. + +> **Two limits.** A delegate cannot be called twice concurrently — the second call comes back as a tool error, so declare a second agent to fan out. And an interrupt (such as an approval) can be answered and resumed only for an agent connection; raised inside a nested Swarm or Graph it reaches the coordinator as a tool error. **Fields**: @@ -54,6 +56,7 @@ entry: team | `connections` | list | Yes | Sub-agents to wire as tools | | `connections[].agent` | string | Yes | Name of the target agent or orchestration | | `connections[].description` | string | Yes | Tool description the LLM sees | +| `connections[].preserve_context` | bool | No | Keep the delegate's history between calls (default `true`). Set `false` for a stateless delegate that starts from its construction-time baseline every call. Rejected for a nested orchestration, or for an agent carrying a session manager | | `session_manager` | dict | No | Override session manager for the forked agent | | `hooks` | list | No | Additional hooks for the forked agent | | `agent_kwargs` | dict | No | Override agent kwargs (merged with entry agent's kwargs) | diff --git a/docs/configuration/Chapter_12.md b/docs/configuration/Chapter_12.md index 8d5953d..677f5db 100644 --- a/docs/configuration/Chapter_12.md +++ b/docs/configuration/Chapter_12.md @@ -53,7 +53,7 @@ entry: manager 1. strands-compose collects all orchestration dependencies. 2. It performs a **topological sort** — inner orchestrations are built before outer ones. 3. Built orchestrations become nodes in the node pool, available for outer orchestrations to reference. -4. For delegate mode, inner orchestrations are wrapped as async tools (just like regular agents). +4. For delegate mode, inner orchestrations become callable tools on the coordinator. ## Circular Dependencies diff --git a/examples/12_streaming/README.md b/examples/12_streaming/README.md index 55007f5..26cde83 100644 --- a/examples/12_streaming/README.md +++ b/examples/12_streaming/README.md @@ -57,8 +57,8 @@ bracketing all per-agent activity. `invoke_async` so both the agent and the queue consumer share the same event loop. **`AnsiRenderer` is optional.** It's a convenience for terminals. In production you'd -consume the queue and convert events to SSE chunks (see `OpenAIStreamConverter`) or -NDJSON (`RawStreamConverter`). +consume the queue and serialize each event into whatever wire format your transport +needs — SSE chunks, NDJSON, or your own envelope. **`queue.flush()`** resets the queue between turns so events from one invocation don't leak into the next. It also resets the `session_start` / `session_end` guards. diff --git a/pyproject.toml b/pyproject.toml index 7d49a7e..e0a5017 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "strands-agents>=1.48.0,<2.0.0", + "strands-agents>=1.52.0,<2.0.0", "pydantic>=2.12.5", "pyyaml>=6.0.0", "mcp>=1.24.0", @@ -36,16 +36,16 @@ agentcore-memory = [ "bedrock-agentcore>=1.4.0", ] ollama = [ - "strands-agents[ollama]>=1.48.0,<2.0.0", + "strands-agents[ollama]>=1.52.0,<2.0.0", ] openai = [ - "strands-agents[openai]>=1.48.0,<2.0.0", + "strands-agents[openai]>=1.52.0,<2.0.0", ] gemini = [ - "strands-agents[gemini]>=1.48.0,<2.0.0", + "strands-agents[gemini]>=1.52.0,<2.0.0", ] anthropic = [ - "strands-agents[anthropic]>=1.48.0,<2.0.0", + "strands-agents[anthropic]>=1.52.0,<2.0.0", ] [project.urls] @@ -57,16 +57,16 @@ Changelog = "https://github.com/strands-compose/sdk-python/blob/main/CHANGELOG.m [dependency-groups] dev = [ - "ty>=0.0.29", + "ty>=0.0.72", "bandit>=1.9.2", - "coverage>=7.12.0", + "coverage>=7.15.4", "pytest-asyncio>=1.2.0", "pytest-cov>=7.0.0", "pytest-mock>=3.15.1", "pytest-xdist>=3.8.0", - "pytest>=9.0.2", - "ruff>=0.14.8", - "rust-just>=1.42.4", + "pytest>=9.1.1", + "ruff>=0.16.3", + "rust-just>=1.58.0", "commitizen>=4.8.4", "pre-commit>=4.3.0", "hypothesis>=6.155.7", diff --git a/src/strands_compose/__init__.py b/src/strands_compose/__init__.py index 2dcbae7..196562a 100644 --- a/src/strands_compose/__init__.py +++ b/src/strands_compose/__init__.py @@ -24,8 +24,7 @@ from .mcp import MCPLifecycle, create_mcp_client, create_mcp_server from .renderers import AnsiRenderer from .tools import ( - node_as_async_tool, - node_as_tool, + multiagent_as_tool, serialize_multiagent_result, ) from .types import EventType, StreamEvent @@ -59,8 +58,7 @@ "load_config", "load_session", "make_event_queue", - "node_as_async_tool", - "node_as_tool", + "multiagent_as_tool", "resolve_infra", "serialize_multiagent_result", ] diff --git a/src/strands_compose/config/interpolation.py b/src/strands_compose/config/interpolation.py index 053326c..65573d1 100644 --- a/src/strands_compose/config/interpolation.py +++ b/src/strands_compose/config/interpolation.py @@ -43,14 +43,16 @@ def interpolate( resolved_vars = dict(variables or {}) resolved_env = env if env is not None else dict(os.environ) - # Pass 1: resolve vars against env only (lenient — keeps ${VAR} if absent) - resolved_vars = {k: _walk_lenient(v, {}, resolved_env) for k, v in resolved_vars.items()} + # Passes 1 and 2 run non-strict: a var may legitimately reference another var + # that is not resolved yet, so unresolved ${...} must survive to be retried. + # Pass 1: resolve vars against env only. + resolved_vars = {k: _walk(v, {}, resolved_env, strict=False) for k, v in resolved_vars.items()} # Pass 2: resolve vars sequentially so each resolved var is immediately # available to subsequent entries (handles chains like A -> B -> C). pass2: dict[str, Any] = {} for k, v in resolved_vars.items(): - pass2[k] = _walk_lenient(v, pass2, resolved_env) + pass2[k] = _walk(v, pass2, resolved_env, strict=False) resolved_vars = pass2 # Validate: remaining ${...} means circular or undefined reference. @@ -64,7 +66,8 @@ def interpolate( f"Check for circular references or undefined variables." ) - return _walk(raw, resolved_vars, resolved_env) + # The config itself is strict: every reference must resolve here. + return _walk(raw, resolved_vars, resolved_env, strict=True) def strip_anchors(raw: dict[str, Any]) -> dict[str, Any]: @@ -83,14 +86,27 @@ def _walk( data: Any, variables: dict[str, Any], env: dict[str, str], + *, + strict: bool, ) -> Any: - """Recursively walk data and interpolate string values.""" + """Recursively walk data and interpolate string values. + + Args: + data: Any parsed YAML value. + variables: Resolved ``vars:`` values. + env: Environment variables. + strict: Raise on an unresolved reference; when ``False`` leave the + original ``${expr}`` in place for the caller to validate later. + + Returns: + The value with strings interpolated. + """ if isinstance(data, dict): - return {k: _walk(v, variables, env) for k, v in data.items()} + return {k: _walk(v, variables, env, strict=strict) for k, v in data.items()} if isinstance(data, list): - return [_walk(item, variables, env) for item in data] + return [_walk(item, variables, env, strict=strict) for item in data] if isinstance(data, str) and "${" in data: - return _interpolate_string(data, variables, env) + return _interpolate_string(data, variables, env, strict=strict) return data @@ -98,19 +114,29 @@ def _interpolate_string( value: str, variables: dict[str, Any], env: dict[str, str], + *, + strict: bool, ) -> Any: - """Interpolate all ${...} patterns in a single string value. + """Interpolate all ``${...}`` patterns in a single string value. + + A string that is exactly one ``${VAR}`` keeps the value's original type + (an int stays an int); a mixed string concatenates everything as text. - If the entire string is a single ``${VAR}`` reference, the resolved value - is returned in its original type (e.g. int stays int). Otherwise, all - resolved values are cast to str and concatenated. + Args: + value: The string to interpolate. + variables: Resolved ``vars:`` values. + env: Environment variables. + strict: Raise on an unresolved reference instead of leaving it in place. + + Returns: + The interpolated value, typed when the whole string was one reference. """ match = _VAR_PATTERN.fullmatch(value) if match is not None: - return _resolve(match.group(1), variables, env) + return _resolve(match.group(1), variables, env, strict=strict) def _replacer(m: re.Match[str]) -> str: - return str(_resolve(m.group(1), variables, env)) + return str(_resolve(m.group(1), variables, env, strict=strict)) return _VAR_PATTERN.sub(_replacer, value) @@ -119,8 +145,24 @@ def _resolve( expr: str, variables: dict[str, Any], env: dict[str, str], + *, + strict: bool, ) -> Any: - """Resolve a single variable expression like ``VAR`` or ``VAR:-default``.""" + """Resolve one expression like ``VAR`` or ``VAR:-default``. + + Args: + expr: The text inside ``${...}``. + variables: Resolved ``vars:`` values. + env: Environment variables. + strict: Raise when unresolved; when ``False`` return the original + ``${expr}`` so the vars pre-passes can run again over it. + + Returns: + The resolved value, or ``${expr}`` unchanged when not strict. + + Raises: + ValueError: Strict mode and the variable has no value and no default. + """ var_name, *rest = expr.split(":-", 1) default: str | None = rest[0] if rest else None @@ -133,64 +175,11 @@ def _resolve( if default is not None: return default + if not strict: + return f"${{{expr}}}" + raise ValueError( f"Variable '${{{var_name}}}' is not set in 'vars:' or environment, " f"and no default was provided.\n" f"Use ${{{var_name}:-fallback}} to set a fallback value." ) - - -# --------------------------------------------------------------------------- -# Lenient variants — used only during the vars pre-resolution passes. -# They return the original ${expr} pattern unchanged instead of raising, so -# unresolved references survive to the post-pass validation step. -# --------------------------------------------------------------------------- - - -def _walk_lenient( - data: Any, - variables: dict[str, Any], - env: dict[str, str], -) -> Any: - """Recursively walk data and interpolate strings; leaves ${VAR} unchanged if unresolved.""" - if isinstance(data, dict): - return {k: _walk_lenient(v, variables, env) for k, v in data.items()} - if isinstance(data, list): - return [_walk_lenient(item, variables, env) for item in data] - if isinstance(data, str) and "${" in data: - return _interpolate_string_lenient(data, variables, env) - return data - - -def _interpolate_string_lenient( - value: str, - variables: dict[str, Any], - env: dict[str, str], -) -> Any: - """Interpolate ${...} patterns; returns ${expr} unchanged if variable not found.""" - match = _VAR_PATTERN.fullmatch(value) - if match is not None: - return _resolve_lenient(match.group(1), variables, env) - - def _replacer(m: re.Match[str]) -> str: - return str(_resolve_lenient(m.group(1), variables, env)) - - return _VAR_PATTERN.sub(_replacer, value) - - -def _resolve_lenient( - expr: str, - variables: dict[str, Any], - env: dict[str, str], -) -> Any: - """Resolve a single variable expression; returns ${expr} unchanged if not found.""" - var_name, *rest = expr.split(":-", 1) - default: str | None = rest[0] if rest else None - - if var_name in variables: - return variables[var_name] - if var_name in env: - return env[var_name] - if default is not None: - return default - return f"${{{expr}}}" diff --git a/src/strands_compose/config/resolvers/orchestrations/__init__.py b/src/strands_compose/config/resolvers/orchestrations/__init__.py index 2d53eaa..55d095d 100644 --- a/src/strands_compose/config/resolvers/orchestrations/__init__.py +++ b/src/strands_compose/config/resolvers/orchestrations/__init__.py @@ -10,7 +10,7 @@ from typing import TYPE_CHECKING -from ....tools import node_as_async_tool, node_as_tool +from ....tools import multiagent_as_tool from .builders import ( OrchestrationBuilder, build_delegate, @@ -75,7 +75,6 @@ def resolve_orchestrations( "build_delegate", "build_graph", "build_swarm", - "node_as_async_tool", - "node_as_tool", + "multiagent_as_tool", "resolve_orchestrations", ] diff --git a/src/strands_compose/config/resolvers/orchestrations/builders.py b/src/strands_compose/config/resolvers/orchestrations/builders.py index 3458bcc..a9131fe 100644 --- a/src/strands_compose/config/resolvers/orchestrations/builders.py +++ b/src/strands_compose/config/resolvers/orchestrations/builders.py @@ -9,15 +9,16 @@ import logging from collections.abc import Mapping -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from strands import Agent from strands.multiagent import GraphBuilder, Swarm from ....exceptions import ConfigurationError -from ....tools import node_as_async_tool +from ....tools import multiagent_as_tool from ....utils import load_object from ...schema import ( + DelegateConnectionDef, DelegateOrchestrationDef, GraphOrchestrationDef, OrchestrationDef, @@ -32,6 +33,7 @@ from strands.models import Model from strands.multiagent.graph import Graph from strands.tools.mcp import MCPClient as StrandsMCPClient + from strands.types.tools import AgentTool from ....types import Node from ...schema import AgentDef, SessionManagerDef @@ -152,6 +154,30 @@ def _dispatch( raise ConfigurationError(f"Unknown orchestration config type: {type(cfg).__name__}") +def _delegate_tool(orch_name: str, conn: DelegateConnectionDef, node: Node) -> AgentTool: + """Wrap one connection's target as a delegate tool. + + ``Agent.as_tool`` is the only path that can resume a sub-agent interrupt, and it + validates its own accepted combinations. ``preserve_context`` on an orchestration + is ours to reject — strands has no say there. + """ + if not isinstance(node, Agent): + if not conn.preserve_context: + raise ConfigurationError( + f"Orchestration '{orch_name}': connection to '{conn.agent}' sets " + f"preserve_context: false, but '{conn.agent}' is a " + f"{type(node).__name__} orchestration with no baseline to reset to.\n" + f"Fix: drop 'preserve_context: false', or point it at an agent." + ) + return multiagent_as_tool(node, name=conn.agent, description=conn.description) + + return node.as_tool( + name=conn.agent, + description=conn.description, + preserve_context=conn.preserve_context, + ) + + def build_delegate( name: str, config: DelegateOrchestrationDef, @@ -198,15 +224,10 @@ def build_delegate( f"Available agents: {sorted(agent_defs)}" ) - # Wrap each connection target as an async delegate tool. - delegate_tools: list[Any] = [] + # Wrap each connection target as a delegate tool. + delegate_tools: list[AgentTool] = [] for conn in config.connections: - target_node = nodes[conn.agent] - delegate_tool = node_as_async_tool( - target_node, - description=conn.description, - ) - delegate_tools.append(delegate_tool) + delegate_tools.append(_delegate_tool(name, conn, nodes[conn.agent])) logger.info("tool=<%s>, orchestration=<%s> | delegate tool prepared", conn.agent, name) # Resolve orchestration-level hooks. diff --git a/src/strands_compose/config/schema.py b/src/strands_compose/config/schema.py index 4f4848b..112f2ac 100644 --- a/src/strands_compose/config/schema.py +++ b/src/strands_compose/config/schema.py @@ -193,6 +193,12 @@ class DelegateConnectionDef(BaseModel): agent: str description: str + preserve_context: bool = True + """Whether the delegate keeps its history between calls. + + ``False`` resets an agent to its construction-time baseline every call. + Incompatible with a session manager, and unsupported for an Swarm and Graph. + """ class DelegateOrchestrationDef(BaseModel): diff --git a/src/strands_compose/converters/__init__.py b/src/strands_compose/converters/__init__.py deleted file mode 100644 index 1c2a486..0000000 --- a/src/strands_compose/converters/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -"""Stream converters for transforming StreamEvents into response formats.""" - -from __future__ import annotations - -from .base import StreamConverter -from .openai import OpenAIStreamConverter -from .raw import RawStreamConverter - -__all__ = [ - "OpenAIStreamConverter", - "RawStreamConverter", - "StreamConverter", -] diff --git a/src/strands_compose/converters/base.py b/src/strands_compose/converters/base.py deleted file mode 100644 index 1736323..0000000 --- a/src/strands_compose/converters/base.py +++ /dev/null @@ -1,54 +0,0 @@ -"""Abstract base class for StreamEvent converters.""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import Any - -from ..types import StreamEvent - - -class StreamConverter(ABC): - """Converts StreamEvent objects into protocol-specific output chunks. - - Each converter is stateful across one completion stream (tracks - message id, created timestamp, tool call index, etc.). Create a - new instance per request — do not share across concurrent requests. - """ - - @abstractmethod - def convert(self, event: StreamEvent) -> list[dict[str, Any]]: - """Convert one StreamEvent into zero or more output chunks. - - Returns a list (possibly empty) of serializable dicts. - The transport layer is responsible for serializing and framing - (e.g. 'data: {json}\\n\\n' for SSE). This method returns data - shapes only — never pre-serialized strings. - - Args: - event: The StreamEvent to convert. - - Returns: - A list of serializable dicts representing output chunks. - """ - ... - - @abstractmethod - def done_marker(self) -> str: - """Terminal sentinel to emit after the stream ends. - - E.g. 'data: [DONE]\\n\\n' for OpenAI SSE. - Return empty string if the protocol needs no terminator. - - Returns: - The terminal string to emit, or empty string if none. - """ - ... - - def content_type(self) -> str: - """MIME type for the HTTP streaming response. - - Returns: - The MIME type string. - """ - return "text/event-stream" diff --git a/src/strands_compose/converters/openai.py b/src/strands_compose/converters/openai.py deleted file mode 100644 index 3a7052f..0000000 --- a/src/strands_compose/converters/openai.py +++ /dev/null @@ -1,381 +0,0 @@ -from __future__ import annotations - -import html -import json -import time as _time -import uuid -from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, Literal - -from ..types import EventType -from .base import StreamConverter - -if TYPE_CHECKING: - from ..types import StreamEvent - - -@dataclass -class _ToolCallFrame: - """Tracks one open tool call or node for the details-block lifecycle.""" - - call_id: str - name: str - arguments_json: str - # Populated by TOOL_END / NODE_STOP - result_text: str | None = field(default=None) - - -class OpenAIStreamConverter(StreamConverter): - """Stateful converter from :class:`~strands_compose.wire.StreamEvent` to OpenAI ``chat.completion.chunk`` dicts. - - Targets Open WebUI and LibreChat. Translates strands events into the - OpenAI Chat Completions streaming protocol (v1) with reasoning extensions - from DeepSeek (``reasoning_content``) and OpenRouter (``reasoning``). - - Single-use per stream. Call :meth:`reset` to reuse across requests. - """ - - def __init__( - self, - *, - entry_agent_name: str, - model_label: str | None = None, - completion_id: str | None = None, - reasoning_field_mode: Literal["both", "deepseek", "openrouter", "none"] = "both", - tool_result_render: Literal["details_block", "none"] = "details_block", - emit_usage_chunk: bool = False, - verbosity: Literal["compact", "narrate"] = "compact", - ) -> None: - """Initialize the OpenAIStreamConverter. - - Args: - entry_agent_name: Name of the agent that owns the user-visible turn. - TOKEN and REASONING from all other agents are suppressed. - NODE_START/NODE_STOP from this agent surface sub-agents as - ``
`` blocks alongside tool calls. - model_label: Value for every chunk's ``model`` field. Defaults to - ``entry_agent_name``. Pass the ``model`` from the incoming - request body so the response echoes it faithfully. - completion_id: Fixed id for the whole stream. Defaults to a fresh - ``chatcmpl-`` on each instantiation. - reasoning_field_mode: Which ``delta`` field(s) carry reasoning. - - - ``"both"`` (default): ``reasoning_content`` (Open WebUI / - DeepSeek) **and** ``reasoning`` (LibreChat / OpenRouter). - - ``"deepseek"``: only ``reasoning_content``. - - ``"openrouter"``: only ``reasoning``. - - ``"none"``: reasoning dropped from the stream. - tool_result_render: How tool calls and results are surfaced. - - - ``"details_block"`` (default): a single completed - ``
`` block emitted on - TOOL_END / NODE_STOP, carrying the call inputs as HTML - attributes and the result as the body. No native - ``delta.tool_calls`` chunks are produced because strands has - already executed the tool by the time the stream emits it; - emitting native deltas without a closing - ``finish_reason: "tool_calls"`` only confuses clients into - looping on the response. - - ``"none"``: tool calls and results are not surfaced at all. - emit_usage_chunk: When ``True``, deliver usage as a separate trailing - chunk with ``choices: []`` matching ``stream_options: {include_usage: true}``. - When ``False`` (default), usage is stapled to the ``finish_reason`` chunk. - verbosity: ``"compact"`` (default) streams only entry-agent tokens. - ``"narrate"`` also streams sub-agent tokens inside the active - node's ``
`` body. - """ - self._entry_agent_name = entry_agent_name - self._model_label: str = model_label if model_label is not None else entry_agent_name - self._completion_id = completion_id or f"chatcmpl-{uuid.uuid4().hex[:24]}" - self._created = int(_time.time()) - self._reasoning_field_mode = reasoning_field_mode - self._tool_result_render = tool_result_render - self._emit_usage_chunk = emit_usage_chunk - self._verbosity = verbosity - - # Mutable stream state — reset on each call to reset() - self._sent_role = False - self._open_tool_calls: dict[str, _ToolCallFrame] = {} - self._open_node_frames: dict[str, _ToolCallFrame] = {} - self._reasoning_tokens: int = 0 - - # ── Public API ──────────────────────────────────────────────────────────── - - def convert(self, event: StreamEvent) -> list[dict[str, Any]]: - """Convert one :class:`~strands_compose.wire.StreamEvent` into OpenAI chunk(s). - - Args: - event: The event to convert. - - Returns: - A list of ``chat.completion.chunk`` dicts (possibly empty). - The transport layer is responsible for ``data: {json}\\n\\n`` framing. - """ - is_entry = event.agent_name == self._entry_agent_name - - dispatch: dict[str, Any] = { - EventType.TOKEN: self._handle_token, - EventType.REASONING: self._handle_reasoning, - EventType.TOOL_START: self._handle_tool_start, - EventType.TOOL_END: self._handle_tool_end, - EventType.AGENT_COMPLETE: self._handle_agent_complete, - EventType.MULTIAGENT_COMPLETE: self._handle_multiagent_complete, - EventType.ERROR: self._handle_error, - EventType.NODE_START: self._handle_node_start, - EventType.NODE_STOP: self._handle_node_stop, - } - - handler = dispatch.get(event.type) - if handler is not None: - return handler(event, is_entry) - return [] - - def done_marker(self) -> str: - """Return the OpenAI SSE stream terminator. - - Returns: - The ``data: [DONE]\\n\\n`` sentinel string. - """ - return "data: [DONE]\n\n" - - def reset(self) -> None: - """Reset all mutable stream state for reuse across requests. - - Preserves constructor configuration and regenerates ``completion_id`` - and ``created`` for the new stream. - """ - self._completion_id = f"chatcmpl-{uuid.uuid4().hex[:24]}" - self._created = int(_time.time()) - self._sent_role = False - self._open_tool_calls = {} - self._open_node_frames = {} - self._reasoning_tokens = 0 - - # ── Internal helpers ────────────────────────────────────────────────────── - - def _base(self) -> dict[str, Any]: - """Shared envelope skeleton for every ``chat.completion.chunk``.""" - return { - "id": self._completion_id, - "object": "chat.completion.chunk", - "created": self._created, - "model": self._model_label, - } - - def _maybe_role(self) -> dict[str, Any]: - """Return ``{"role": "assistant"}`` once, then an empty dict.""" - if not self._sent_role: - self._sent_role = True - return {"role": "assistant"} - return {} - - def _content_chunk(self, content: str) -> dict[str, Any]: - """Build a single ``delta.content`` chunk.""" - chunk = self._base() - chunk["choices"] = [ - {"index": 0, "delta": {**self._maybe_role(), "content": content}, "finish_reason": None} - ] - return chunk - - def _terminal_chunks( - self, - usage_in: dict[str, Any], - *, - error_msg: str | None = None, - ) -> list[dict[str, Any]]: - """Build the terminal finish_reason chunk and optional trailing usage chunk. - - Always emits ``finish_reason: "stop"``. ``"tool_calls"`` is never used - because by the time AGENT_COMPLETE fires every tool has already run inside the - strands loop — emitting ``"tool_calls"`` would cause clients to wait for - results that never arrive. - """ - finish_chunk = self._base() - if error_msg is not None: - finish_chunk["choices"] = [{"index": 0, "delta": {}, "finish_reason": "error"}] - finish_chunk["error"] = {"message": error_msg, "type": "agent_error"} - finish_chunk["usage"] = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} - return [finish_chunk] - - finish_chunk["choices"] = [{"index": 0, "delta": {}, "finish_reason": "stop"}] - - usage_payload: dict[str, Any] = { - "prompt_tokens": usage_in.get("input_tokens", 0), - "completion_tokens": usage_in.get("output_tokens", 0), - "total_tokens": usage_in.get("total_tokens", 0), - } - cached_tokens = usage_in.get("cache_read_input_tokens", 0) - if cached_tokens: - usage_payload["prompt_tokens_details"] = {"cached_tokens": cached_tokens} - if self._reasoning_tokens > 0: - usage_payload["completion_tokens_details"] = { - "reasoning_tokens": self._reasoning_tokens - } - - if self._emit_usage_chunk: - usage_chunk = self._base() - usage_chunk["choices"] = [] - usage_chunk["usage"] = usage_payload - return [finish_chunk, usage_chunk] - - finish_chunk["usage"] = usage_payload - return [finish_chunk] - - # ── Details-block helpers ───────────────────────────────────────────────── - - @staticmethod - def _html_attr(value: str) -> str: - """HTML-escape a string for use as an attribute value.""" - return html.escape(value, quote=True) - - def _details_closer( - self, - call_id: str, - name: str, - arguments_json: str, - result_text: str | None, - ) -> str: - """Return the completed ``
`` HTML block, replacing the opener.""" - escaped_args = self._html_attr(arguments_json) - escaped_name = self._html_attr(name) - body = html.escape(result_text) if result_text else "" - return ( - f'
\n' - f"Tool: {escaped_name}\n" - f"{body}\n" - f"
\n" - ) - - # ── Per-event handlers ──────────────────────────────────────────────────── - - def _handle_token(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """TOKEN → ``delta.content`` for entry agent; suppressed for sub-agents.""" - if not is_entry: - if self._verbosity == "narrate" and self._open_node_frames: - return [self._content_chunk(event.data.get("text", ""))] - return [] - - chunk = self._base() - chunk["choices"] = [ - { - "index": 0, - "delta": {**self._maybe_role(), "content": event.data.get("text", "")}, - "finish_reason": None, - } - ] - return [chunk] - - def _handle_reasoning(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """REASONING → reasoning delta fields for entry agent; suppressed for sub-agents.""" - if not is_entry: - return [] - - text = event.data.get("text", "") - self._reasoning_tokens += max(1, len(text) // 4) - - delta: dict[str, Any] = {**self._maybe_role()} - if self._reasoning_field_mode in ("both", "deepseek"): - delta["reasoning_content"] = text - if self._reasoning_field_mode in ("both", "openrouter"): - delta["reasoning"] = text - - # Nothing to emit when mode is "none" (only role key present or empty) - if set(delta.keys()) <= {"role"}: - return [] - - chunk = self._base() - chunk["choices"] = [{"index": 0, "delta": delta, "finish_reason": None}] - return [chunk] - - def _handle_tool_start(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """TOOL_START → bookkeeping only; no chunks are emitted. - - The completed tool call is rendered by :meth:`_handle_tool_end` as a - single ``
`` block. No native ``delta.tool_calls`` - chunk is produced because strands has already executed the tool by the - time the stream surfaces it. - """ - if not is_entry: - return [] - - tool_use_id = event.data.get("tool_use_id") or f"call_{uuid.uuid4().hex[:24]}" - tool_name = event.data.get("tool_name", "") - arguments_json = json.dumps(event.data.get("tool_input", {})) - - self._open_tool_calls[tool_use_id] = _ToolCallFrame( - call_id=tool_use_id, - name=tool_name, - arguments_json=arguments_json, - ) - return [] - - def _handle_tool_end(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """TOOL_END → ``
`` closer, or silent.""" - if not is_entry: - return [] - - frame = self._open_tool_calls.pop(event.data.get("tool_use_id", ""), None) - if self._tool_result_render == "details_block" and frame is not None: - return [ - self._content_chunk( - self._details_closer( - frame.call_id, - frame.name, - frame.arguments_json, - event.data.get("tool_result"), - ) - ) - ] - return [] - - def _handle_node_start(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """NODE_START → bookkeeping only; the completed node renders at NODE_STOP.""" - if not is_entry: - return [] - - node_id = event.data.get("node_id", "") - if node_id in self._open_node_frames: - return [] - - call_id = f"call_{uuid.uuid4().hex[:16]}" - self._open_node_frames[node_id] = _ToolCallFrame( - call_id=call_id, - name=node_id, - arguments_json="{}", - ) - return [] - - def _handle_node_stop(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """NODE_STOP → ``
`` closer for the sub-agent node.""" - if not is_entry: - return [] - - frame = self._open_node_frames.pop(event.data.get("node_id", ""), None) - if self._tool_result_render == "details_block" and frame is not None: - return [ - self._content_chunk( - self._details_closer( - frame.call_id, frame.name, frame.arguments_json, frame.result_text - ) - ) - ] - return [] - - def _handle_agent_complete(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """AGENT_COMPLETE → terminal chunks for entry agent; silent for sub-agents.""" - if not is_entry: - return [] - return self._terminal_chunks(event.data.get("usage", {})) - - def _handle_multiagent_complete( - self, event: StreamEvent, is_entry: bool - ) -> list[dict[str, Any]]: - """MULTIAGENT_COMPLETE → terminal chunks; silent for sub-orchestrations.""" - if not is_entry: - return [] - return self._terminal_chunks(event.data.get("usage", {})) - - def _handle_error(self, event: StreamEvent, is_entry: bool) -> list[dict[str, Any]]: - """ERROR → ``finish_reason: "error"`` terminal chunk.""" - return self._terminal_chunks({}, error_msg=event.data.get("message", "An error occurred")) diff --git a/src/strands_compose/converters/raw.py b/src/strands_compose/converters/raw.py deleted file mode 100644 index 40a7a5c..0000000 --- a/src/strands_compose/converters/raw.py +++ /dev/null @@ -1,31 +0,0 @@ -"""Raw pass-through StreamEvent converter.""" - -from __future__ import annotations - -from typing import Any - -from ..types import StreamEvent -from .base import StreamConverter - - -class RawStreamConverter(StreamConverter): - """Converts StreamEvents to raw JSON dicts (newline-delimited).""" - - def convert(self, event: StreamEvent) -> list[dict[str, Any]]: - """Pass through as dict. - - Args: - event: The StreamEvent to convert. - - Returns: - A single-element list containing the event's dict representation. - """ - return [event.asdict()] - - def done_marker(self) -> str: - """No terminator needed for raw streams. - - Returns: - An empty string. - """ - return "" diff --git a/src/strands_compose/tools/__init__.py b/src/strands_compose/tools/__init__.py index 33bd0ee..6dce716 100644 --- a/src/strands_compose/tools/__init__.py +++ b/src/strands_compose/tools/__init__.py @@ -2,8 +2,8 @@ Provides helpers for: - Loading ``@tool``-decorated functions from files, modules, and directories. -- Wrapping ``Agent`` / ``MultiAgentBase`` nodes as ``AgentTool`` instances - (``node_as_tool``, ``node_as_async_tool``) for delegation. +- Wrapping a ``MultiAgentBase`` as an ``AgentTool`` (``multiagent_as_tool``) for + delegation; an ``Agent`` uses ``strands.Agent.as_tool`` directly. - Serializing multi-agent results with full execution metadata. """ @@ -18,18 +18,14 @@ resolve_tool_spec, resolve_tool_specs, ) -from .wrappers import ( - node_as_async_tool, - node_as_tool, -) +from .wrappers import multiagent_as_tool __all__ = [ "load_tool_function", "load_tools_from_directory", "load_tools_from_file", "load_tools_from_module", - "node_as_async_tool", - "node_as_tool", + "multiagent_as_tool", "resolve_tool_spec", "resolve_tool_specs", "serialize_multiagent_result", diff --git a/src/strands_compose/tools/extractors.py b/src/strands_compose/tools/extractors.py index d1a7e68..dcdf60d 100644 --- a/src/strands_compose/tools/extractors.py +++ b/src/strands_compose/tools/extractors.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from collections.abc import Mapping from typing import Any from strands.agent.agent_result import AgentResult @@ -17,13 +18,49 @@ def _message_from_text(text: str) -> Message: return {"role": "assistant", "content": [{"text": text}]} +def extract_block_text(block: Mapping[str, Any]) -> str | None: + """Return a content block's text, or ``None`` when it carries none. + + Also reads ``citationsContent``, whose text sits nested in ``content[*].text``, + so a plain ``"text" in block`` check silently misses a cited answer. + + Args: + block: A single content block from a ``Message``. + + Returns: + The block's text, or ``None``. + """ + text = block.get("text") + if isinstance(text, str) and text: + return text + + citations = block.get("citationsContent") + if isinstance(citations, Mapping): + items = citations.get("content") + if isinstance(items, list): + parts = [ + item["text"] + for item in items + if isinstance(item, Mapping) and isinstance(item.get("text"), str) + ] + if parts: + # Newline-separated, matching how AgentResult.__str__ separates them. + return "\n".join(parts) + + return None + + def extract_text(message: Message | None) -> str: - """Return the last text block from a message, or an empty string.""" + """Return the last text-carrying block from a message, or an empty string.""" if not message: return "" for block in reversed(message.get("content", [])): - if isinstance(block, dict) and "text" in block: - return block["text"] + if not isinstance(block, dict): + continue + # A trailing empty block must not mask real text earlier in the message. + text = extract_block_text(block) + if text: + return text return "" @@ -41,7 +78,7 @@ def extract_last_message(result: Any) -> Message: return result.message if isinstance(result, MultiAgentResult): - last_node_id = resolve_last_node_id(result) + last_node_id = _resolve_last_node_id(result) if last_node_id and last_node_id in result.results: message = extract_last_message(result.results[last_node_id]) if message is not None: @@ -68,26 +105,20 @@ def extract_last_message(result: Any) -> Message: return _message_from_text(str(result)) -def resolve_last_node_id(result: MultiAgentResult) -> str | None: - """Determine the id of the last node that executed in a multi-agent result. +def _resolve_last_node_id(result: MultiAgentResult) -> str | None: + """Return the id of the last node that executed, or ``None``. + + Reads ``SwarmResult.node_history`` / ``GraphResult.execution_order``, since + ``results`` is keyed by insertion order and cannot answer this. Args: result: A ``MultiAgentResult`` (or subclass). Returns: - The node id string, or ``None`` if it cannot be determined. + The node id, or ``None`` if undeterminable. """ - # SwarmResult.node_history — list[SwarmNode], each has .node_id - node_history: list[Any] | None = getattr(result, "node_history", None) - if node_history: - return str(node_history[-1].node_id) - - # GraphResult.execution_order — list[GraphNode], each has .node_id - execution_order: list[Any] | None = getattr(result, "execution_order", None) - if execution_order: - return str(execution_order[-1].node_id) - - return None + history = getattr(result, "node_history", None) or getattr(result, "execution_order", None) + return str(history[-1].node_id) if history else None def serialize_multiagent_result(result: MultiAgentResult) -> dict[str, Any]: @@ -113,7 +144,7 @@ def serialize_multiagent_result(result: MultiAgentResult) -> dict[str, Any]: """ data = result.to_dict() - last_node_id = resolve_last_node_id(result) + last_node_id = _resolve_last_node_id(result) data["last_node_id"] = last_node_id final_message = extract_last_message(result) @@ -129,23 +160,13 @@ def serialize_multiagent_result(result: MultiAgentResult) -> dict[str, Any]: # GraphResult extras — execution_order, edges, node counts execution_order: list[Any] | None = getattr(result, "execution_order", None) if execution_order is not None: - edges_raw: list[Any] = getattr(result, "edges", []) or [] - entry_points_raw: list[Any] = getattr(result, "entry_points", []) or [] - - edges: list[list[str]] = [] - for edge in edges_raw: - if isinstance(edge, tuple) and len(edge) == 2: - edges.append([str(edge[0].node_id), str(edge[1].node_id)]) - else: - # GraphEdge dataclass with from_node / to_node attributes - from_id = str(getattr(getattr(edge, "from_node", None), "node_id", edge)) - to_id = str(getattr(getattr(edge, "to_node", None), "node_id", edge)) - edges.append([from_id, to_id]) - + # GraphResult.edges is list[tuple[GraphNode, GraphNode]]. + edges = [[str(a.node_id), str(b.node_id)] for a, b in getattr(result, "edges", None) or []] + entry_points = getattr(result, "entry_points", None) or [] data["graph"] = { "execution_order": [str(n.node_id) for n in execution_order], "edges": edges, - "entry_points": [str(getattr(ep, "node_id", ep)) for ep in entry_points_raw], + "entry_points": [str(getattr(ep, "node_id", ep)) for ep in entry_points], "completed_nodes": getattr(result, "completed_nodes", 0), "failed_nodes": getattr(result, "failed_nodes", 0), "interrupted_nodes": getattr(result, "interrupted_nodes", 0), diff --git a/src/strands_compose/tools/wrappers.py b/src/strands_compose/tools/wrappers.py index 4a2e6bd..31a7836 100644 --- a/src/strands_compose/tools/wrappers.py +++ b/src/strands_compose/tools/wrappers.py @@ -1,139 +1,80 @@ -"""Node wrapping utilities for delegation. +"""Wrap a ``MultiAgentBase`` as a delegate tool — strands has no ``as_tool`` for one. -Provides ``node_as_tool`` and ``node_as_async_tool`` for wrapping -``Agent`` / ``MultiAgentBase`` nodes as ``AgentTool`` instances. - -Key Features: - - Sync and async tool wrappers for Agent and MultiAgentBase nodes - - Automatic tool name resolution from agent_id or node id - - Message content preservation from single-agent and multi-agent results +An ``Agent`` needs nothing here; :meth:`strands.Agent.as_tool` covers it. """ from __future__ import annotations -from typing import TYPE_CHECKING, Any, cast +import logging +from typing import TYPE_CHECKING, Any -from strands import Agent from strands.tools.decorator import DecoratedFunctionTool, tool -from strands.types.content import Message -from .extractors import extract_last_message, extract_text +from .extractors import extract_block_text, extract_last_message if TYPE_CHECKING: - from ..types import Node - - -# ToolResultContent only accepts these 4 keys (subset of ContentBlock's 10). -# Passing model-only keys (toolUse, reasoningContent, …) would produce malformed content. -_TOOL_RESULT_CONTENT_KEYS = ("document", "image", "json", "text") - - -def _resolve_tool_name(node: Node, name: str | None) -> str: - """Resolve the tool name for a node. + from strands.multiagent.base import MultiAgentBase + from strands.types.content import Message - For ``Agent`` nodes, defaults to ``agent.agent_id``. - For ``MultiAgentBase`` nodes, defaults to ``node.id`` or ``"sub_orchestration"``. - - Args: - node: Agent or MultiAgentBase instance. - name: Explicit tool name override, or ``None`` to use the default. - - Returns: - The resolved tool name string. - """ - if name is not None: - return name - if isinstance(node, Agent): - return node.agent_id - return getattr(node, "id", "sub_orchestration") +logger = logging.getLogger(__name__) def _message_to_tool_result(message: Message) -> dict[str, Any]: - """Map a ``Message`` to a ``ToolResult`` dict (``strands.types.tools.ToolResult``). + """Map a ``Message`` to a ``ToolResult``, bypassing the decorator's stringification.""" + content = [ + {"text": text} + for block in message.get("content", []) + if isinstance(block, dict) and (text := extract_block_text(block)) + ] + return {"status": "success", "content": content or [{"text": ""}]} - Returning a pre-shaped ``{"status": ..., "content": [...]}`` dict bypasses - ``DecoratedFunctionTool._wrap_tool_result``'s plain-text auto-wrapping, so - non-text blocks (``image``, ``document``, ``json``) are preserved across - the delegation boundary. Only ``ToolResultContent`` keys are kept — model- - only blocks such as ``toolUse`` and ``reasoningContent`` are dropped. - Args: - message: The final ``Message`` returned by a sub-agent or orchestration. - - Returns: - A ``ToolResult``-shaped dict ready for the Strands decorator to pass through. - """ - content: list[dict[str, Any]] = [] - for block in message.get("content", []): - source_block = cast(dict[str, Any], block) - tool_result_block = { - key: source_block[key] for key in _TOOL_RESULT_CONTENT_KEYS if key in source_block - } - if tool_result_block: - content.append(tool_result_block) - - if content: - return {"status": "success", "content": content} - - return {"status": "success", "content": [{"text": extract_text(message)}]} - - -def node_as_tool( - node: Node, +def multiagent_as_tool( + node: MultiAgentBase, *, name: str | None = None, description: str, ) -> DecoratedFunctionTool: - """Wrap an Agent or MultiAgentBase as an ``AgentTool`` for delegation. + """Wrap a Swarm or Graph as a tool so an agent can delegate to it. - For Agent nodes, invokes the agent and returns the final message content - as a Strands tool result. For MultiAgentBase nodes (Swarm, Graph), - resolves the last executed node and returns its final message content. + An interrupt raised inside the orchestration cannot be resumed across this + boundary and is reported to the caller as a tool error. Args: - node: Agent or MultiAgentBase instance. - name: Tool name (defaults to node id). - description: Tool description for the parent LLM. + node: The orchestration to wrap. + name: Tool name (defaults to the orchestration's id). + description: Tool description for the calling LLM. Returns: - An ``AgentTool`` (``DecoratedFunctionTool``) wrapping the node. + A ``DecoratedFunctionTool`` wrapping the orchestration. """ - tool_name = _resolve_tool_name(node, name) - - @tool(name=tool_name, description=description) - def delegate(input: str) -> dict[str, Any]: - result = node(input) - return _message_to_tool_result(extract_last_message(result)) - - return delegate - - -def node_as_async_tool( - node: Node, - *, - name: str | None = None, - description: str, -) -> DecoratedFunctionTool: - """Wrap an Agent or MultiAgentBase as an async ``AgentTool`` for delegation. - - For Agent nodes, uses ``invoke_async`` for live event streaming. For - MultiAgentBase nodes, awaits ``invoke_async``. In both cases the final - message content is returned as a Strands tool result so non-text blocks - such as images and documents are preserved when possible. - - Args: - node: Agent or MultiAgentBase instance. - name: Tool name. - description: Tool description for the parent LLM. - - Returns: - An ``AgentTool`` (``DecoratedFunctionTool``) wrapping the node. - """ - tool_name = _resolve_tool_name(node, name) + tool_name = name if name is not None else getattr(node, "id", "sub_orchestration") @tool(name=tool_name, description=description) async def delegate(input: str) -> dict[str, Any]: result = await node.invoke_async(input) + + # Returning the result as-is would reach the caller as an empty success, + # making a pending approval look granted. + if interrupts := getattr(result, "interrupts", None): + names = ", ".join(str(getattr(item, "name", item)) for item in interrupts) + logger.warning( + "tool=<%s>, interrupts=<%s> | delegate orchestration interrupted, cannot resume", + tool_name, + names, + ) + return { + "status": "error", + "content": [ + { + "text": ( + f"Delegate '{tool_name}' was interrupted ({names}) and cannot be " + f"resumed across this delegation boundary." + ) + } + ], + } + return _message_to_tool_result(extract_last_message(result)) return delegate diff --git a/tests/resolve/test_delegation.py b/tests/resolve/test_delegation.py index 59cdb12..94f3326 100644 --- a/tests/resolve/test_delegation.py +++ b/tests/resolve/test_delegation.py @@ -1,4 +1,4 @@ -"""Delegation wrapping — node_as_tool / node_as_async_tool naming. +"""Delegate wiring — which adapter each connection target gets, and what is rejected. Result/message extraction (``extractors.py``) is covered in ``runtime/test_result_extraction.py``; this file stays focused on the wrapping. @@ -6,26 +6,171 @@ from __future__ import annotations +import pytest from strands import Agent +from strands import tool as strands_tool +from strands.hooks import BeforeToolCallEvent +from strands.multiagent import GraphBuilder, Swarm +from strands.session import FileSessionManager +from strands.tools.decorator import DecoratedFunctionTool +from strands.types.tools import AgentTool -from strands_compose.tools import node_as_async_tool, node_as_tool -from tests.fakes import FakeModel +from strands_compose.config.resolvers.orchestrations.builders import _delegate_tool +from strands_compose.config.schema import DelegateConnectionDef +from strands_compose.exceptions import ConfigurationError +from strands_compose.tools import multiagent_as_tool +from tests.fakes import FakeModel, ToolThenTextModel -def _agent(agent_id: str) -> Agent: - return Agent(model=FakeModel(), agent_id=agent_id) +def _agent(agent_id: str, **kwargs) -> Agent: + return Agent(model=FakeModel(), agent_id=agent_id, **kwargs) -def test_node_as_tool_defaults_name_to_agent_id(): - tool = node_as_tool(_agent("helper"), description="Delegate to helper") +def _swarm(node_id: str) -> Swarm: + return Swarm(id=node_id, nodes=[_agent("a")], entry_point=None) + + +def _conn(agent: str, **kwargs) -> DelegateConnectionDef: + return DelegateConnectionDef(agent=agent, description="do work", **kwargs) + + +async def _call(tool: AgentTool, prompt: str) -> None: + async for _ in tool.stream( + {"toolUseId": "t1", "name": tool.tool_name, "input": {"input": prompt}}, {} + ): + pass + + +# ── multiagent_as_tool ─────────────────────────────────────────────────────── + + +def test_multiagent_as_tool_defaults_name_to_the_orchestration_id(): + tool = multiagent_as_tool(_swarm("team"), description="d") + assert tool.tool_name == "team" + + +def test_multiagent_as_tool_accepts_explicit_name(): + tool = multiagent_as_tool(_swarm("team"), name="ask_team", description="d") + assert tool.tool_name == "ask_team" + + +# ── adapter choice ─────────────────────────────────────────────────────────── + + +def test_agent_connection_uses_the_strands_native_adapter(): + """An Agent must go through Agent.as_tool so interrupts can propagate and resume.""" + tool = _delegate_tool("team", _conn("helper"), _agent("helper")) + + assert tool.tool_type == "agent" + assert not isinstance(tool, DecoratedFunctionTool) + assert tool.tool_name == "helper" + + +def test_orchestration_connection_falls_back_to_the_multiagent_wrapper(): + """A Swarm has no upstream as_tool, so it keeps the hand-rolled wrapper.""" + tool = _delegate_tool("outer", _conn("team"), _swarm("team")) + + assert isinstance(tool, DecoratedFunctionTool) + assert tool.tool_name == "team" + + +def test_tool_name_tracks_the_connection_not_the_node_id(): + """The LLM sees the name the YAML used to reference the target.""" + tool = _delegate_tool("team", _conn("helper"), _agent("helper")) + assert tool.tool_name == "helper" + + +# ── preserve_context ───────────────────────────────────────────────────────── + + +async def test_preserve_context_true_accumulates_history(): + agent = _agent("helper") + tool = _delegate_tool("team", _conn("helper", preserve_context=True), agent) + + await _call(tool, "first") + await _call(tool, "second") + + assert len(agent.messages) == 4 + + +async def test_preserve_context_false_resets_between_calls(): + agent = _agent("helper") + tool = _delegate_tool("team", _conn("helper", preserve_context=False), agent) + + await _call(tool, "first") + await _call(tool, "second") + + assert len(agent.messages) == 2 + + +def test_preserve_context_defaults_to_preserving_history(): + """Compose defaults to true; strands' Agent.as_tool defaults to false.""" + assert _conn("helper").preserve_context is True + + +# ── rejected combinations ──────────────────────────────────────────────────── + + +def test_preserve_context_false_with_a_session_manager_is_rejected(tmp_path): + """strands owns this rule; its error must reach the user unwrapped.""" + agent = _agent( + "helper", + session_manager=FileSessionManager(session_id="s1", storage_dir=str(tmp_path)), + ) + + with pytest.raises(ValueError, match="cannot be used with an agent that has a session manager"): + _delegate_tool("team", _conn("helper", preserve_context=False), agent) + + +def test_preserve_context_true_allows_a_session_managed_agent(tmp_path): + agent = _agent( + "helper", + session_manager=FileSessionManager(session_id="s2", storage_dir=str(tmp_path)), + ) + + tool = _delegate_tool("team", _conn("helper", preserve_context=True), agent) + assert tool.tool_name == "helper" -def test_node_as_tool_accepts_explicit_name(): - tool = node_as_tool(_agent("helper"), name="ask_helper", description="d") - assert tool.tool_name == "ask_helper" +def test_preserve_context_false_on_an_orchestration_is_rejected(): + """A Swarm cannot be reset to a baseline, so the request must not be ignored.""" + with pytest.raises(ConfigurationError, match="no baseline to reset to"): + _delegate_tool("outer", _conn("team", preserve_context=False), _swarm("team")) + + +async def test_orchestration_interrupt_becomes_a_tool_error(): + """An interrupt inside a Swarm/Graph cannot be resumed across the boundary. + + Reporting success would make a pending approval look granted, so the caller + must see an error instead. + """ + + @strands_tool + def risky(name: str) -> str: + return f"did {name}" + + worker = Agent(model=ToolThenTextModel(tool_name="risky"), agent_id="worker", tools=[risky]) + worker.add_hook( + lambda event: event.interrupt(name="approve_risky", reason="needs sign-off"), + BeforeToolCallEvent, + ) + + builder = GraphBuilder() + builder.add_node(worker, node_id="worker") + builder.set_entry_point("worker") + graph = builder.build() + + tool = _delegate_tool("outer", _conn("pipeline"), graph) + results = [ + event + async for event in tool.stream( + {"toolUseId": "t1", "name": "pipeline", "input": {"input": "go"}}, {} + ) + ] -def test_node_as_async_tool_defaults_name_to_agent_id(): - tool = node_as_async_tool(_agent("worker"), description="d") - assert tool.tool_name == "worker" + payload = results[-1].tool_result + assert payload["status"] == "error" + assert "approve_risky" in payload["content"][0]["text"] + assert "cannot be resumed" in payload["content"][0]["text"] diff --git a/tests/runtime/test_converters.py b/tests/runtime/test_converters.py deleted file mode 100644 index 29f6c63..0000000 --- a/tests/runtime/test_converters.py +++ /dev/null @@ -1,131 +0,0 @@ -"""StreamEvent -> protocol chunk converters (OpenAI + raw pass-through). - -The OpenAI chunk shape is a real external contract, so structural assertions -here are legitimate — but we assert on shape/fields, not exact prose. -""" - -from __future__ import annotations - -from strands_compose.converters.openai import OpenAIStreamConverter -from strands_compose.converters.raw import RawStreamConverter -from strands_compose.types import EventType, StreamEvent - - -def _openai() -> OpenAIStreamConverter: - return OpenAIStreamConverter(entry_agent_name="entry") - - -def test_entry_token_becomes_openai_content_delta(): - chunks = _openai().convert( - StreamEvent(type=EventType.TOKEN, agent_name="entry", data={"text": "hi"}) - ) - assert chunks[0]["object"] == "chat.completion.chunk" - assert chunks[0]["choices"][0]["delta"]["content"] == "hi" - - -def test_sub_agent_token_is_suppressed_in_compact_mode(): - chunks = _openai().convert( - StreamEvent(type=EventType.TOKEN, agent_name="worker", data={"text": "x"}) - ) - assert chunks == [] - - -def test_agent_complete_emits_stop_with_usage(): - event = StreamEvent( - type=EventType.AGENT_COMPLETE, - agent_name="entry", - data={"usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}}, - ) - chunks = _openai().convert(event) - finish = chunks[-1] - assert finish["choices"][0]["finish_reason"] == "stop" - assert finish["usage"]["total_tokens"] == 15 - - -def test_error_event_emits_error_finish_reason(): - chunks = _openai().convert( - StreamEvent(type=EventType.ERROR, agent_name="entry", data={"message": "boom"}) - ) - assert chunks[0]["choices"][0]["finish_reason"] == "error" - - -def test_openai_done_marker_is_openai_sentinel(): - assert _openai().done_marker() == "data: [DONE]\n\n" - - -def test_raw_converter_passes_event_through_as_dict(): - event = StreamEvent(type=EventType.TOKEN, agent_name="a", data={"text": "hi"}) - chunks = RawStreamConverter().convert(event) - assert chunks == [event.asdict()] - - -def test_raw_converter_has_no_done_marker(): - assert RawStreamConverter().done_marker() == "" - - -def test_reasoning_populates_both_reasoning_fields_in_both_mode(): - event = StreamEvent(type=EventType.REASONING, agent_name="entry", data={"text": "thinking"}) - delta = _openai().convert(event)[0]["choices"][0]["delta"] - assert delta["reasoning_content"] == "thinking" - assert delta["reasoning"] == "thinking" - - -def test_tool_start_then_end_renders_a_details_block(): - conv = _openai() - conv.convert( - StreamEvent( - type=EventType.TOOL_START, - agent_name="entry", - data={"tool_use_id": "t1", "tool_name": "search", "tool_input": {"q": "x"}}, - ) - ) - chunks = conv.convert( - StreamEvent( - type=EventType.TOOL_END, - agent_name="entry", - data={"tool_use_id": "t1", "tool_result": "found it"}, - ) - ) - content = chunks[0]["choices"][0]["delta"]["content"] - assert "search" in content - assert "found it" in content - - -def test_node_start_then_stop_renders_a_details_block(): - conv = _openai() - conv.convert( - StreamEvent(type=EventType.NODE_START, agent_name="entry", data={"node_id": "researcher"}) - ) - chunks = conv.convert( - StreamEvent(type=EventType.NODE_STOP, agent_name="entry", data={"node_id": "researcher"}) - ) - assert "researcher" in chunks[0]["choices"][0]["delta"]["content"] - - -def test_multiagent_complete_emits_terminal_stop(): - event = StreamEvent(type=EventType.MULTIAGENT_COMPLETE, agent_name="entry", data={"usage": {}}) - chunks = _openai().convert(event) - assert chunks[-1]["choices"][0]["finish_reason"] == "stop" - - -def test_usage_chunk_mode_emits_separate_trailing_usage_chunk(): - conv = OpenAIStreamConverter(entry_agent_name="entry", emit_usage_chunk=True) - event = StreamEvent( - type=EventType.AGENT_COMPLETE, - agent_name="entry", - data={"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}}, - ) - chunks = conv.convert(event) - assert chunks[-1]["choices"] == [] - assert chunks[-1]["usage"]["total_tokens"] == 2 - - -def test_reset_clears_stream_state(): - conv = _openai() - conv.convert(StreamEvent(type=EventType.TOKEN, agent_name="entry", data={"text": "hi"})) - conv.reset() - # After reset the role prelude is sent again on the next content chunk. - delta = conv.convert( - StreamEvent(type=EventType.TOKEN, agent_name="entry", data={"text": "again"}) - )[0]["choices"][0]["delta"] - assert delta.get("role") == "assistant" diff --git a/uv.lock b/uv.lock index 4da57c6..4b1e9c8 100644 --- a/uv.lock +++ b/uv.lock @@ -2062,11 +2062,11 @@ requires-dist = [ { name = "mcp", specifier = ">=1.24.0" }, { name = "pydantic", specifier = ">=2.12.5" }, { name = "pyyaml", specifier = ">=6.0.0" }, - { name = "strands-agents", specifier = ">=1.48.0,<2.0.0" }, - { name = "strands-agents", extras = ["anthropic"], marker = "extra == 'anthropic'", specifier = ">=1.48.0,<2.0.0" }, - { name = "strands-agents", extras = ["gemini"], marker = "extra == 'gemini'", specifier = ">=1.48.0,<2.0.0" }, - { name = "strands-agents", extras = ["ollama"], marker = "extra == 'ollama'", specifier = ">=1.48.0,<2.0.0" }, - { name = "strands-agents", extras = ["openai"], marker = "extra == 'openai'", specifier = ">=1.48.0,<2.0.0" }, + { name = "strands-agents", specifier = ">=1.52.0,<2.0.0" }, + { name = "strands-agents", extras = ["anthropic"], marker = "extra == 'anthropic'", specifier = ">=1.52.0,<2.0.0" }, + { name = "strands-agents", extras = ["gemini"], marker = "extra == 'gemini'", specifier = ">=1.52.0,<2.0.0" }, + { name = "strands-agents", extras = ["ollama"], marker = "extra == 'ollama'", specifier = ">=1.52.0,<2.0.0" }, + { name = "strands-agents", extras = ["openai"], marker = "extra == 'openai'", specifier = ">=1.52.0,<2.0.0" }, ] provides-extras = ["agentcore-memory", "ollama", "openai", "gemini", "anthropic"] @@ -2074,17 +2074,17 @@ provides-extras = ["agentcore-memory", "ollama", "openai", "gemini", "anthropic" dev = [ { name = "bandit", specifier = ">=1.9.2" }, { name = "commitizen", specifier = ">=4.8.4" }, - { name = "coverage", specifier = ">=7.12.0" }, + { name = "coverage", specifier = ">=7.15.4" }, { name = "hypothesis", specifier = ">=6.155.7" }, { name = "pre-commit", specifier = ">=4.3.0" }, - { name = "pytest", specifier = ">=9.0.2" }, + { name = "pytest", specifier = ">=9.1.1" }, { name = "pytest-asyncio", specifier = ">=1.2.0" }, { name = "pytest-cov", specifier = ">=7.0.0" }, { name = "pytest-mock", specifier = ">=3.15.1" }, { name = "pytest-xdist", specifier = ">=3.8.0" }, - { name = "ruff", specifier = ">=0.14.8" }, - { name = "rust-just", specifier = ">=1.42.4" }, - { name = "ty", specifier = ">=0.0.29" }, + { name = "ruff", specifier = ">=0.16.3" }, + { name = "rust-just", specifier = ">=1.58.0" }, + { name = "ty", specifier = ">=0.0.72" }, ] [[package]]