feat: add AgentAsTool

src/strands/__init__.py

@@ -1,6 +1,7 @@
 """A framework for building, deploying, and managing AI agents."""
 
 from . import agent, models, telemetry, types
+from .agent._agent_as_tool import AgentAsTool
 from .agent.agent import Agent
 from .agent.base import AgentBase
 from .event_loop._retry import ModelRetryStrategy
@@ -11,6 +12,7 @@
 
 __all__ = [
     "Agent",
+    "AgentAsTool",
     "AgentBase",
     "AgentSkills",
     "agent",

src/strands/agent/__init__.py

@@ -10,6 +10,7 @@
 from typing import Any
 
 from ..event_loop._retry import ModelRetryStrategy
+from ._agent_as_tool import AgentAsTool
 from .agent import Agent
 from .agent_result import AgentResult
 from .base import AgentBase
@@ -24,6 +25,7 @@
     "Agent",
     "AgentBase",
     "AgentResult",
+    "AgentAsTool",
     "ConversationManager",
     "NullConversationManager",
     "SlidingWindowConversationManager",

src/strands/agent/_agent_as_tool.py

@@ -0,0 +1,237 @@
+"""Agent-as-tool adapter.
+
+This module provides the AgentAsTool class that wraps an Agent (or any AgentBase) as a tool
+so it can be passed to another agent's tool list.
+"""
+
+import copy
+import logging
+from typing import Any
+
+from typing_extensions import override
+
+from ..agent.state import AgentState
+from ..types._events import AgentAsToolStreamEvent, ToolResultEvent
+from ..types.content import Messages
+from ..types.tools import AgentTool, ToolGenerator, ToolSpec, ToolUse
+from .base import AgentBase
+
+logger = logging.getLogger(__name__)
+
+
+class AgentAsTool(AgentTool):
+    """Adapter that exposes an Agent as a tool for use by other agents.
+
+    The tool accepts a single ``input`` string parameter, invokes the wrapped
+    agent, and returns the text response.
+
+    Example:
+        ```python
+        from strands import Agent
+        from strands.agent import AgentAsTool
+
+        researcher = Agent(name="researcher", description="Finds information")
+
+        # Use directly
+        tool = AgentAsTool(researcher, name="researcher", description="Finds information")
+
+        # Or via convenience method
+        tool = researcher.as_tool()
+
+        # Start each invocation with a fresh conversation
+        tool = researcher.as_tool(preserve_context=False)
+
+        writer = Agent(name="writer", tools=[tool])
+        writer("Write about AI agents")
+        ```
+    """
+
+    def __init__(
+        self,
+        agent: AgentBase,
+        *,
+        name: str,
+        description: str,
+        preserve_context: bool = False,
+    ) -> None:
+        r"""Initialize the agent-as-tool adapter.
+
+        Args:
+            agent: The agent to wrap as a tool.
+            name: Tool name. Must match the pattern ``[a-zA-Z0-9_\\-]{1,64}``.
+            description: Tool description.
+            preserve_context: Whether to preserve the agent's conversation history across
+                invocations. When False, the agent's messages and state are reset to the
+                values they had at construction time before each call, ensuring every
+                invocation starts from the same baseline regardless of any external
+                interactions with the agent. Defaults to False. Only effective when the
+                wrapped agent exposes a mutable ``messages`` list and/or an ``AgentState``
+                (e.g. ``strands.agent.Agent``).
+        """
+        super().__init__()
+        self._agent = agent
+        self._tool_name = name
+        self._description = description
+        self._preserve_context = preserve_context
+
+        # When preserve_context=False, we snapshot the agent's initial state so we can
+        # restore it before each invocation. This mirrors GraphNode.reset_executor_state().
+        # We require an Agent instance for this since AgentBase doesn't guarantee
+        # messages/state attributes.
+        self._initial_messages: Messages = []
+        self._initial_state: AgentState = AgentState()
+
+        if not preserve_context:
+            from .agent import Agent
+
+            if not isinstance(agent, Agent):
+                raise TypeError(f"preserve_context=False requires an Agent instance, got {type(agent).__name__}")
+            self._initial_messages = copy.deepcopy(agent.messages)
+            self._initial_state = AgentState(agent.state.get())
+
+    @property
+    def agent(self) -> AgentBase:
+        """The wrapped agent instance."""
+        return self._agent
+
+    @property
+    def tool_name(self) -> str:
+        """Get the tool name."""
+        return self._tool_name
+
+    @property
+    def tool_spec(self) -> ToolSpec:
+        """Get the tool specification."""
+        return {
+            "name": self._tool_name,
+            "description": self._description,
+            "inputSchema": {
+                "json": {
+                    "type": "object",
+                    "properties": {
+                        "input": {
+                            "type": "string",
+                            "description": "The input to send to the agent tool.",
+                        },
+                    },
+                    "required": ["input"],
+                }
+            },
+        }
+
+    @property
+    def tool_type(self) -> str:
+        """Get the tool type."""
+        return "agent"
+
+    @override
+    async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
+        """Invoke the wrapped agent via streaming and yield events.
+
+        Intermediate agent events are wrapped in AgentAsToolStreamEvent so the caller
+        can distinguish sub-agent progress from regular tool events. The final
+        AgentResult is yielded as a ToolResultEvent.
+
+        Args:
+            tool_use: The tool use request containing the input parameter.
+            invocation_state: Context for the tool invocation.
+            **kwargs: Additional keyword arguments.
+
+        Yields:
+            AgentAsToolStreamEvent for intermediate events, then ToolResultEvent with the final response.
+        """
+        tool_input = tool_use["input"]
+        if isinstance(tool_input, dict):
+            prompt = tool_input.get("input", "")
+        elif isinstance(tool_input, str):
+            prompt = tool_input
+        else:
+            logger.warning("tool_name=<%s> | unexpected input type: %s", self._tool_name, type(tool_input))
+            prompt = str(tool_input)
+
+        tool_use_id = tool_use["toolUseId"]
+
+        if not self._preserve_context:
+            self._reset_agent_state(tool_use_id)
+
+        logger.debug("tool_name=<%s>, tool_use_id=<%s> | invoking agent", self._tool_name, tool_use_id)
+
+        try:
+            result = None
+            async for event in self._agent.stream_async(prompt):
+                if "result" in event:
+                    result = event["result"]
+                else:
+                    yield AgentAsToolStreamEvent(tool_use, event, self)
+
+            if result is None:
+                yield ToolResultEvent(
+                    {
+                        "toolUseId": tool_use_id,
+                        "status": "error",
+                        "content": [{"text": "Agent did not produce a result"}],
+                    }
+                )
+                return
+
+            if result.structured_output:
+                yield ToolResultEvent(
+                    {
+                        "toolUseId": tool_use_id,
+                        "status": "success",
+                        "content": [{"json": result.structured_output.model_dump()}],
+                    }
+                )
+            else:
+                yield ToolResultEvent(
+                    {
+                        "toolUseId": tool_use_id,
+                        "status": "success",
+                        "content": [{"text": str(result)}],
+                    }
+                )
+
+        except Exception as e:
+            logger.warning(
+                "tool_name=<%s>, tool_use_id=<%s> | agent invocation failed: %s",
+                self._tool_name,
+                tool_use_id,
+                e,
+            )
+            yield ToolResultEvent(
+                {
+                    "toolUseId": tool_use_id,
+                    "status": "error",
+                    "content": [{"text": f"Agent error: {e}"}],
+                }
+            )
+
+    def _reset_agent_state(self, tool_use_id: str) -> None:
+        """Reset the wrapped agent to its initial state.
+
+        Restores messages and state to the values captured at construction time.
+        This mirrors the pattern used by ``GraphNode.reset_executor_state()``.
+
+        Args:
+            tool_use_id: Tool use ID for logging context.
+        """
+        from .agent import Agent
+
+        # isinstance narrows the type for mypy; __init__ guarantees this when preserve_context=False
+        if not isinstance(self._agent, Agent):
+            return
+
+        logger.debug(
+            "tool_name=<%s>, tool_use_id=<%s> | resetting agent to initial state",
+            self._tool_name,
+            tool_use_id,
+        )
+        self._agent.messages = copy.deepcopy(self._initial_messages)
+        self._agent.state = AgentState(self._initial_state.get())
+
+    @override
+    def get_display_properties(self) -> dict[str, str]:
+        """Get properties for UI display."""
+        properties = super().get_display_properties()
+        properties["Agent"] = getattr(self._agent, "name", "unknown")
+        return properties

src/strands/agent/agent.py

@@ -62,6 +62,7 @@
 from ..types.content import ContentBlock, Message, Messages, SystemContentBlock
 from ..types.exceptions import ConcurrencyException, ContextWindowOverflowException
 from ..types.traces import AttributeValue
+from ._agent_as_tool import AgentAsTool
 from .agent_result import AgentResult
 from .base import AgentBase
 from .conversation_manager import (
@@ -612,6 +613,40 @@ async def structured_output_async(self, output_model: type[T], prompt: AgentInpu
             finally:
                 await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self, invocation_state={}))
 
+    def as_tool(
+        self,
+        name: str | None = None,
+        description: str | None = None,
+        preserve_context: bool = False,
+    ) -> AgentAsTool:
+        r"""Convert this agent into a tool for use by another agent.
+
+        Args:
+            name: Tool name. Must match the pattern ``[a-zA-Z0-9_\\-]{1,64}``.
+                Defaults to the agent's name.
+            description: Tool description. Defaults to the agent's description.
+            preserve_context: Whether to preserve the agent's conversation history across
+                invocations. When False, the agent's messages and state are reset to the
+                values they had at construction time before each call, ensuring every
+                invocation starts from the same baseline regardless of any external
+                interactions with the agent. Defaults to False.
+
+        Returns:
+            An AgentAsTool wrapping this agent.
+
+        Example:
+            ```python
+            researcher = Agent(name="researcher", description="Finds information")
+            writer = Agent(name="writer", tools=[researcher.as_tool()])
+            writer("Write about AI agents")
+            ```
+        """
+        if not name:
+            name = self.name
+        if not description:
+            description = self.description or f"Use the {name} agent as a tool by providing a natural language input"
+        return AgentAsTool(self, name=name, description=description, preserve_context=preserve_context)
+
     def cleanup(self) -> None:
         """Clean up resources used by the agent.
 

src/strands/types/_events.py

@@ -21,6 +21,7 @@
 
 if TYPE_CHECKING:
     from ..agent import AgentResult
+    from ..agent._agent_as_tool import AgentAsTool
     from ..multiagent.base import MultiAgentResult, NodeResult
 
 
@@ -323,6 +324,31 @@ def tool_use_id(self) -> str:
         return cast(ToolUse, cast(dict, self.get("tool_stream_event")).get("tool_use"))["toolUseId"]
 
 
+class AgentAsToolStreamEvent(ToolStreamEvent):
+    """Event emitted when an agent-as-tool yields intermediate events during execution.
+
+    Extends ToolStreamEvent with a reference to the originating AgentAsTool so callers
+    can distinguish sub-agent stream events from regular tool stream events and access
+    the wrapped agent, tool name, description, etc.
+    """
+
+    def __init__(self, tool_use: ToolUse, tool_stream_data: Any, agent_as_tool: "AgentAsTool") -> None:
+        """Initialize with tool streaming data and agent-tool reference.
+
+        Args:
+            tool_use: The tool invocation producing the stream.
+            tool_stream_data: The yielded event from the sub-agent execution.
+            agent_as_tool: The AgentAsTool instance that produced this event.
+        """
+        super().__init__(tool_use, tool_stream_data)
+        self._agent_as_tool = agent_as_tool
+
+    @property
+    def agent_as_tool(self) -> "AgentAsTool":
+        """The AgentAsTool instance that produced this event."""
+        return self._agent_as_tool
+
+
 class ToolCancelEvent(TypedEvent):
     """Event emitted when a user cancels a tool call from their BeforeToolCallEvent hook."""