Related Pages

Related topics: System Architecture, Agent System

Getting Started with Strands Agents

Strands Agents is a model-driven SDK for building and running AI agents in Python. The framework provides a simple yet powerful abstraction that handles the agent loop, tool execution, and model interactions, allowing developers to focus on application logic rather than infrastructure.

Overview

Strands Agents is designed to be lightweight, flexible, and model-agnostic. It supports multiple model providers including Amazon Bedrock, Anthropic, Google Gemini, LiteLLM, Llama, Ollama, OpenAI, and Writer. The SDK natively integrates with the Model Context Protocol (MCP), enabling access to thousands of pre-built tools and services.

Key capabilities include:

• Python-based tool creation using decorators
• MCP server integration for external tools
• Multi-agent orchestration support
• Structured output using Pydantic models
• Built-in telemetry and metrics
• Streaming support
• Conversation management with context window handling

Source: README.md

Installation

Requirements

• Python 3.10 or higher
• pip or another Python package manager

Package Installation

# Install the core Strands Agents SDK
pip install strands-agents

# Install the tools package (includes built-in tools)
pip install strands-agents-tools

Source: README.md

Setting Up a Virtual Environment

# Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows use: .venv\Scripts\activate

# Install Strands packages
pip install strands-agents strands-agents-tools

Source: README.md

Quick Start

Your First Agent

The simplest way to create an agent is to use the built-in @tool decorator and pass tools to the Agent constructor:

from strands import Agent, tool

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression.
    
    Args:
        expression: A mathematical expression to evaluate (e.g., "2 + 2")
    """
    return str(eval(expression))

# Create an agent with the calculator tool
agent = Agent(tools=[calculator])

# Run the agent
response = agent("What is 15 * 23?")
print(response)

Source: strands-py/src/strands/agent/agent.py

Using Built-in Tools

The strands-agents-tools package provides pre-built tools that can be imported directly:

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])
response = agent("Calculate the square root of 1764")

Source: README.md

Core Concepts

The Agent

The Agent is the central component of the Strands SDK. It orchestrates the interaction between the user, the model, and tools.

from strands import Agent

agent = Agent(
    model=None,           # Uses default BedrockModel if None
    tools=[],             # List of available tools
    system_prompt=None,   # Optional system instructions
    **kwargs              # Additional configuration
)

Source: strands-py/src/strands/agent/agent.py

Tool Execution

Tools in Strands are Python functions decorated with @tool. The docstring is used by the LLM to understand the tool's purpose and generate appropriate calls.

from strands import Agent, tool

@tool
def word_count(text: str) -> int:
    """Count words in text.
    
    This docstring is used by the LLM to understand the tool's purpose.
    """
    return len(text.split())

agent = Agent(tools=[word_count])

Source: README.md

Hot Reloading Tools from Directory

For development, you can enable automatic tool loading from a directory:

from strands import Agent

# Agent will watch ./tools/ directory for changes
agent = Agent(load_tools_from_directory=True)
response = agent("Use any tools you find in the tools directory")

Source: README.md

Model Configuration

Default Model (Amazon Bedrock)

By default, Strands uses the Amazon Bedrock model provider. For this to work, you need:

• AWS credentials configured
• Model access enabled (e.g., Claude 4 Sonnet in us-west-2)

Using Amazon Bedrock Explicitly

from strands import Agent
from strands.models import BedrockModel

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    temperature=0.3,
    streaming=True,  # Enable/disable streaming
)
agent = Agent(model=bedrock_model)
agent("Tell me about Agentic AI")

Source: README.md

Using Google Gemini

from strands import Agent
from strands.models.gemini import GeminiModel

gemini_model = GeminiModel(
    client_args={
        "api_key": "your_gemini_api_key",
    },
    model_id="gemini-2.5-flash",
    params={"temperature": 0.7}
)
agent = Agent(model=gemini_model)
agent("Tell me about Agentic AI")

Source: README.md

Using Ollama

from strands import Agent
from strands.models.ollama import OllamaModel

ollama_model = OllamaModel(
    model_id="llama3.2",
    base_url="http://localhost:11434"
)
agent = Agent(model=ollama_model)

Available Model Providers

Source: README.md

Model Context Protocol (MCP) Integration

Strands provides native support for MCP servers, enabling integration with thousands of external tools and services.

Basic MCP Client Usage

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# Create an MCP client for the AWS Documentation MCP server
aws_docs_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(
            command="uvx",
            args=["awslabs.aws-documentation-mcp-server@latest"]
        )
    )
)

with aws_docs_client:
    agent = Agent(tools=aws_docs_client.list_tools_sync())
    response = agent("Tell me about Amazon Bedrock")

Source: strands-py/src/strands/tools/mcp/mcp_client.py

MCP Task Configuration

For long-running tool calls, you can configure task-based execution with polling and TTL settings:

from strands.tools.mcp import MCPClient
from strands.tools.mcp.mcp_tasks import TasksConfig, DEFAULT_TASK_CONFIG

# Configure task settings
tasks_config: TasksConfig = {
    "ttl": timedelta(minutes=2),        # Task time-to-live
    "poll_timeout": timedelta(minutes=10)  # Polling timeout
}

mcp_client = MCPClient(
    server_factory=...,
    tasks_config=tasks_config
)

Source: strands-py/src/strands/tools/mcp/mcp_tasks.py

Source: strands-py/src/strands/tools/mcp/mcp_tasks.py

Structured Output

Strands supports structured output using Pydantic models, ensuring responses conform to a specific schema.

from strands import Agent
from pydantic import BaseModel

class WeatherResponse(BaseModel):
    temperature: float
    unit: str
    condition: str

agent = Agent(
    model=...,
    tools=[...],
    structured_output_model=WeatherResponse
)

result = agent("What's the weather like in Seattle?")
# result will be a WeatherResponse instance

Source: strands-py/src/strands/tools/structured_output/__init__.py

Tool Executors

Strands provides different execution strategies for tools:

Concurrent Execution

Tools are executed in parallel when possible:

from strands.tools.executors import ConcurrentToolExecutor

agent = Agent(
    tools=[tool1, tool2],
    tool_executor=ConcurrentToolExecutor()
)

Source: strands-py/src/strands/tools/executors/__init__.py

Sequential Execution

Tools are executed one at a time:

from strands.tools.executors import SequentialToolExecutor

agent = Agent(
    tools=[tool1, tool2],
    tool_executor=SequentialToolExecutor()
)

Source: strands-py/src/strands/tools/executors/__init__.py

Telemetry and Observability

Strands includes built-in telemetry support using OpenTelemetry for tracing and metrics.

Configuration

from strands.telemetry import StrandsTelemetry

# Quick setup with console exporter
StrandsTelemetry().setup_console_exporter()

# Setup with OTLP exporter
StrandsTelemetry().setup_otlp_exporter()

# Chain multiple exporters
StrandsTelemetry().setup_console_exporter().setup_otlp_exporter()

# Setup meter provider
StrandsTelemetry().setup_meter(enable_console_exporter=True, enable_otlp_exporter=True)

Source: strands-py/src/strands/telemetry/config.py

Available Metrics

Source: strands-py/src/strands/telemetry/metrics_constants.py

Environment Variables

Source: strands-py/src/strands/telemetry/config.py

Agent Loop Architecture

The agent operates in a continuous loop that processes user input, calls the model, executes tools as needed, and returns results.

graph TD
    A[User Input] --> B[Agent Loop]
    B --> C[Model Inference]
    C --> D{Stop Reason?}
    D -->|tool_use| E[Execute Tool]
    D -->|end_turn| F[Return Result]
    E --> B
    F --> G[Response]
    
    H[Context Window] --> B
    B --> I[Conversation Manager]
    I --> H

Stop Reasons

Source: strands-py/src/strands/agent/agent.py

Skills Integration

Strands supports the AgentSkills.io integration for progressive disclosure of instructions:

from strands import Agent
from strands.vended_plugins.skills import Skill, AgentSkills

# Load skill from filesystem
skill = Skill.from_file("./skills/pdf-processing")

# Load all skills from a directory
skills = Skill.from_directory("./skills/")

# Create plugin and attach to agent
plugin = AgentSkills(skills=["./skills/pdf-processing"])
agent = Agent(plugins=[plugin])

Source: strands-py/src/strands/vended_plugins/skills/__init__.py

Known Limitations and Issues

Thread Safety

The MetricsClient singleton is not thread-safe in version 1.41.0. If using multi-threaded applications, be aware of potential race conditions when accessing metrics. See Issue #2342.

Gemini Model Safety Settings

GeminiModel may crash with a TypeError when safety settings block content and usage_metadata is None. See Issue #2347.

MCP Progress Updates

Strands does not currently support MCP progress updates for long-running tasks, which is available in other MCP implementations like FastMCP. See Feature Request #1812.

Python Wheel Package

When publishing wheels, ensure that README.md and LICENSE files are properly included. The current setup uses a relative path that may not work with all Python build backends. See Issue #2351.

Development Setup

For contributing to the Strands SDK:

# Navigate to the strands-py subdirectory
cd strands-py

# Install hatch for development
pip install hatch

# Run unit tests
hatch test

# Format and lint code
hatch fmt

Source: README.md

See Also

• Agent Loop Documentation
• Structured Output Guide
• MCP Integration Guide
• Telemetry and Observability
• Production Deployment Guide
• API Reference
• Contributing Guide

Related Pages

Related topics: Agent System, Model Providers, Tool System

System Architecture

The Strands Agents SDK is a model-driven framework for building and running AI agents. The architecture follows a simple yet powerful agent loop design that scales from basic conversational assistants to complex autonomous multi-agent workflows. This document provides a comprehensive overview of the system's architectural components, their interactions, and design patterns.

Overview

The SDK is organized as a monorepo (consolidating from multiple repositories as noted in Issue #2286) with the core implementation located in the strands-py/ subdirectory. The architecture emphasizes modularity, allowing users to customize model providers, tool execution strategies, conversation management, and agent behavior through a plugin system. Source: README.md

High-Level Architecture

graph TD
    subgraph "User Application Layer"
        A[User Code] --> B[Agent]
    end

    subgraph "Agent Core"
        B --> C[Agent Loop]
        C --> D[Model Client]
        C --> E[Tool Executor]
        C --> F[Session Manager]
        C --> G[Event Loop Handler]
    end

    subgraph "Model Providers"
        D --> H[BedrockModel]
        D --> I[GeminiModel]
        D --> J[OllamaModel]
        D --> K[OpenAIModel]
        D --> L[Custom Provider]
    end

    subgraph "Tool System"
        E --> M[Built-in Tools]
        E --> N[MCP Client]
        N --> O[MCP Servers]
        E --> P[Plugin Tools]
    end

    subgraph "Session & Context"
        F --> Q[Conversation History]
        G --> R[Structured Output]
        G --> S[Telemetry/Metrics]
    end

    subgraph "Multi-Agent"
        B --> T[MultiAgentPlugin]
        T --> U[Swarm Orchestrator]
        T --> V[Graph Orchestrator]
        T --> W[A2A Communication]
    end

Core Components

Agent

The Agent class is the central entry point for building AI agents. It orchestrates the interaction between model providers, tools, and session management through an event-driven loop. Source: strands-py/src/strands/agent/agent.py

The Agent class provides:

• Model integration: Configurable model providers (Bedrock, Gemini, OpenAI, Ollama, and more)
• Tool management: Dynamic loading and execution of tools
• Session persistence: Conversation history management across interactions
• Plugin architecture: Extensible system for adding custom behaviors
• Streaming support: Real-time response streaming capabilities

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])
response = agent("What is the square root of 1764?")

Model Abstraction Layer

The model layer provides a unified interface for interacting with various LLM providers. This abstraction allows the SDK to be model-agnostic while supporting provider-specific features. Source: strands-py/src/strands/models/model.py

Tool System Architecture

The tool system enables agents to interact with external capabilities. Tools are registered with the agent and invoked through a configurable executor. Source: strands-py/src/strands/tools/tools.py

graph LR
    A[Agent Loop] --> B[Tool Executor]
    B --> C[ConcurrentToolExecutor]
    B --> D[SequentialToolExecutor]
    C --> E[Thread Pool]
    D --> F[Single Thread]
    E --> G[Tool 1]
    E --> H[Tool 2]
    E --> I[Tool N]

#### Tool Executors

The SDK provides built-in tool execution strategies. Source: strands-py/src/strands/tools/executors/__init__.py

Model Context Protocol (MCP) Integration

The MCP client provides native support for the Model Context Protocol, enabling agents to connect to MCP servers and access pre-built tools. Source: strands-py/src/strands/tools/mcp/mcp_client.py

sequenceDiagram
    participant Agent
    participant MCPClient
    participant MCPSession
    participant MCPServer

    Agent->>MCPClient: list_tools_sync()
    MCPClient->>MCPSession: list_tools()
    MCPSession->>MCPServer: ListToolsRequest
    MCPServer-->>MCPSession: ListToolsResult
    MCPSession-->>MCPClient: tools[]
    MCPClient-->>Agent: MCPAgentTool[]

    Agent->>MCPClient: call_tool(name, args)
    MCPClient->>MCPSession: call_tool()
    MCPSession->>MCPServer: CallToolRequest
    MCPServer-->>MCPSession: CallToolResult
    MCPSession-->>MCPClient: MCPToolResult
    MCPClient-->>Agent: Tool Result

#### MCP Task-Augmented Execution

For long-running tool operations, the SDK supports task-augmented execution as defined in the MCP specification (2025-11-25). This feature uses a three-phase approach: create task, poll for completion, and retrieve results. Source: strands-py/src/strands/tools/mcp/mcp_tasks.py

Note: This is an experimental feature. The MCP specification and Strands implementation are subject to change. Source: strands-py/src/strands/tools/mcp/mcp_tasks.py:14-24

graph TD
    A[Tool Call Request] --> B{Server Supports Tasks?}
    B -->|Yes| C[call_tool_as_task]
    B -->|No| D[call_tool]
    C --> E[Create Task]
    E --> F[Poll for Status]
    F --> G{Terminal Status?}
    G -->|No| F
    G -->|Yes| H[get_task_result]
    H --> I[Return Result]
    D --> J[Execute Directly]
    J --> I

Agent Loop Architecture

The agent loop is the core execution engine that orchestrates model interactions, tool calls, and session management. Source: strands-py/src/strands/event_loop/event_loop.py

graph TD
    A[User Input] --> B[Start Cycle]
    B --> C[Call Model API]
    C --> D{Response Type?}
    D -->|Tool Use| E[Handle Tool Execution]
    E --> F[Execute Tools]
    F --> G[Collect Results]
    G --> H[Append to History]
    H --> B
    D -->|End Turn| I[Return Response]
    D -->|Max Tokens| J[Raise MaxTokensReachedException]

Event Loop Lifecycle

The event loop manages the cycle of:

1. Start cycle: Initialize metrics and tracing for the cycle
2. Model invocation: Call the configured model with current conversation history
3. Handle response: Process the model's response (tool call, text response, or error)
4. Tool execution: Execute requested tools and collect results
5. End cycle: Update metrics and append results to conversation history

Source: strands-py/src/strands/telemetry/metrics_constants.py

Structured Output

The SDK supports structured output through a tool-based mechanism. When enabled, the model is forced to invoke a structured output tool to extract typed responses. Source: strands-py/src/strands/tools/structured_output/__init__.py

Session Management

The session manager handles conversation history, message persistence, and context management. It provides strategies for managing context windows as conversations grow. Source: strands-py/src/strands/session/session_manager.py

Conversation Managers

Recent releases (v1.40.0+) introduced proactive context compression for conversation managers. This feature helps manage context window limits by intelligently compressing conversation history when needed. Source: strands-py/src/strands/session/session_manager.py

Plugin Architecture

The plugin system provides a powerful mechanism for extending agent capabilities. Plugins can contribute tools, modify agent behavior, and integrate with external systems. Source: designs/0001-plugins.md

graph TD
    A[Agent] --> B[Plugin System]
    B --> C[MultiAgentPlugin]
    B --> D[Custom Plugins]
    C --> E[Swarm Orchestrator]
    C --> F[Graph Orchestrator]
    D --> G[Tool Plugins]
    D --> H[Hook Plugins]

Available Plugins

#### AgentSkills Plugin

The AgentSkills plugin integrates with AgentSkills.io, enabling progressive disclosure of instructions. Metadata is injected into the system prompt upfront, and full instructions are loaded on demand via tools. Source: strands-py/src/strands/vended_plugins/skills/__init__.py

#### MultiAgentPlugin

Added in v1.41.0, the MultiAgentPlugin supports Swarm and Graph orchestrators for coordinating multiple agents. Source: strands-py/src/strands/multiagent/a2a/_converters.py

Multi-Agent Systems

The SDK supports building multi-agent systems through the Agent-to-Agent (A2A) protocol. This enables complex workflows where specialized agents collaborate to complete tasks. Source: strands-py/src/strands/multiagent/a2a/_converters.py

graph LR
    A[Orchestrator Agent] -->|A2A| B[Specialist Agent 1]
    A -->|A2A| C[Specialist Agent 2]
    A -->|A2A| D[Specialist Agent N]
    
    B -->|Task Complete| A
    C -->|Task Complete| A
    D -->|Task Complete| A

Task State Mapping

The SDK maps A2A task lifecycle states to appropriate Strands stop reasons:

Source: strands-py/src/strands/multiagent/a2a/_converters.py:72-82

Telemetry and Metrics

The SDK emits structured metrics for observability, including cycle counts, tool execution metrics, token usage, and latency measurements. Source: strands-py/src/strands/telemetry/metrics_constants.py

Known Issue: The MetricsClient singleton is not thread-safe (see Issue #2342). This is a known limitation in v1.41.0.

Context Management

Context management strategies define how conversation history is maintained and when compression occurs. The SDK provides configurable strategies for different use cases. Source: designs/0003-context-management.md

Proactive Context Compression

Introduced in v1.40.0, proactive context compression helps prevent context window exhaustion by compressing conversation history before limits are reached. The system includes a fallback trim point for tool-heavy conversations. Source: strands-py/src/strands/session/session_manager.py

Common Architecture Patterns

Tool Invocation Flow

sequenceDiagram
    participant AgentLoop
    participant ToolExecutor
    participant Tool
    
    AgentLoop->>ToolExecutor: execute_tools(tool_requests)
    ToolExecutor->>Tool: call(tool_name, arguments)
    Tool-->>ToolExecutor: ToolResult
    ToolExecutor-->>AgentLoop: ToolEvent

Error Handling

The architecture follows consistent error handling patterns:

• Tool errors are captured and returned as ToolResult with error status
• The agent loop can be configured to fail on first error or collect all results
• Max token errors raise MaxTokensReachedException for unrecoverable states

Directory Structure

strands-py/src/strands/
├── agent/           # Agent core implementation
├── models/         # Model provider implementations
├── tools/          # Tool system and executors
│   ├── mcp/        # MCP client and task support
│   ├── executors/  # Tool execution strategies
│   └── structured_output/  # Structured output support
├── session/        # Session and conversation management
├── event_loop/     # Agent event loop
├── telemetry/      # Metrics and observability
├── multiagent/     # Multi-agent system support
│   └── a2a/        # Agent-to-agent protocol
└── vended_plugins/ # Built-in plugins (skills, etc.)

See Also

• Agent Loop Documentation
• MCP Integration Guide
• Plugin Development Guide
• Quick Start Guide
• Production Deployment Guide
• API Reference

Related Pages

Related topics: Model Providers, Tool System, Session & Conversation Management

Agent System

The Agent System is the core component of Strands Agents, providing a model-driven approach to building and running AI agents. From simple conversational assistants to complex autonomous workflows, the Agent System delivers a lightweight yet powerful framework that scales from local development to production deployment.

Source: strands-py/src/strands/agent/base.py:1-50

Overview

The Agent System implements a model-driven agent loop that processes inputs, calls tools when necessary, and generates responses. It is designed to be:

• Lightweight & Flexible: A simple agent loop that just works and is fully customizable
• Model Agnostic: Supports multiple model providers including Amazon Bedrock, Anthropic, Gemini, LiteLLM, Llama, Ollama, OpenAI, and Writer
• Advanced Capable: Supports multi-agent systems, autonomous agents, and streaming
• Tool-Enabled: Native MCP support for accessing thousands of pre-built tools

Source: README.md

Architecture

The Agent System follows a layered architecture with the agent loop at its core, interacting with tools, conversation managers, and model providers.

graph TD
    A[User Input] --> B[Agent]
    B --> C[Model Provider]
    C --> D{Stop Reason?}
    D -->|tool_use| E[Tool Executor]
    E --> B
    D -->|end_turn| F[Response]
    B --> G[Conversation Manager]
    B --> H[Telemetry]
    B --> I[Hooks]

Source: strands-py/src/strands/event_loop/event_loop.py:1-100

Event Loop

The event loop is the core execution engine of the Agent System. It handles:

1. Initializing cycle state and metrics - Setting up per-turn context
2. Checking execution limits - Verifying max_tokens and other constraints
3. Processing messages with the model - Sending requests to the model provider
4. Handling tool execution requests - Managing tool calls and results
5. Managing recursive calls - Supporting multi-turn tool interactions
6. Collecting and reporting metrics - Tracking performance via telemetry
7. Error handling and recovery - Managing exceptions gracefully

Source: strands-py/src/strands/event_loop/event_loop.py:100-200

#### Cycle Flow

sequenceDiagram
    participant User
    participant Agent
    participant EventLoop
    participant Model
    participant ToolExecutor
    
    User->>Agent: invoke(input)
    Agent->>EventLoop: event_loop_cycle()
    EventLoop->>Model: process_message()
    Model-->>EventLoop: stop_reason = tool_use
    EventLoop->>ToolExecutor: handle_tool_execution()
    ToolExecutor->>Model: execute_tool()
    Model-->>EventLoop: tool_result
    EventLoop->>EventLoop: recurse (next cycle)
    EventLoop-->>Agent: final_response
    Agent-->>User: result

Tool System

Tools extend the agent's capabilities beyond text generation. The Agent System supports:

Source: strands-py/src/strands/agent/agent.py:100-150

Core Components

Agent Class

The Agent class is the primary entry point for creating AI agents.

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])
response = agent("What is the square root of 1764?")

Source: strands-py/src/strands/agent/agent.py:1-100

#### Agent Parameters

Source: strands-py/src/strands/agent/agent.py:50-100

State Management

The Agent maintains state across multiple invocations, tracking conversation history and execution context.

classDiagram
    class AgentState {
        +messages: list[Message]
        +request_state: dict
        +invocation_state: InvocationState
    }
    
    class InvocationState {
        +event_loop_cycle_count: int
        +total_tokens: int
        +last_stop_reason: StopReason
    }

Source: strands-py/src/strands/agent/state.py:1-50

Agent Result

The AgentResult class encapsulates the response from an agent invocation.

from strands.agent.agent_result import AgentResult

result: AgentResult = agent("Hello")
print(result.message)        # The response message
print(result.conversation)  # Full conversation history
print(result.metrics)       # Event loop metrics

Source: strands-py/src/strands/agent/agent_result.py:1-50

#### AgentResult Properties

Source: strands-py/src/strands/agent/agent_result.py:50-100

Agent as Tool

Agents can be used as tools by other agents, enabling multi-agent workflows. This is implemented via the _AgentAsTool wrapper class.

graph LR
    A[Parent Agent] -->|calls tool| B[Agent as Tool]
    B --> C[Child Agent]
    C -->|invokes| D[Model]
    D -->|response| C
    C -->|result| B
    B -->|tool result| A

Source: strands-py/src/strands/agent/_agent_as_tool.py:1-50

from strands import Agent

child_agent = Agent(model=small_model, tools=[...])
parent_agent = Agent(
    model=large_model,
    tools=[child_agent.as_tool(name="analyzer", description="...")]
)

Source: strands-py/src/strands/agent/_agent_as_tool.py:100-150

Hooks System

The hooks system provides a way to customize agent behavior at key points during execution.

Available Hooks

Source: strands-py/src/strands/hooks/events.py:1-50

Hook Callback Signature

Hooks can be passed as callable callbacks directly in the Agent constructor:

from strands import Agent

def my_hook(event, context):
    print(f"Event: {event}")
    print(f"Context: {context}")

agent = Agent(
    model=model,
    hooks=my_hook,
    tools=[...]
)

Source: strands-py/src/strands/hooks/registry.py:1-50

Tool Executors

Tool executors define how tools are executed within the agent loop.

Available Executors

Source: strands-py/src/strands/tools/executors/__init__.py:1-50

Configuration

from strands import Agent
from strands.tools.executors import ConcurrentToolExecutor

agent = Agent(
    model=model,
    tools=independent_tools,
    tool_executor=ConcurrentToolExecutor()
)

Conversation Management

The conversation manager controls how conversation history is maintained and managed.

Default Manager

By default, Strands uses SlidingWindowConversationManager which maintains a sliding window of the most recent messages.

from strands import Agent
from strands.types.conversation import SlidingWindowConversationManager

agent = Agent(
    model=model,
    conversation_manager=SlidingWindowConversationManager(
        max_messages=20  # Keep last 20 messages
    )
)

Source: strands-py/src/strands/types/conversation.py

Proactive Context Compression

The v1.40.0 release added proactive context compression to conversation managers, which automatically manages context window usage when approaching limits.

Source: Release v1.40.0

Structured Output

Agents can return structured data conforming to a Pydantic model.

from strands import Agent
from pydantic import BaseModel

class WeatherResponse(BaseModel):
    temperature: float
    unit: str
    conditions: str

agent = Agent(
    model=model,
    structured_output_model=WeatherResponse
)

result = agent("What's the weather like?")
# result is a WeatherResponse instance

Source: strands-py/src/strands/tools/structured_output/__init__.py:1-50

Telemetry

The Agent System integrates with OpenTelemetry for observability.

Metrics

Source: strands-py/src/strands/telemetry/metrics_constants.py:1-50

Configuration

from strands.telemetry.config import StrandsTelemetry

# Setup with console exporter
StrandsTelemetry().setup_console_exporter()

# Setup with OTLP exporter
StrandsTelemetry().setup_otlp_exporter()

Source: strands-py/src/strands/telemetry/config.py:1-50

Common Patterns

Basic Agent

from strands import Agent

agent = Agent()
response = agent("Hello, world!")
print(response)

Tool-Using Agent

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])
response = agent("What is 15 * 23?")

Multi-Modal Agent

from strands import Agent
from strands.types.content import image_url, text

agent = Agent()
response = agent([
    image_url(url="https://example.com/image.jpg"),
    text("What's in this image?")
])

Streaming Agent

from strands import Agent
from strands.models import BedrockModel

model = BedrockModel(streaming=True)
agent = Agent(model=model)

for event in agent.stream("Tell me a story"):
    print(event, end="", flush=True)

Troubleshooting

MetricsClient Thread Safety

The MetricsClient singleton is not thread-safe in versions prior to the fix. If using multi-threaded workloads, ensure you are on the latest version.

Source: Issue #2342

GeminiModel TypeError

When using safety_settings with GeminiModel, ensure you handle cases where usage_metadata may be None. This is fixed in recent versions.

Source: Issue #2347

Max Tokens Reached

When the agent reaches the max_tokens limit, it fails with a MaxTokensReachedException. Configure appropriate token limits in your model settings.

Source: strands-py/src/strands/event_loop/event_loop.py:200-250

See Also

• Agent Loop Documentation
• Quick Start Guide
• MCP Integration
• Tools System
• Model Providers
• Multi-Agent Systems
• API Reference

Related Pages

Related topics: Agent System, System Architecture

Model Providers

Model Providers are the abstraction layer in Strands Agents that enables the framework to work with various LLM backends through a unified interface. Each model provider implements the Model interface and handles the specific API requirements, authentication mechanisms, request/response formatting, and feature support for its respective service.

Overview

Strands Agents is model agnostic, meaning you can swap between different LLM providers without changing your agent code. The SDK ships with native support for multiple providers, from cloud-hosted services like Amazon Bedrock and Google Gemini to local deployments via Ollama or LlamaCPP.

graph TD
    A[Agent] --> B[Model Interface]
    B --> C[BedrockModel]
    B --> D[AnthropicModel]
    B --> E[GeminiModel]
    B --> F[OpenAIModel]
    B --> G[OllamaModel]
    B --> H[LiteLLMModel]
    B --> I[MistralModel]
    B --> J[SageMakerModel]
    B --> K[WriterModel]
    B --> L[LlamaCPPModel]
    B --> M[LlamaAPIModel]

Key Capabilities

Source: strands-py/src/strands/models/model.py

Related Pages

Related topics: Agent System, MCP Integration

Tool System

The Strands Agents Tool System provides a flexible, extensible mechanism for defining, managing, and executing tools that AI agents can invoke during their reasoning loops. Tools enable agents to interact with external systems, perform computations, access data, and execute actions beyond the model's native capabilities.

Overview

The Tool System consists of several interconnected components that work together to provide a seamless tool execution experience:

graph TD
    A[Agent] -->|invokes| B[Tool Registry]
    B -->|selects| C[Tool Executor]
    C -->|executes| D[Python Tools]
    C -->|executes| E[MCP Tools]
    D -->|returns| F[Tool Result]
    E -->|returns| F
    G[Tool Decorator] -->|defines| D
    H[MCPClient] -->|exposes| E

Key capabilities:

• Define tools using Python decorators with automatic Pydantic schema generation
• Execute tools sequentially or concurrently based on agent decisions
• Integrate with Model Context Protocol (MCP) servers for extended tooling
• Support hot-reloading of tools from directories for development
• Structured output generation using tool-based constraints

Tool Definition

The `@tool` Decorator

The primary mechanism for defining tools in Strands is the @tool decorator. This decorator transforms a Python function into a tool that the agent can invoke. Source: decorator.py:1-50

from strands import Agent, tool

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression.
    
    Args:
        expression: A mathematical expression to evaluate.
    """
    return str(eval(expression))

agent = Agent(tools=[calculator])

Docstring Parsing

The decorator automatically extracts tool metadata from the function's docstring. The description is derived from the docstring content, excluding the Args: section. Source: decorator.py:100-150

Docstring format:

@tool
def word_count(text: str) -> int:
    """Count words in text.

    This description is used by the LLM to understand the tool's purpose.
    
    Args:
        text: The text to count words in.
    
    Returns:
        The number of words in the text.
    """
    return len(text.split())

The @tool decorator extracts the description by parsing the docstring and excluding everything from Args: onwards. This ensures the LLM receives only the purpose description without parameter details.

Pydantic Schema Generation

Tool parameter schemas are automatically generated from Python type hints using Pydantic. The decorator creates an input model that validates and serializes tool arguments. Source: decorator.py:150-200

@tool
def search_documents(
    query: str,
    limit: int = 10,
    filters: dict | None = None
) -> list[dict]:
    """Search documents matching a query."""
    ...

The generated schema includes:

• Parameter names and types
• Default values (optional parameters)
• Type constraints (min/max for numbers, patterns for strings)
• Required vs optional classification

Tool Specification

ToolSpec Data Model

Tools are represented internally by a ToolSpec dictionary containing metadata that describes the tool's interface. Source: tools.py

Input Schema Structure

{
  "name": "calculator",
  "description": "Evaluate a mathematical expression",
  "inputSchema": {
    "json": {
      "type": "object",
      "properties": {
        "expression": {
          "type": "string",
          "description": "A mathematical expression to evaluate"
        }
      },
      "required": ["expression"]
    }
  }
}

Tool Execution

Execution Strategies

Strands provides two built-in execution strategies for handling tool calls. Source: executors/__init__.py:1-20

#### Sequential Tool Executor

Tools are executed one at a time in the order determined by the agent. This is the default behavior. Source: executors/sequential.py

from strands.tools.executors import SequentialToolExecutor

executor = SequentialToolExecutor()
agent = Agent(tool_executor=executor)

#### Concurrent Tool Executor

Tools are executed in parallel when the agent decides they are independent. Source: executors/concurrent.py

from strands.tools.executors import ConcurrentToolExecutor

executor = ConcurrentToolExecutor(max_workers=4)
agent = Agent(tool_executor=executor)

Execution Flow

sequenceDiagram
    participant Agent
    participant Executor
    participant Tool1
    participant Tool2
    
    Agent->>Executor: execute_tools([tool_calls])
    
    alt Sequential Execution
        Executor->>Tool1: call(args)
        Tool1-->>Executor: result
        Executor->>Tool2: call(args)
        Tool2-->>Executor: result
    else Concurrent Execution
        Executor->>Tool1: call(args)
        Executor->>Tool2: call(args)
        Tool1-->>Executor: result
        Tool2-->>Executor: result
    end
    
    Executor-->>Agent: [results]

MCP Integration

Model Context Protocol Overview

The MCP (Model Context Protocol) integration enables Strands agents to use tools from MCP servers. MCP provides a standardized way to expose tools from various sources including local processes, HTTP servers, and cloud services. Source: mcp_client.py:1-100

MCPClient Configuration

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# Connect to an MCP server
mcp_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(
            command="uvx",
            args=["awslabs.aws-documentation-mcp-server@latest"]
        )
    )
)

with mcp_client:
    agent = Agent(tools=mcp_client.list_tools_sync())
    response = agent("Tell me about Amazon Bedrock")

MCP Task-Augmented Execution

For long-running tools, MCP supports task-augmented execution that enables polling for completion and retrieving results asynchronously. Source: mcp_tasks.py:1-40

Task execution flow:

graph TD
    A[Tool Call] --> B{Server Supports Tasks?}
    B -->|Yes| C[call_tool_as_task]
    B -->|No| D[call_tool Direct]
    C --> E[Get Task ID]
    E --> F[poll_task]
    F --> G{Terminal Status?}
    G -->|No| F
    G -->|Yes| H[get_task_result]
    H --> I[Return Result]
    D --> J[Return Result]

Note: MCP task progress updates are not yet fully implemented in the current release. See GitHub Issue #1812 for tracking progress on this feature.

Structured Output

Tool-Based Structured Output

Strands supports structured output generation by constraining the model to use a dedicated tool that returns structured data. This ensures type-safe, predictable responses for data-intensive tasks. Source: structured_output/__init__.py:1-20

from pydantic import BaseModel
from strands import Agent
from strands_tools import calculator  # Must provide structured output

class MathResult(BaseModel):
    expression: str
    result: float
    steps: list[str]

agent = Agent(
    tools=[calculator],
    structured_output=MathResult
)

Converting Pydantic Models to Tools

Use convert_pydantic_to_tool_spec to create a tool specification from a Pydantic model:

from strands.tools.structured_output import convert_pydantic_to_tool_spec
from pydantic import BaseModel

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str

tool_spec = convert_pydantic_to_tool_spec(SearchResult)

Tool Watching and Hot Reloading

Directory-Based Tool Loading

Strands supports automatic tool discovery and hot-reloading from a directory. This is particularly useful during development. Source: watcher.py:1-50

from strands import Agent

# Agent watches ./tools/ directory for changes
agent = Agent(load_tools_from_directory=True)
response = agent("Use any tools you find in the tools directory")

Watch Behavior

The tool watcher monitors the specified directory for:

• New .py files containing @tool decorated functions
• Modifications to existing tool files
• Deletions of tool files

Changes are automatically reflected without restarting the agent.

Telemetry and Metrics

The Tool System emits telemetry for monitoring and observability:

Source: metrics_constants.py:1-25

Common Patterns

Building Multi-Step Workflows

from strands import Agent, tool

@tool
def fetch_data(source: str) -> dict:
    """Fetch data from a source."""
    ...

@tool
def transform_data(data: dict, format: str) -> str:
    """Transform data to specified format."""
    ...

@tool
def save_result(data: str, destination: str) -> bool:
    """Save result to destination."""
    ...

agent = Agent(tools=[fetch_data, transform_data, save_result])
result = agent("Fetch user data, transform to JSON, and save to /tmp/data.json")

Combining MCP and Python Tools

from strands import Agent
from strands.tools.mcp import MCPClient
from strands_tools import calculator

# MCP server for database access
db_client = MCPClient(
    lambda: stdio_client(StdioServerParameters(...))
)

# Python tool for calculations
with db_client:
    agent = Agent(tools=[calculator, *db_client.list_tools_sync()])

Troubleshooting

Tool Not Found

If an agent fails to find a tool:

1. Verify the tool is in the agent's tool list
2. Check that the tool name matches exactly (case-sensitive)
3. Ensure the tool's description is clear and descriptive

Tool Execution Errors

Common causes:

• Type validation failures: Arguments don't match the expected schema
• Missing required parameters: Required arguments not provided
• Timeout errors: Tool exceeded the configured timeout
• Import errors: Tool dependencies not available

MCP Connection Issues

For MCP server connection problems:

1. Verify the MCP server command and arguments are correct
2. Check that the MCP server package is installed
3. Ensure proper authentication for cloud-based MCP servers
4. Review server logs for specific error messages

See Also

• Agent Loop - Understanding how agents use tools
• MCP Server - Building custom MCP servers
• Agent Skills - Progressive instruction loading
• Telemetry Configuration - Setting up observability

Related Pages

Related topics: Agent System, Plugin System

Multi-Agent Patterns

Multi-agent patterns in Strands Agents enable sophisticated AI workflows where multiple specialized agents collaborate to solve complex tasks. Rather than relying on a single agent with all capabilities, multi-agent systems allow you to decompose problems into smaller, focused subtasks handled by specialized agents that can share context, delegate work, and coordinate their activities.

Overview

The Strands Agents SDK provides two primary orchestration patterns for multi-agent systems:

Both patterns are accessible through the MultiAgentPlugin, which integrates seamlessly with the standard Agent API. Additionally, the SDK supports the Agent-to-Agent (A2A) protocol for distributed multi-agent systems where agents may run in separate processes or on different machines.

Source: strands-py/src/strands/plugins/multiagent_plugin.py:1-50

Architecture

Multi-Agent System Components

graph TB
    subgraph "Multi-Agent System"
        A["User Request"] --> B["Orchestrator"]
        B --> C1["Agent Node 1"]
        B --> C2["Agent Node 2"]
        B --> C3["Agent Node N"]
        C1 <--> D["Shared Context"]
        C2 <--> D
        C3 <--> D
        C1 --> E["Response"]
        C2 --> E
        C3 --> E
    end
    
    subgraph "Plugin Integration"
        F["MultiAgentPlugin"] --> B
    end

Orchestrator Hierarchy

The multi-agent architecture follows a base class pattern:

classDiagram
    class BaseMultiAgent {
        +agents: dict[str, Agent]
        +results: dict[str, Any]
        +run(request: str) MultiAgentResult
    }
    
    class GraphOrchestrator {
        +nodes: list~Node~
        +edges: list~Edge~
        +execute(request: str) MultiAgentResult
    }
    
    class SwarmOrchestrator {
        +run(request: str) MultiAgentResult
    }
    
    BaseMultiAgent <|-- GraphOrchestrator
    BaseMultiAgent <|-- SwarmOrchestrator

Source: strands-py/src/strands/multiagent/base.py:1-100

Graph Orchestrator

The Graph Orchestrator manages multi-agent execution using a directed graph structure where:

• Nodes represent individual agents with specific roles
• Edges define execution order and data flow between agents
• Execution follows topological ordering based on dependencies

Node Configuration

Source: strands-py/src/strands/multiagent/graph.py:1-150

Execution Flow

graph LR
    A["User Request"] --> B["Graph Orchestrator"]
    B --> C["Build Execution Plan"]
    C --> D{"Dependencies Ready?"}
    D -->|Yes| E["Execute Node"]
    D -->|No| F["Wait for Dependencies"]
    F --> D
    E --> G{"More Nodes?"}
    G -->|Yes| D
    G -->|No| H["Aggregate Results"]
    H --> I["Response"]

Usage Example

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent.graph import GraphOrchestrator, Node, Edge

# Create specialized agents
researcher = Agent(
    model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
    system_prompt="You are a research specialist. Find and summarize relevant information."
)

writer = Agent(
    model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
    system_prompt="You are a technical writer. Create clear documentation from research."
)

# Define graph structure
orchestrator = GraphOrchestrator(
    nodes=[
        Node(agent=researcher, name="research", description="Research task"),
        Node(agent=writer, name="writing", description="Documentation task"),
    ],
    edges=[
        Edge(source="research", target="writing"),
    ]
)

# Execute workflow
result = orchestrator.run("Explain quantum computing")

Source: strands-py/src/strands/multiagent/graph.py:100-200

Swarm Orchestrator

The Swarm Orchestrator implements a dynamic collaboration pattern where agents interact fluidly, passing control between each other based on task requirements. Unlike the structured graph approach, swarm execution allows for more open-ended problem solving where the path through agents is determined at runtime.

Collaboration Model

graph TD
    A["User Request"] --> B["Coordinator Agent"]
    B --> C["Specialist 1"]
    C -->|Handoff| D["Specialist 2"]
    D -->|Handoff| E["Specialist 3"]
    E -->|Complete| F["Response"]
    
    G["Context Sharing"] -.-> C
    G -.-> D
    G -.-> E

Handoff Mechanism

The swarm pattern relies on explicit handoffs between agents. When one agent determines another specialist should handle a task, it returns a structured handoff request that the orchestrator routes to the appropriate agent.

Source: strands-py/src/strands/multiagent/swarm.py:1-100

Usage Example

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent.swarm import SwarmOrchestrator

# Create swarm agents
coordinator = Agent(
    model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
    system_prompt="You coordinate complex tasks and delegate to specialists."
)

researcher = Agent(
    model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
    system_prompt="You research topics thoroughly and hand off to writers when done."
)

# Create orchestrator
orchestrator = SwarmOrchestrator(
    agents={
        "coordinator": coordinator,
        "researcher": researcher,
    }
)

# Execute with dynamic agent routing
result = orchestrator.run("Research and summarize recent AI developments")

MultiAgentPlugin Integration

The MultiAgentPlugin provides a unified interface for integrating both Graph and Swarm orchestrators into the standard Agent workflow.

Plugin Architecture

graph TB
    A["Agent with Plugin"] --> B["MultiAgentPlugin"]
    B --> C{"Orchestrator Type?"}
    C -->|Graph| D["GraphOrchestrator"]
    C -->|Swarm| E["SwarmOrchestrator"]
    D --> F["Node Execution"]
    E --> G["Agent Handoffs"]
    F --> H["MultiAgentResult"]
    G --> H

Source: strands-py/src/strands/plugins/multiagent_plugin.py:1-80

Plugin Configuration

Usage with Agent

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent.graph import GraphOrchestrator, Node
from strands.plugins.multiagent_plugin import MultiAgentPlugin

# Define agents
agent_a = Agent(model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"))
agent_b = Agent(model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"))

# Create orchestrator
orchestrator = GraphOrchestrator(
    nodes=[
        Node(agent=agent_a, name="first"),
        Node(agent=agent_b, name="second"),
    ]
)

# Wrap in plugin
plugin = MultiAgentPlugin(orchestrator=orchestrator)

# Use with standard Agent
agent = Agent(plugins=[plugin])
result = agent("Complex multi-step task")

Agent-to-Agent (A2A) Protocol

The A2A protocol enables communication between agents running in separate processes or on different machines. Strands Agents provides an A2A server implementation that adapts the standard Agent to the A2A protocol.

A2A Architecture

graph TB
    subgraph "A2A Network"
        A["A2A Client 1"] <-->|"A2A Protocol"| B["A2A Server"]
        C["A2A Client 2"] <-->|"A2A Protocol"| B
        D["A2A Client N"] <-->|"A2A Protocol"| B
    end
    
    B --> E["Strands Agent"]
    B --> F["Task Store"]
    B --> G["Event Queue"]

Server Configuration

Source: strands-py/src/strands/multiagent/a2a/server.py:20-60

A2A Server Usage

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent.a2a.server import A2AServer

# Create and wrap agent
agent = Agent(
    model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
    system_prompt="You are a helpful assistant."
)

# Create A2A server
server = A2AServer(
    agent=agent,
    host="0.0.0.0",
    port=9000,
    skills=[
        {"id": "assistant", "name": "General Assistant", "description": "General help"}
    ]
)

# Start server
server.start()

Task State Management

The A2A protocol maintains task state throughout execution:

Source: strands-py/src/strands/multiagent/a2a/_converters.py:1-50

Event System

Multi-agent systems emit events for observability and debugging. The event system provides hooks for monitoring agent activities, task state changes, and handoffs.

Available Events

Source: strands-py/src/strands/experimental/hooks/multiagent/events.py:1-50

Choosing Between Patterns

Error Handling

Retry Policies

Both orchestrators support configurable retry policies for failed agent invocations:

from strands.multiagent.graph import Node, RetryPolicy

policy = RetryPolicy(
    max_attempts=3,
    initial_delay=1.0,
    max_delay=60.0,
    exponential_base=2.0
)

node = Node(
    agent=my_agent,
    name="robust_task",
    retry_policy=policy
)

Error Propagation

Errors in multi-agent systems can be handled at multiple levels:

1. Node level: Per-agent retry policies
2. Graph level: on_error configuration to define fallback behavior
3. Orchestrator level: Global error handlers

Source: strands-py/src/strands/multiagent/graph.py:150-200

Best Practices

1. Define Clear Agent Roles: Each agent should have a well-defined responsibility to avoid ambiguity in handoffs.

1. Use Appropriate Patterns:

• Use Graph when the workflow is predictable and agents have clear dependencies
• Use Swarm when tasks are exploratory and the path through agents is determined at runtime
• Use A2A when agents run in separate processes or need to scale independently

1. Implement Proper Error Handling: Configure retry policies and error handlers for robust multi-agent systems.

1. Monitor Agent Handoffs: Track handoff patterns to optimize agent collaboration.

1. Consider Context Management: Shared context between agents can improve efficiency but may increase complexity.

See Also

• Agent Loop Documentation
• Tool Usage Patterns
• MCP Integration
• Structured Output
• Agent Builder
• A2A Protocol Specification

Related Pages

Related topics: Agent System, Model Providers

Bidirectional Streaming (Experimental)

⚠️ Experimental Feature: Bidirectional streaming is currently in experimental status. APIs may change in future releases as we refine the feature based on user feedback and evolving model capabilities.

Bidirectional Streaming enables real-time voice and audio conversations with persistent streaming connections in the Strands Agents SDK. Unlike traditional request-response patterns where each interaction is independent, bidirectional streaming maintains long-running conversations where users can interrupt, provide continuous input, and receive real-time audio responses.

Overview

The bidirectional streaming feature provides a BidiAgent that maintains a persistent connection with supported model providers. This enables:

• Real-time audio conversations: Speak directly to the agent and receive spoken responses
• User interruptions: Stop the model mid-generation to redirect the conversation
• Continuous input: Send text or audio at any point during the conversation
• Tool execution: Use tools within persistent streaming sessions
• Multiple I/O modalities: Support for both audio and text inputs/outputs

Source: strands-py/src/strands/experimental/bidi/models/model.py:1-47

Supported Model Providers

The following model providers support bidirectional streaming:

Each provider implements the BidiModel protocol and handles provider-specific connection protocols while exposing a standardized event-based API.

Source: README.md

Architecture

High-Level Component Architecture

graph TB
    subgraph "Application Layer"
        App[Application Code]
    end
    
    subgraph "Strands Bidi Layer"
        BidiAgent[BidiAgent]
        BidiLoop[BidiAgentLoop]
    end
    
    subgraph "I/O Layer"
        BidiAudioIO[BidiAudioIO]
        BidiTextIO[BidiTextIO]
    end
    
    subgraph "Model Provider Layer"
        BidiNovaSonic[BidiNovaSonicModel]
        BidiGemini[BidiGeminiLiveModel]
        BidiOpenAI[BidiOpenAIRealtimeModel]
    end
    
    subgraph "External Services"
        NovaSonic[Amazon Nova Sonic]
        GeminiLive[Google Gemini Live]
        OpenAIRealtime[OpenAI Realtime API]
    end
    
    App --> BidiAgent
    BidiAgent --> BidiLoop
    BidiLoop --> BidiAudioIO
    BidiLoop --> BidiTextIO
    BidiLoop --> BidiNovaSonic
    BidiLoop --> BidiGemini
    BidiLoop --> BidiOpenAI
    BidiNovaSonic --> NovaSonic
    BidiGemini --> GeminiLive
    BidiOpenAI --> OpenAIRealtime

BidiAgent Loop Flow

sequenceDiagram
    participant App as Application
    participant BidiAgent as BidiAgent
    participant BidiLoop as BidiAgentLoop
    participant Model as BidiModel
    participant Provider as Model Provider

    App->>BidiAgent: run(inputs, outputs)
    BidiAgent->>BidiLoop: __init__(model, inputs, outputs)
    BidiAgent->>BidiLoop: run()
    BidiLoop->>Model: start(system_prompt, tools, messages)
    Model->>Provider: Establish connection
    Provider-->>Model: Connection established
    Model-->>BidiLoop: Connected

    loop Conversation Loop
        BidiLoop->>Model: receive()
        Provider-->>Model: BidiOutputEvent
        Model-->>BidiLoop: BidiOutputEvent
        
        alt Audio Output
            BidiLoop->>BidiLoop.outputs: audio_io.output().write(audio)
        end
        
        alt Tool Call
            BidiLoop->>BidiLoop: execute tool
            BidiLoop->>Model: send(ToolResultEvent)
        end
        
        alt User Input
            BidiLoop.outputs.input()-->BidiLoop: BidiInputEvent
            BidiLoop->>Model: send(input_event)
        end
    end

    alt Stop Conversation
        App->>BidiAgent: signal stop
        BidiAgent->>BidiLoop: cancel()
        BidiLoop->>Model: stop connection
    end

Source: strands-py/src/strands/experimental/bidi/agent/loop.py

Installation

Package Extras

Bidirectional streaming requires additional dependencies beyond the core Strands Agents package:

# Server-side only (no audio I/O dependencies)
pip install strands-agents[bidi]

# With audio I/O support (includes PyAudio dependency)
pip install strands-agents[bidi,bidi-io]

The bidi extra includes the core bidirectional streaming components. The additional bidi-io extra installs audio processing dependencies required for microphone input and speaker output.

Source: README.md

Core Components

BidiModel Protocol

The BidiModel protocol defines the contract for all bidirectional streaming models:

from strands.experimental.bidi.models.model import BidiModel

class BidiModel(Protocol):
    """Protocol for bidirectional streaming models."""
    
    config: dict[str, Any]
    
    async def start(
        self,
        system_prompt: str | None = None,
        tools: list[ToolSpec] | None = None,
        messages: Messages | None = None,
        **kwargs: Any,
    ) -> None:
        """Establish a persistent streaming connection."""
        ...
    
    async def receive(self) -> AsyncIterable[BidiOutputEvent]:
        """Receive events from the model."""
        ...
    
    async def send(self, content: BidiInputEvent | ToolResultEvent) -> None:
        """Send content to the model."""
        ...

Key methods:

• start(): Establishes the persistent connection before any send/receive operations
• receive(): Returns an async iterable that yields output events until the connection closes
• send(): Transmits user input, audio, images, or tool results during an active session

Source: strands-py/src/strands/experimental/bidi/models/model.py:48-101

BidiModelTimeoutError

Bidirectional models are configured with connection time limits (e.g., Nova Sonic keeps connections open for 8 minutes maximum). When a timeout occurs, the agent loop is designed to restart the model connection seamlessly:

from strands.experimental.bidi.models.model import BidiModelTimeoutError

try:
    async for event in model.receive():
        # Process events
        pass
except BidiModelTimeoutError:
    # Handle timeout - agent loop may auto-reconnect
    pass

Source: strands-py/src/strands/experimental/bidi/models/model.py:113-120

Event Types

#### Input Events

Input events are sent from the client to the model:

Source: strands-py/src/strands/experimental/bidi/types/events.py

#### Output Events

Output events are received from the model and include:

• Audio output (streamed audio data)
• Text transcripts (user and model speech-to-text)
• Tool calls (requests to execute tools)
• Control signals (session state changes)

I/O Components

#### BidiAudioIO

Audio I/O component for microphone input and speaker output:

from strands.experimental.bidi.io import BidiAudioIO

audio_io = BidiAudioIO(
    input_sample_rate=16000,    # Microphone sample rate
    output_sample_rate=24000,   # Speaker sample rate
    input_channels=1,
    output_channels=1,
)

Source: strands-py/src/strands/experimental/bidi/io/audio.py

#### BidiTextIO

Text I/O component for terminal/console-based interactions:

from strands.experimental.bidi.io import BidiTextIO

text_io = BidiTextIO()

Source: strands-py/src/strands/experimental/bidi/io/text.py

BidiAgent

The main agent class for bidirectional streaming:

from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models import BidiNovaSonicModel
from strands.experimental.bidi.io import BidiAudioIO, BidiTextIO
from strands.experimental.bidi.tools import stop_conversation
from strands_tools import calculator

# Configure model
model = BidiNovaSonicModel(
    model_id="us.amazon.nova-pro-v1:0",
    session_duration_minutes=5,
)

# Configure I/O
audio_io = BidiAudioIO()
text_io = BidiTextIO()

# Create agent with tools
agent = BidiAgent(
    model=model,
    tools=[calculator, stop_conversation],
)

# Run with audio and text I/O
await agent.run(
    inputs=[audio_io.input(), text_io.input()],  # Speak OR type
    outputs=[audio_io.output(), text_io.output()]
)

Source: strands-py/src/strands/experimental/bidi/agent/agent.py

Built-in Tools

#### stop_conversation

A built-in tool to gracefully stop the bidirectional streaming session:

from strands.experimental.bidi.tools import stop_conversation

# Available in all BidiAgent instances
agent = BidiAgent(model=model, tools=[stop_conversation])

When invoked, this tool signals the agent to end the conversation gracefully, closing the connection cleanly.

Source: strands-py/src/strands/experimental/bidi/tools/stop_conversation.py

Model Implementations

BidiNovaSonicModel

Amazon Nova Sonic bidirectional streaming model:

from strands.experimental.bidi.models import BidiNovaSonicModel

model = BidiNovaSonicModel(
    model_id="us.amazon.nova-pro-v1:0",
    session_duration_minutes=5,  # Connection timeout (max ~8 min)
    temperature=0.7,
)

Configuration Options:

Source: strands-py/src/strands/experimental/bidi/models/nova_sonic.py

BidiGeminiLiveModel

Google Gemini Live bidirectional streaming model:

from strands.experimental.bidi.models import BidiGeminiLiveModel

model = BidiGeminiLiveModel(
    model_id="gemini-2.0-flash-live",
    api_key="your_api_key",  # Or set GOOGLE_API_KEY env var
)

Source: strands-py/src/strands/experimental/bidi/models/gemini_live.py

BidiOpenAIRealtimeModel

OpenAI Realtime API bidirectional streaming model:

from strands.experimental.bidi.models import BidiOpenAIRealtimeModel

model = BidiOpenAIRealtimeModel(
    model_id="gpt-4o-realtime-preview",
    api_key="your_api_key",  # Or set OPENAI_API_KEY env var
)

Source: strands-py/src/strands/experimental/bidi/models/openai_realtime.py

Usage Patterns

Basic Voice Conversation

import asyncio
from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models import BidiNovaSonicModel
from strands.experimental.bidi.io import BidiAudioIO

async def main():
    model = BidiNovaSonicModel(model_id="us.amazon.nova-pro-v1:0")
    audio_io = BidiAudioIO()
    
    agent = BidiAgent(model=model)
    
    await agent.run(
        inputs=[audio_io.input()],
        outputs=[audio_io.output()]
    )

asyncio.run(main())

Multi-Modal Input with Voice and Text

import asyncio
from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models import BidiGeminiLiveModel
from strands.experimental.bidi.io import BidiAudioIO, BidiTextIO
from strands_tools import calculator

async def main():
    model = BidiGeminiLiveModel(
        model_id="gemini-2.0-flash-live",
        api_key="your_api_key"
    )
    
    audio_io = BidiAudioIO()
    text_io = BidiTextIO()
    
    agent = BidiAgent(
        model=model,
        tools=[calculator]
    )
    
    await agent.run(
        inputs=[audio_io.input(), text_io.input()],  # Speak OR type
        outputs=[audio_io.output(), text_io.output()]  # Both audio and text
    )

asyncio.run(main())

Using Tools in Streaming Sessions

import asyncio
from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models import BidiOpenAIRealtimeModel
from strands.experimental.bidi.io import BidiAudioIO
from strands_tools import calculator, python_executor

async def main():
    model = BidiOpenAIRealtimeModel(
        model_id="gpt-4o-realtime-preview",
    )
    
    agent = BidiAgent(
        model=model,
        tools=[calculator, python_executor]
    )
    
    await agent.run(
        inputs=[BidiAudioIO().input()],
        outputs=[BidiAudioIO().output()]
    )

asyncio.run(main())

Connection Lifecycle

stateDiagram-v2
    [*] --> Disconnected: Initial state
    Disconnected --> Connecting: agent.run()
    Connecting --> Connected: Connection established
    Connected --> Receiving: model.receive() loop starts
    Receiving --> Receiving: BidiOutputEvent received
    Receiving --> Sending: User input/tool result
    Sending --> Receiving: Send complete
    Receiving --> Disconnected: Connection closed/timeout
    Receiving --> [*]: Conversation ended
    Connecting --> Error: Connection failed
    Error --> [*]: Error handled

Session Timeout Behavior

Each model provider has specific connection timeout limits:

When a timeout occurs, the BidiModelTimeoutError is raised. The BidiAgent loop is designed to handle this gracefully by attempting to restart the connection when appropriate.

Source: strands-py/src/strands/experimental/bidi/models/model.py:113-120

Error Handling

Common Errors

Best Practices

1. Handle timeouts gracefully: Always catch BidiModelTimeoutError for robust applications
2. Use context managers: Ensure proper cleanup of audio resources
3. Validate I/O configuration: Ensure sample rates match provider requirements
4. Monitor connection state: Implement health checks for long-running sessions

Differences from Standard Agent

See Also

• Quickstart Guide - Official quickstart for bidirectional streaming
• Agent Loop - Understanding the standard agent loop
• Model Context Protocol (MCP) - Tool integration via MCP
• Amazon Bedrock Integration - Nova Sonic model configuration
• strands-agents GitHub - SDK repository

Related Pages

Related topics: Agent System

Session & Conversation Management

This page covers the session and conversation management system in Strands Agents SDK. It explains how agents maintain state across interactions, persist sessions to various backends, and manage conversation context to optimize for context window limits.

Overview

Strands Agents provides a comprehensive session and conversation management system that enables:

• State Persistence: Save and restore agent state across application restarts
• Context Management: Automatically manage conversation history within model context limits
• Multiple Storage Backends: Store sessions in local files, S3, or custom repositories
• Checkpointing: Experimental support for capturing and restoring agent execution state

The system is designed to be flexible, allowing developers to choose the appropriate conversation manager and session storage strategy for their use case.

Source: session_manager.py

Architecture Overview

graph TB
    subgraph "Session Layer"
        SM[SessionManager]
        FS[FileSessionManager]
        S3S[S3SessionManager]
        RSM[RepositorySessionManager]
        SR[SessionRepository]
    end
    
    subgraph "Conversation Layer"
        CM[ConversationManager]
        SWCM[SlidingWindowConversationManager]
        SUCM[SummarizingConversationManager]
    end
    
    subgraph "Agent Layer"
        A[Agent]
        SNAP[Snapshot]
        CP[Checkpoint]
    end
    
    SM --> FS
    SM --> S3S
    SM --> RSM
    RSM --> SR
    
    A --> CM
    CM --> SWCM
    CM --> SUCM
    
    A --> SNAP
    A --> CP

Session Management

The session management system handles persisting and retrieving agent state. Sessions can be stored in different backends depending on deployment requirements.

Source: session_manager.py:1-50

SessionManager

The SessionManager is the main entry point for session operations. It provides a unified interface regardless of the underlying storage backend.

Key Responsibilities:

• Load existing sessions by ID
• Save current session state
• Create new sessions
• Delete sessions
• Manage session metadata

Source: session_manager.py

Storage Backends

Strands supports multiple storage backends for session persistence:

#### FileSessionManager

Stores sessions as JSON files in a specified directory. Suitable for single-instance deployments or local development.

from strands.session import FileSessionManager

session_manager = FileSessionManager(
    storage_dir="/path/to/sessions"
)

Source: file_session_manager.py

#### S3SessionManager

Stores sessions in Amazon S3 buckets. Ideal for distributed deployments where multiple application instances need to access shared session state.

from strands.session import S3SessionManager

session_manager = S3SessionManager(
    bucket_name="my-sessions-bucket",
    prefix="agents/"
)

Source: s3_session_manager.py

#### RepositorySessionManager

A customizable backend that uses a SessionRepository interface. Allows integration with databases or custom storage systems.

from strands.session import RepositorySessionManager, SessionRepository

class MyDatabaseRepository(SessionRepository):
    async def save(self, session_id: str, state: dict) -> None:
        # Custom save logic
        pass
    
    async def load(self, session_id: str) -> dict | None:
        # Custom load logic
        pass

session_manager = RepositorySessionManager(repository=MyDatabaseRepository())

Source: repository_session_manager.py

SessionRepository Interface

The SessionRepository defines the contract for session storage operations:

Source: session_repository.py

Conversation Management

Conversation managers control how conversation history is maintained and optimized. They handle context window management through various strategies.

Source: conversation_manager.py

Base ConversationManager

The base ConversationManager class provides the foundation for conversation management:

Core Methods:

Source: conversation_manager.py

SlidingWindowConversationManager

The default conversation manager that maintains a sliding window of the most recent messages. When the context window approaches its limit, older messages are trimmed from the conversation.

Features:

• Configurable window size (number of messages to retain)
• Automatic trimming when context limits are approached
• Fallback trim point for tool-heavy conversations

Source: sliding_window_conversation_manager.py

from strands.agent.conversation_manager import SlidingWindowConversationManager

agent = Agent(
    conversation_manager=SlidingWindowConversationManager(
        max_messages=20,  # Keep last 20 messages
        trim_ratio=0.8    # Start trimming when 80% full
    )
)

Configuration Options:

Source: sliding_window_conversation_manager.py

SummarizingConversationManager

This manager periodically summarizes conversation history to preserve important context while reducing token usage. Useful for long-running agent interactions.

Features:

• Token-based trimming thresholds
• Automatic summarization of older messages
• Configurable summary triggers

Source: summarizing_conversation_manager.py

from strands.agent.conversation_manager import SummarizingConversationManager

agent = Agent(
    conversation_manager=SummarizingConversationManager(
        max_tokens=100000,      # Start summarization at 100k tokens
        summary_trigger=0.9,   # Trigger at 90% of max
        model="anthropic-sonnet-4-20250514"
    )
)

Proactive Context Compression

Introduced in v1.40.0, proactive context compression anticipates context window limits before they are reached. This feature uses a lookup table for context window limits and performs compression in advance rather than reactively.

Source: designs/0008-proactive-context-compression.md

Key Benefits:

• Prevents context overflow errors by acting before limits are reached
• Maintains conversation quality by preserving recent context
• Reduces latency spikes from sudden large trims

sequenceDiagram
    participant Agent
    participant CM as ConversationManager
    participant Model
    
    Agent->>CM: add_message(user_input)
    CM->>CM: check_context_usage()
    alt Context approaching limit
        CM->>CM: proactive_trim()
        Note over CM: Trim before limit reached
    end
    Agent->>Model: process_with_messages()
    Model-->>Agent: response

Source: designs/0008-proactive-context-compression.md

Snapshots

Snapshots provide a way to capture and restore the complete state of an agent. They are used internally for session persistence and can be used programmatically for state management.

Source: types/_snapshot.py

Snapshot Data Structure

Capturable Fields:

• messages — Full conversation history
• state — Agent state variables
• conversation_manager_state — Conversation manager state
• interrupt_state — Current interrupt state
• system_prompt — System prompt content blocks

Source: types/_snapshot.py

Creating and Loading Snapshots

# Create a snapshot
snapshot = agent.create_snapshot(
    fields=["messages", "state", "conversation_manager_state"]
)

# Save snapshot
import json
with open("session.json", "w") as f:
    json.dump(snapshot.to_dict(), f)

# Later, restore from snapshot
import json
with open("session.json", "r") as f:
    data = json.load(f)
    
snapshot = Snapshot.from_dict(data)
agent.load_snapshot(snapshot)

Checkpoint (Experimental)

The checkpoint feature enables capturing and restoring agent execution state during a running conversation. This is particularly useful for:

• Long-running tasks that may need to be interrupted
• Recovery from failures
• State inspection during debugging

Source: checkpoint/checkpoint.py

from strands.experimental.checkpoint import Checkpoint

# Create a checkpoint
checkpoint = Checkpoint.capture(agent)

# Store or transmit the checkpoint
checkpoint_data = checkpoint.serialize()

# Restore from checkpoint
restored_checkpoint = Checkpoint.deserialize(checkpoint_data)
restored_checkpoint.restore(agent)

Note: Checkpoint functionality is experimental and subject to change. It was introduced in v1.37.0.

Source: checkpoint/checkpoint.py

Usage Patterns

Basic Session Usage

from strands import Agent
from strands.session import FileSessionManager

# Create session manager
session_manager = FileSessionManager(storage_dir="./sessions")

# Create agent with session management
agent = Agent(
    tools=[calculator],
    session_manager=session_manager,
    session_id="user-123-session-001"
)

# Agent state is automatically persisted
agent("Hello, analyze this data")
# ... later ...
agent("Continue with the analysis")  # Picks up where left off

Multi-Agent Sessions

from strands import Agent
from strands.session import S3SessionManager

session_manager = S3SessionManager(
    bucket_name="my-app-sessions",
    prefix="multi-agent/"
)

# Shared session across agents
orchestrator = Agent(
    system_prompt="You coordinate a team of specialized agents.",
    session_manager=session_manager,
    session_id="project-alpha"
)

researcher = Agent(
    system_prompt="You research topics thoroughly.",
    session_manager=session_manager,
    session_id="project-alpha"  # Same session
)

Custom Conversation Manager

from strands.agent.conversation_manager import SlidingWindowConversationManager

# Optimize for brief interactions
short_window = SlidingWindowConversationManager(
    max_messages=10,
    trim_ratio=0.7  # Start trimming earlier
)

# Optimize for detailed research
long_window = SlidingWindowConversationManager(
    max_messages=100,
    trim_ratio=0.9  # Allow more context
)

agent = Agent(
    tools=[research_tool, analysis_tool],
    conversation_manager=long_window
)

Common Patterns and Best Practices

Session ID Strategy

Conversation Manager Selection

Error Handling

from strands.exceptions import SessionNotFoundError

try:
    agent = Agent(
        session_manager=session_manager,
        session_id="nonexistent-session"
    )
except SessionNotFoundError:
    # Create new session
    agent = Agent(session_manager=session_manager)

Related Components

Agent State Management

The agent integrates with session and conversation management through its state property:

# Custom state
agent.state["task_progress"] = 0
agent.state["visited_urls"] = []

# Access in system prompt
agent = Agent(
    system_prompt="You have completed {state.task_progress} steps.",
    state=agent.state
)

Telemetry Integration

Session operations are automatically traced when telemetry is enabled:

from strands.telemetry import StrandsTelemetry

StrandsTelemetry.configure(
    service_name="my-agent-service",
    exporter="console"  # or "otlp", "langfuse", etc.
)

See Also

• Agent Architecture — Overview of the Agent class and its components
• Tools & MCP Integration — How tools are integrated with conversation management
• Telemetry & Observability — Monitoring session operations
• Context Window Management — Advanced context optimization techniques

Related Pages

Related topics: Tool System, Agent System

MCP Integration

Overview

The Model Context Protocol (MCP) Integration in Strands Agents provides native support for connecting to MCP servers, enabling agents to use tools, resources, and prompts exposed by external MCP-compliant servers. This integration allows seamless access to thousands of pre-built tools through the standardized MCP interface.

The MCP client implementation handles:

• Transport management - Establishing and maintaining connections via stdio or HTTP/SSE
• Tool discovery - Listing and filtering available tools from MCP servers
• Tool execution - Both direct execution and task-augmented execution for long-running operations
• Resource access - Reading resources and resource templates from servers
• Prompt retrieval - Fetching and using server-defined prompts

Source: strands-py/src/strands/tools/mcp/mcp_client.py:1-50

Architecture

Component Overview

The MCP integration consists of several interconnected components:

graph TD
    A[Agent] --> B[MCPClient]
    B --> C[ClientSession]
    C --> D[MCP Server]
    
    B --> E[Tool Provider]
    E --> F[MCPAgentTool]
    
    B --> G[Task Executor]
    G --> H[call_tool_as_task]
    G --> I[call_tool]
    
    C --> J[Resources]
    C --> K[Prompts]
    
    L[Background Thread] -.-> C
    M[Event Loop] -.-> L

Threading Model

The MCP client uses a background thread with its own event loop to manage the MCP connection. This design ensures that the main thread remains responsive while maintaining a stable MCP session.

sequenceDiagram
    participant Main as Main Thread
    participant Background as Background Thread
    participant MCP as MCP Server
    
    Main->>Background: start()
    Background->>MCP: Initialize connection
    MCP-->>Background: Session initialized
    Background-->>Main: Ready signal
    
    loop Tool Calls
        Main->>Background: call_tool_async()
        Background->>MCP: Execute tool
        MCP-->>Background: Tool result
        Background-->>Main: Return result
    end
    
    Main->>Background: close()
    Background->>MCP: Close session
    Background-->>Main: Closed

Source: strands-py/src/strands/tools/mcp/mcp_client.py:180-220

MCPClient Class

Initialization

The MCPClient class is the main entry point for MCP integration:

from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

client = MCPClient(
    transport_callable=lambda: stdio_client(
        StdioServerParameters(command="uvx", args=["some-mcp-server@latest"])
    ),
    tool_filters={"rejected": ["internal.*"]},  # Optional filtering
    prefix="mcp_",  # Optional tool name prefix
    startup_timeout=timedelta(seconds=30),  # Connection timeout
)

Configuration Options

Source: strands-py/src/strands/tools/mcp/mcp_client.py:100-150

Tool Filtering

The MCP client supports filtering which tools are loaded from a server. This is useful when you want to exclude internal tools or limit the tool surface area.

Filter Configuration

from strands.tools.mcp import MCPClient, ToolFilters

# Only allow specific tools
filters: ToolFilters = {
    "allowed": ["calculator", "search", "^fetch_.*"],
    "rejected": ["debug.*", "admin.*"]
}

client = MCPClient(
    transport_callable=transport,
    tool_filters=filters
)

Filter Matching Rules

Tools are filtered in this order:

1. If allowed is specified, only tools matching these patterns are included
2. Tools matching rejected patterns are then excluded

Pattern types supported:

• String exact match: "calculator"
• Regex pattern: "^fetch_.*" (compiled automatically)
• Callback function: Custom filtering logic

def custom_filter(tool: AgentTool, **kwargs) -> bool:
    return "production" in tool.name

filters: ToolFilters = {"allowed": [custom_filter]}

Source: strands-py/src/strands/tools/mcp/mcp_client.py:60-80

Task-Augmented Execution

The MCP specification defines task-augmented execution for long-running tools. This feature allows the SDK to:

1. Create a task for a tool execution
2. Poll for completion status
3. Retrieve the final result

This pattern is particularly useful for tools that take minutes to complete, as it provides better feedback mechanisms.

Source: strands-py/src/strands/tools/mcp/mcp_tasks.py:1-30

Task Configuration

from datetime import timedelta
from strands.tools.mcp import MCPClient, TasksConfig

tasks_config: TasksConfig = {
    "ttl": timedelta(minutes=2),        # Task time-to-live
    "poll_timeout": timedelta(minutes=10)  # Polling timeout
}

client = MCPClient(
    transport_callable=transport,
    tasks_config=tasks_config
)

Default Configuration

Source: strands-py/src/strands/tools/mcp/mcp_tasks.py:25-35

Execution Flow

graph TD
    A[Tool Call Request] --> B{Server supports tasks?}
    B -->|Yes| C{Tool supports tasks?}
    B -->|No| G[Direct call_tool]
    C -->|TASK_REQUIRED| D[Use call_tool_as_task]
    C -->|TASK_OPTIONAL| D
    C -->|forbidden| G
    
    D --> E[Create task]
    E --> F[Poll until complete]
    F --> H[Return result]
    G --> I[Execute directly]
    I --> H

The SDK automatically determines whether to use task-augmented execution based on:

1. Server capability: tasks.requests.tools.call
2. Tool-level setting: taskSupport in tool specification

Source: strands-py/src/strands/tools/mcp/mcp_client.py:300-380

Resources and Prompts

Listing Resources

with client:
    # List available resources
    resources = client.list_resources_sync()
    print(f"Found {len(resources.resources)} resources")
    
    # List resource templates
    templates = client.list_resource_templates_sync()
    print(f"Found {len(templates.resourceTemplates)} templates")

Reading Resources

from mcp.types import AnyUrl

# Read a specific resource
resource_uri = AnyUrl("file:///path/to/resource")
content = client.read_resource_sync(resource_uri)

Source: strands-py/src/strands/tools/mcp/mcp_resources.py

Using Prompts

# List available prompts
prompts = client.list_prompts_sync()

# Get a specific prompt
prompt_result = client.get_prompt_sync(
    prompt_id="my-prompt",
    args={"variable": "value"}
)

Usage Examples

Basic MCP Server Connection

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# Create MCP client
mcp_client = MCPClient(
    transport_callable=lambda: stdio_client(
        StdioServerParameters(
            command="uvx",
            args=["awslabs.aws-documentation-mcp-server@latest"]
        )
    )
)

# Use with agent
with mcp_client:
    agent = Agent(tools=mcp_client.list_tools_sync())
    response = agent("Tell me about Amazon Bedrock")

Filtering Tools

from strands import Agent
from strands.tools.mcp import MCPClient

mcp_client = MCPClient(
    transport_callable=lambda: stdio_client(
        StdioServerParameters(command="uvx", args=["some-server@latest"])
    ),
    tool_filters={
        "rejected": ["internal.*", ".*debug.*"]
    }
)

with mcp_client:
    tools = mcp_client.list_tools_sync()
    print(f"Loaded {len(tools)} tools (filtered)")

With Task-Augmented Execution

from datetime import timedelta
from strands.tools.mcp import MCPClient, TasksConfig

mcp_client = MCPClient(
    transport_callable=transport,
    tasks_config=TasksConfig(
        ttl=timedelta(minutes=5),
        poll_timeout=timedelta(minutes=30)
    )
)

Error Handling

Common Exceptions

Session State Management

The client maintains a session state that can be checked:

if client._is_session_active():
    tools = client.list_tools_sync()
else:
    print("Client session is not active")

Always use the context manager (with client:) or explicitly call start() and close() to manage session lifecycle properly.

Source: strands-py/src/strands/tools/mcp/mcp_client.py:400-450

Instrumenting MCP Operations

The MCP integration includes OpenTelemetry instrumentation for observability:

from strands.tools.mcp.mcp_instrumentation import mcp_instrumentation

# Instrumentation is automatically enabled when MCPClient is used
# It can be explicitly triggered:
mcp_instrumentation()

This provides:

• Trace propagation for tool calls
• Span attributes for MCP operations
• Timing metrics for performance monitoring

Source: strands-py/src/strands/tools/mcp/mcp_instrumentation.py

Known Limitations

Progress Updates for Tasks

As noted in GitHub Issue #1812, the current implementation does not support progress updates for long-running tasks. The MCP specification defines progress reporting mechanisms (as implemented in FastMCP), but Strands Agents currently lacks this feature.

Community Note: Progress updates are essential for tasks that take minutes to complete, as they provide feedback to users during execution. This is a requested enhancement.

MetricsClient Thread Safety

As reported in GitHub Issue #2342, the MetricsClient singleton is not thread-safe. When using multiple MCP clients concurrently, this may cause metrics collection issues.

Error Flag Preservation

As of v1.38.0, the CallToolResult.isError flag is properly preserved in MCPToolResult, ensuring error states are correctly propagated from MCP servers.

Source: strands-py/src/strands/tools/mcp/mcp_client.py:250-280

Best Practices

1. Always use context manager: Use with client: to ensure proper cleanup
2. Filter tools: Limit tool exposure with tool_filters to reduce attack surface
3. Set appropriate timeouts: Configure startup_timeout and tasks_config based on your server's characteristics
4. Check session state: Verify _is_session_active() before operations if using manual lifecycle management
5. Handle errors gracefully: Catch MCPClientInitializationError for connection failures

See Also

• Agent Loop Documentation
• Tool Executor Strategies
• Structured Output Tools
• Official MCP Specification
• GitHub Issue #1812 - Progress Updates Feature Request
• GitHub Issue #2286 - Monorepo Consolidation (includes MCP Server)

Related Pages

Related topics: Multi-Agent Patterns, Tool System

Plugin System

The Strands Agents Plugin System provides a modular, extensible architecture for adding capabilities to agents. Plugins serve as self-contained packages that can inject tools, modify system prompts, register callbacks, and orchestrate multi-agent workflows. This design allows developers to compose sophisticated agent behaviors from reusable, independently-versioned components.

Overview

Plugins in Strands are designed around three core principles:

1. Composition over inheritance: Plugins are composed into an agent rather than subclassing it
2. Declarative configuration: Plugin behavior is defined through structured configuration
3. Lifecycle hooks: Plugins participate in the agent's execution lifecycle at defined points

The plugin system supports multiple plugin types including:

• Agent Plugins: Inject tools, modify prompts, and register callbacks
• Multi-Agent Plugins: Coordinate multiple agents using Swarm or Graph orchestrators
• Skills Plugins: Enable progressive disclosure of instructions via AgentSkills.io integration

Source: designs/0001-plugins.md:1-50

Architecture

The plugin system comprises several interconnected components that work together to provide a cohesive extension mechanism.

graph TD
    A[Agent] --> B[PluginRegistry]
    B --> C[Plugin Decorator]
    B --> D[Plugin Discovery]
    
    C --> E[BasePlugin]
    E --> F[AgentSkills]
    E --> G[MultiAgentPlugin]
    E --> H[ContextOffloader]
    E --> I[Custom Plugins]
    
    D --> J[tools/ directory]
    D --> K[Python decorators]
    
    L[Agent Loop] --> M[Pre-Cycle Hooks]
    L --> N[Post-Cycle Hooks]
    L --> O[Tool Callbacks]
    
    style E fill:#f9f,stroke:#333,stroke-width:2px
    style A fill:#bbf,stroke:#333,stroke-width:2px

Component Overview

Source: strands-py/src/strands/plugins/plugin.py:1-100

Base Plugin Interface

All plugins inherit from the BasePlugin abstract base class, which defines the required interface for plugin implementation.

Plugin Structure

from strands.plugins import BasePlugin

class MyPlugin(BasePlugin):
    def __init__(self, config: dict | None = None):
        super().__init__(config)
    
    @property
    def name(self) -> str:
        return "my_plugin"
    
    def get_tools(self) -> list[FunctionInfo]:
        """Return list of tools provided by this plugin."""
        return []
    
    def get_system_prompt(self) -> str | None:
        """Return system prompt modifications."""
        return None
    
    def register_callbacks(self, callbacks: dict) -> None:
        """Register lifecycle callbacks."""
        pass

Required Interface Methods

Source: strands-py/src/strands/plugins/plugin.py:50-150

Plugin Registry

The PluginRegistry serves as the central authority for managing plugins within an agent. It handles plugin registration, tool aggregation, and lifecycle coordination.

Registration Flow

sequenceDiagram
    participant Agent
    participant Registry
    participant Plugin1
    participant Plugin2
    
    Agent->>Registry: register_plugin(plugin)
    Registry->>Plugin1: validate()
    Plugin1-->>Registry: validated
    Registry->>Registry: add to _plugins dict
    Registry->>Plugin1: get_tools()
    Plugin1-->>Registry: [tool1, tool2]
    Registry->>Registry: merge into _tools
    
    Note over Registry: Repeat for all plugins
    
    Agent->>Registry: get_all_tools()
    Registry-->>Agent: combined tools list

Registry Operations

Source: strands-py/src/strands/plugins/registry.py:1-80

AgentSkills Plugin

The AgentSkills plugin integrates with AgentSkills.io to enable progressive disclosure of skill instructions. This follows a pattern where metadata is injected into the system prompt upfront, and full instructions are loaded on demand via a tool.

Skill Loading Patterns

from strands import Agent
from strands.vended_plugins.skills import AgentSkills, Skill

# Load from filesystem via classmethods
skill = Skill.from_file("./skills/pdf-processing")
skills = Skill.from_directory("./skills/")

# Let the plugin resolve paths automatically
plugin = AgentSkills(skills=["./skills/pdf-processing"])
agent = Agent(plugins=[plugin])

Skill Structure

Each skill consists of:

1. Metadata (metadata.json): Contains skill name, description, and tags for upfront injection
2. Instructions (instructions.md): Full skill instructions loaded on demand

How It Works

graph LR
    A[Skill Directory] -->|Load metadata| B[System Prompt]
    A -->|Load on demand| C[Tool Call]
    
    B --> D[Agent knows skill exists]
    C --> E[Full instructions retrieved]
    
    D --> F[Context Window Efficiency]
    E --> G[Detailed Task Execution]

The AgentSkills plugin creates a tool for each skill that, when called, loads and returns the full skill instructions. This allows the agent to request detailed instructions only when needed, preserving context window space.

Source: strands-py/src/strands/vended_plugins/skills/agent_skills.py:1-100

Source: strands-py/src/strands/vended_plugins/skills/skill.py:1-60

Multi-Agent Plugin

The MultiAgentPlugin (introduced in v1.41.0) enables coordination of multiple agents using Swarm or Graph orchestrators. This allows complex workflows where different specialized agents collaborate on tasks.

Orchestration Patterns

Basic Usage

from strands import Agent
from strands.plugins import MultiAgentPlugin

# Create specialized agents
researcher = Agent(system_prompt="You are a research assistant...")
writer = Agent(system_prompt="You are a technical writer...")

# Create multi-agent plugin with graph orchestrator
multi_agent = MultiAgentPlugin(
    agents={
        "researcher": researcher,
        "writer": writer,
    },
    orchestrator="graph",  # or "swarm"
    entry_point="researcher",
)

agent = Agent(plugins=[multi_agent])

Agent Handoff Flow (Swarm)

graph TD
    A[User Request] --> B[Entry Agent]
    B --> C{Requires Specialist?}
    C -->|Yes| D[Handoff to Specialist]
    C -->|No| E[Handle Request]
    D --> F[Specialist Agent]
    F --> G{Requires Another Agent?}
    G -->|Yes| H[Handoff Back]
    G -->|No| I[Return Result]
    H --> B

Source: strands-py/src/strands/plugins/multiagent_plugin.py:1-100

Plugin Discovery

The plugin discovery mechanism enables automatic loading of plugins from the filesystem, supporting hot-reloading during development.

Discovery Locations

Hot Reloading Configuration

from strands import Agent

# Enable automatic tool loading and reloading from ./tools/
agent = Agent(load_tools_from_directory=True)
response = agent("Use any tools you find in the tools directory")

The discovery system watches the specified directory for changes and reloads plugins dynamically without restarting the agent.

Source: strands-py/src/strands/plugins/_discovery.py:1-60

Tool-Based Plugins

The @tool decorator provides a simple way to create plugins from Python functions. This is the recommended approach for most use cases.

Basic Tool Creation

from strands import Agent, tool

@tool
def word_count(text: str) -> int:
    """Count words in text.
    
    This docstring is used by the LLM to understand the tool's purpose.
    """
    return len(text.split())

agent = Agent(tools=[word_count])
response = agent("How many words are in this sentence?")

Tool Specification Extraction

The decorator automatically extracts metadata from the function:

1. Name: Function name (or custom override)
2. Description: Function's docstring (excluding Args section)
3. Input Schema: JSON schema derived from function signature

graph LR
    A[Python Function] -->|Inspect| B[Docstring Parser]
    A -->|Inspect| C[Signature Analyzer]
    B --> D[ToolSpec JSON]
    C --> D
    D --> E[LLM Tool Call]

Decorator Source

The decorator processes the function to create a standardized ToolSpec:

def extract_metadata(self) -> ToolSpec:
    """Extract metadata from the function to create a tool specification."""
    func_name = self.func.__name__
    description = self._extract_description_from_docstring()
    input_schema = self.input_model.model_json_schema()
    return {
        "name": func_name,
        "description": description,
        "inputSchema": {"json": input_schema}
    }

Source: strands-py/src/strands/plugins/decorator.py:1-80

Lifecycle Hooks

Plugins can register callbacks to participate in the agent's execution lifecycle. Hooks are called at specific points during the agent loop.

Available Hooks

Hook Registration

from strands import Agent

def my_pre_tool_callback(tool_name: str, arguments: dict) -> dict:
    """Modify tool arguments before execution."""
    # Add logging, validation, or argument transformation
    return arguments

agent = Agent(
    callbacks={
        "on_pre_tool_call": my_pre_tool_callback,
    }
)

Source: strands-py/src/strands/vended_plugins/steering/core/handler.py:1-80

Configuration

Agent Plugin Configuration

from strands import Agent

agent = Agent(
    plugins=[
        # Pass configuration dict to plugin
        MyPlugin(config={
            "option1": "value1",
            "option2": True,
        }),
    ],
    # Callback hooks
    callbacks={
        "on_pre_tool_call": my_callback,
    },
)

Plugin Configuration Schema

Common Patterns

Pattern 1: Simple Tool Plugin

from strands.plugins import tool

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression.
    
    Args:
        expression: A mathematical expression to evaluate.
    """
    return str(eval(expression))

Pattern 2: Stateful Plugin

from strands.plugins import BasePlugin

class DataStorePlugin(BasePlugin):
    def __init__(self, config: dict | None = None):
        super().__init__(config)
        self._store = {}
    
    @property
    def name(self) -> str:
        return "data_store"
    
    def get_tools(self) -> list:
        return [self._save, self._load]
    
    @tool
    def _save(self, key: str, value: str) -> str:
        """Save a key-value pair."""
        self._store[key] = value
        return f"Saved: {key}"
    
    @tool
    def _load(self, key: str) -> str:
        """Load a value by key."""
        return self._store.get(key, "Not found")

Pattern 3: Multi-Agent Workflow

from strands.plugins import MultiAgentPlugin

triage = Agent(system_prompt="Route requests to specialists...")
research = Agent(system_prompt="Research topics thoroughly...")
respond = Agent(system_prompt="Generate user-friendly responses...")

workflow = MultiAgentPlugin(
    agents={
        "triage": triage,
        "research": research,
        "respond": respond,
    },
    orchestrator="graph",
    entry_point="triage",
)

agent = Agent(plugins=[workflow])

Troubleshooting

Plugin Not Loading

1. Check import paths: Ensure the plugin module is importable
2. Verify plugin inheritance: Plugin must inherit from BasePlugin
3. Check configuration: Ensure config dict is properly structured

Tools Not Appearing

1. Verify get_tools() returns: Must return list[FunctionInfo]
2. Check decorator usage: Ensure @tool decorator is applied
3. Review tool schema: Validate input schema is correct JSON

Callback Not Triggering

1. Check hook name: Ensure callback key matches available hooks
2. Verify signature: Callbacks must accept expected parameters
3. Check return value: Some hooks expect modified arguments to be returned

See Also

• Agent Loop Documentation
• Tools System
• MCP Integration
• Telemetry System
• Multi-Agent Architectures
• AgentSkills.io Integration

Doramagic Pitfall Log

Doramagic extracted 8 source-linked risk signals. Review them before installing or handing real data to the project.

1. Project risk: Project risk needs validation

• Severity: medium
• Finding: Project risk is backed by a source signal: Project risk needs validation. Treat it as a review item until the current version is checked.
• User impact: The project should not be treated as fully validated until this signal is reviewed.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: identity.distribution | github_repo:983715534 | https://github.com/strands-agents/sdk-python | repo=sdk-python; install=strands-agents

2. Capability assumption: README/documentation is current enough for a first validation pass.

• Severity: medium
• Finding: README/documentation is current enough for a first validation pass.
• User impact: The project should not be treated as fully validated until this signal is reviewed.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: capability.assumptions | github_repo:983715534 | https://github.com/strands-agents/sdk-python | README/documentation is current enough for a first validation pass.

3. Maintenance risk: Maintainer activity is unknown

• Severity: medium
• Finding: Maintenance risk is backed by a source signal: Maintainer activity is unknown. Treat it as a review item until the current version is checked.
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: evidence.maintainer_signals | github_repo:983715534 | https://github.com/strands-agents/sdk-python | last_activity_observed missing

4. Security or permission risk: no_demo

• Severity: medium
• Finding: no_demo
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: downstream_validation.risk_items | github_repo:983715534 | https://github.com/strands-agents/sdk-python | no_demo; severity=medium

5. Security or permission risk: No sandbox install has been executed yet; downstream must verify before user use.

• Severity: medium
• Finding: No sandbox install has been executed yet; downstream must verify before user use.
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: risks.safety_notes | github_repo:983715534 | https://github.com/strands-agents/sdk-python | No sandbox install has been executed yet; downstream must verify before user use.

6. Security or permission risk: no_demo

• Severity: medium
• Finding: no_demo
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: risks.scoring_risks | github_repo:983715534 | https://github.com/strands-agents/sdk-python | no_demo; severity=medium

7. Maintenance risk: issue_or_pr_quality=unknown

• Severity: low
• Finding: issue_or_pr_quality=unknown。
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: evidence.maintainer_signals | github_repo:983715534 | https://github.com/strands-agents/sdk-python | issue_or_pr_quality=unknown

8. Maintenance risk: release_recency=unknown

• Severity: low
• Finding: release_recency=unknown。
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: evidence.maintainer_signals | github_repo:983715534 | https://github.com/strands-agents/sdk-python | release_recency=unknown

Community Discussion Evidence

Doramagic exposes project-level community discussion separately from official documentation. Review these links before using sdk-python with real data or production workflows.

• [[FEATURE] MCP Task Progress Updates](https://github.com/strands-agents/sdk-python/issues/1812) - github / github_issue
• [[Awareness] Consolidating Strands SDK Repositories into a Monorepo](https://github.com/strands-agents/sdk-python/issues/2286) - github / github_issue
• Python wheel missing README and LICENSE in published package - github / github_issue
• [[BUG] GeminiModel crashes with TypeError when safety_settings block cont](https://github.com/strands-agents/sdk-python/issues/2347) - github / github_issue
• [[BUG] MetricsClient singleton is not thread-safe](https://github.com/strands-agents/sdk-python/issues/2342) - github / github_issue
• v1.41.0 - github / github_release
• v1.40.0 - github / github_release
• v1.39.0 - github / github_release
• v1.38.0 - github / github_release
• v1.37.0 - github / github_release
• v1.36.0 - github / github_release
• v1.35.0 - github / github_release