Strands Agents

https://strandsagents.com/docs/user-guide/quickstart/python/
Admin
Strands Agents is an open-source framework developed by AWS for building, observing, and securing...
Tokens:4,779
Snippets:34
Trust Score:8.6
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install Strands Agents SDK

Source: https://strandsagents.com

Install the SDK using pip for Python or npm for Node.js environments.

```shell
pip install strands-agents
```

```shell
npm install @strands-agents/sdk
```

--------------------------------

### Automate Lead Routing in Python

Source: https://strandsagents.com

Define tools for lead classification and assignment, then register them with an Agent instance.

```python
from strands import Agent, tool

@tool
def classify_lead(email: str, company: str) -> dict:
    """Score and classify an inbound lead."""
    firmographics = crm.lookup(company)
    return {
        "score": compute_icp_score(firmographics),
        "segment": firmographics["industry"],
    }

@tool
def route_to_rep(lead_id: str, region: str) -> str:
    """Assign a lead to the right sales rep."""
    rep = crm.get_rep_for_region(region)
    crm.assign(lead_id, rep)
    return f"Assigned to {rep}"

agent = Agent(
    tools=[classify_lead, route_to_rep],
)

agent("New lead: jane@acme.com, Acme Corp, US-West")
```

```python
from strands import Agent, tool

@tool
def classify_lead(email: str, company: str) -> dict:
    """Score and classify an inbound lead."""
    firmographics = crm.lookup(company)
    return {
        "score": compute_icp_score(firmographics),
        "segment": firmographics["industry"],
    }

@tool
def route_to_rep(lead_id: str, region: str) -> str:
    """Assign a lead to the right sales rep."""
    rep = crm.get_rep_for_region(region)
    crm.assign(lead_id, rep)
    return f"Assigned to {rep}"

agent = Agent(
    tools=[classify_lead, route_to_rep],
)

agent("New lead: jane@acme.com, Acme Corp, US-West")
```

--------------------------------

### Build a Python AI Support Assistant

Source: https://strandsagents.com

Configures an agent with MCP knowledge base integration, a custom refund tool, and a hook to require human approval for sensitive actions.

```python
from strands import Agent, tool
from strands.tools.mcp import MCPClient
from strands.hooks import BeforeToolCallEvent
from strands.agent import SlidingWindowConversationManager
from mcp import stdio_client, StdioServerParameters

kb = MCPClient(lambda: stdio_client(
    StdioServerParameters(command="uvx", args=["kb-server"])
))

@tool
def issue_refund(order_id: str, amount: float) -> str:
    """Process a customer refund."""
    return payments.refund(order_id, amount)

def approve_refunds(event: BeforeToolCallEvent):
    """Pause for human approval before processing refunds."""
    if event.tool_use["name"] == "issue_refund":
        response = event.interrupt(
            "refund_approval", reason=event.tool_use["input"]
        )
        if response != "APPROVE":
            event.cancel_tool = "Refund not approved."

agent = Agent(
    system_prompt="Support assistant. Use the KB. "
    "Refunds require approval.",
    tools=[kb, issue_refund],
    hooks=[approve_refunds],
    conversation_manager=SlidingWindowConversationManager(
        window_size=20
    ),
)
```

--------------------------------

### Automate Lead Routing in TypeScript

Source: https://strandsagents.com

Implement tools using Zod schemas for input validation and invoke the agent asynchronously.

```typescript
import { Agent, tool } from '@strands-agents/sdk'
import z from 'zod'

const classifyLead = tool({
  name: 'classify_lead',
  description: 'Score and classify a lead.',
  inputSchema: z.object({
    email: z.string(),
    company: z.string(),
  }),
  callback: ({ email, company }) => {
    const data = crm.lookup(company)
    return {
      score: computeIcpScore(data),
      segment: data.industry,
    }
  },
})

const routeToRep = tool({
  name: 'route_to_rep',
  description: 'Assign a lead to a rep.',
  inputSchema: z.object({
    leadId: z.string(),
    region: z.string(),
  }),
  callback: ({ leadId, region }) => {
    const rep = crm.getRepForRegion(region)
    crm.assign(leadId, rep)
    return `Assigned to ${rep}`
  },
})

const agent = new Agent({
  tools: [classifyLead, routeToRep],
})

await agent.invoke(
  'New lead: jane@acme.com, Acme Corp, US-West'
)
```

```typescript
import { Agent, tool } from '@strands-agents/sdk'
import z from 'zod'

const classifyLead = tool({
  name: 'classify_lead',
  description: 'Score and classify a lead.',
  inputSchema: z.object({
    email: z.string(),
    company: z.string(),
  }),
  callback: ({ email, company }) => {
    const data = crm.lookup(company)
    return {
      score: computeIcpScore(data),
      segment: data.industry,
    }
  },
})

const routeToRep = tool({
  name: 'route_to_rep',
  description: 'Assign a lead to a rep.',
  inputSchema: z.object({
    leadId: z.string(),
    region: z.string(),
  }),
  callback: ({ leadId, region }) => {
    const rep = crm.getRepForRegion(region)
    crm.assign(leadId, rep)
    return `Assigned to ${rep}`
  },
})

const agent = new Agent({
  tools: [classifyLead, routeToRep],
})

await agent.invoke(
  'New lead: jane@acme.com, Acme Corp, US-West'
)
```

--------------------------------

### Define a tool and agent in Python

Source: https://strandsagents.com

Use the @tool decorator to define functions and pass them to the Agent constructor.

```python
from strands import Agent, tool

@tool
def search_logs(query: str, hours: int = 24) -> list:
    """Search application logs by keyword."""
    return log_api.search(query, hours)

agent = Agent(
    tools=[search_logs],
)

agent("Find all timeout errors from the last 6 hours")
```

--------------------------------

### Build a TypeScript AI Support Assistant

Source: https://strandsagents.com

Configures an agent using the TypeScript SDK with MCP transport, Zod-validated tools, and hook-based tool cancellation.

```typescript
import {
  Agent, tool, McpClient,
  BeforeToolCallEvent,
  SlidingWindowConversationManager,
} from '@strands-agents/sdk'
import { StdioClientTransport } from
  '@modelcontextprotocol/sdk/client/stdio.js'
import z from 'zod'

const kb = new McpClient({
  transport: new StdioClientTransport({
    command: 'npx',
    args: ['kb-server'],
  }),
})

const issueRefund = tool({
  name: 'issue_refund',
  description: 'Process a refund.',
  inputSchema: z.object({
    orderId: z.string(),
    amount: z.number(),
  }),
  callback: ({ orderId, amount }) =>
    payments.refund(orderId, amount),
})

const agent = new Agent({
  systemPrompt: 'Support assistant. '
    + 'Use KB. Refunds require approval.',
  tools: [kb, issueRefund],
  conversationManager:
    new SlidingWindowConversationManager({
      windowSize: 20,
    }),
})

// Cancel refunds (interrupt coming soon to TS)
agent.addHook(BeforeToolCallEvent, (event) => {
  if (event.toolUse.name === 'issue_refund') {
    event.cancel = 'Refund approval required.'
  }
})
```

--------------------------------

### Build a Research Agent in Python

Source: https://strandsagents.com

Uses Pydantic models to define structured output for agent-generated research briefings.

```python
from pydantic import BaseModel, Field
from strands import Agent
from strands_tools import http_request
from pathlib import Path

class Briefing(BaseModel):
    headline: str = Field(description="One-line summary")
    developments: list[str] = Field(
        description="Key developments"
    )
    sources: list[str] = Field(
        description="URLs consulted"
    )

agent = Agent(
    system_prompt="Research assistant. Search the web, "
    "find developments from the last 24 hours, "
    "and produce a briefing with citations.",
    tools=[http_request],
)

result = agent(
    "What happened in AI agent frameworks yesterday?",
    structured_output_model=Briefing,
)

briefing = result.structured_output
Path("briefings/daily.md").write_text(
    f"# {briefing.headline}\n\n"
    + "\n".join(f"- {d}" for d in briefing.developments)
)
```

```python
from pydantic import BaseModel, Field
from strands import Agent
from strands_tools import http_request
from pathlib import Path

class Briefing(BaseModel):
    headline: str = Field(description="One-line summary")
    developments: list[str] = Field(
        description="Key developments"
    )
    sources: list[str] = Field(
        description="URLs consulted"
    )

agent = Agent(
    system_prompt="Research assistant. Search the web, "
    "find developments from the last 24 hours, "
    "and produce a briefing with citations.",
    tools=[http_request],
)

result = agent(
    "What happened in AI agent frameworks yesterday?",
    structured_output_model=Briefing,
)

briefing = result.structured_output
Path("briefings/daily.md").write_text(
    f"# {briefing.headline}\n\n"
    + "\n".join(f"- {d}" for d in briefing.developments)
)
```

--------------------------------

### Configure MCP Server

Source: https://strandsagents.com

Add the Strands Agents MCP server to your AI coding assistant configuration.

```shell
claude mcp add strands uvx strands-agents-mcp-server
```

```json
{
  "mcpServers": {
    "strands-agents": {
      "command": "uvx",
      "args": ["strands-agents-mcp-server"]
    }
  }
}
```

```json
{
  "mcpServers": {
    "strands-agents": {
      "command": "uvx",
      "args": ["strands-agents-mcp-server"],
      "disabled": false,
      "autoApprove": ["search_docs", "fetch_doc"]
    }
  }
}
```

```json
{
  "servers": {
    "strands-agents": {
      "command": "uvx",
      "args": ["strands-agents-mcp-server"]
    }
  }
}
```

--------------------------------

### Build a Research Agent in TypeScript

Source: https://strandsagents.com

Uses Zod schemas to define structured output for agent-generated research briefings.

```typescript
import { Agent } from '@strands-agents/sdk'
import { httpRequest } from '@strands-agents/tools'
import z from 'zod'
import { writeFileSync } from 'fs'

const BriefingSchema = z.object({
  headline: z.string().describe('Summary'),
  developments: z.array(z.string())
    .describe('Key developments'),
  sources: z.array(z.string())
    .describe('URLs consulted'),
})

const agent = new Agent({
  systemPrompt: 'Research assistant. '
    + 'Search the web. Cite sources.',
  tools: [httpRequest],
})

const result = await agent.invoke(
  'AI agent frameworks: what happened yesterday?',
  { structuredOutputSchema: BriefingSchema },
)

const briefing = result.structuredOutput
writeFileSync('briefings/daily.md',
  `# ${briefing.headline}\n\n`
  + briefing.developments
    .map((d: string) => `- ${d}`)
    .join('\n')
)
```

```typescript
import { Agent } from '@strands-agents/sdk'
import { httpRequest } from '@strands-agents/tools'
import z from 'zod'
import { writeFileSync } from 'fs'

const BriefingSchema = z.object({
  headline: z.string().describe('Summary'),
  developments: z.array(z.string())
    .describe('Key developments'),
  sources: z.array(z.string())
    .describe('URLs consulted'),
})

const agent = new Agent({
  systemPrompt: 'Research assistant. '
    + 'Search the web. Cite sources.',
  tools: [httpRequest],
})

const result = await agent.invoke(
  'AI agent frameworks: what happened yesterday?',
  { structuredOutputSchema: BriefingSchema },
)

const briefing = result.structuredOutput
writeFileSync('briefings/daily.md',
  `# ${briefing.headline}\n\n`
  + briefing.developments
    .map((d: string) => `- ${d}`)
    .join('\n')
)
```

--------------------------------

### Define a Query Quality Steering Policy

Source: https://strandsagents.com

Extends SteeringHandler to provide feedback on SQL queries, such as requiring WHERE clauses or limiting the number of joins.

```python
from strands.vended_plugins.steering import (
    SteeringHandler, Guide, Proceed,
)

class QueryQualityPolicy(SteeringHandler):
    async def steer_before_tool(
        self, *, agent, tool_use, **kwargs
    ):
        sql = tool_use["input"].get("query", "").upper()
        if "SELECT" in sql and "WHERE" not in sql:
            return Guide(
                reason="Add a WHERE clause and LIMIT."
            )
        if sql.upper().count("JOIN") > 3:
            return Guide(
                reason="4+ joins. Break into smaller queries."
            )
        return Proceed(reason="Query looks good.")

agent = Agent(
    tools=[query_database],
    plugins=[QueryQualityPolicy()],
)
```

--------------------------------

### Define a Research Agent in Python

Source: https://strandsagents.com

Create an agent with a file-saving tool and a hook to enforce source citations.

```python
from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from pathlib import Path

@tool
def save_report(title: str, content: str) -> str:
    """Save a research report to disk."""
    path = f"reports/{title}.md"
    Path(path).write_text(content)
    return f"Saved {path}"

def require_sources(event: BeforeToolCallEvent):
    name = event.tool_use["name"]
    inp = str(event.tool_use["input"])
    if name == "save_report" and "[source]" not in inp:
        event.cancel_tool = "Add source citations."

agent = Agent(
    tools=[save_report],
    hooks=[require_sources],
)
agent("Research AI agent frameworks")
```

--------------------------------

### Add conversation summarization in Python

Source: https://strandsagents.com

Configure the agent with a SummarizingConversationManager to handle conversation history.

```python
from strands.agent import SummarizingConversationManager

# Same agent, now with summarization.
agent = Agent(
    tools=[search_logs],
    conversation_manager=SummarizingConversationManager(),
)
```

--------------------------------

### Implement a Read-Only Guardrail

Source: https://strandsagents.com

Uses a BeforeToolCallEvent hook to intercept database queries and cancel tool execution if write operations are detected.

```python
from strands import Agent
from strands.hooks import BeforeToolCallEvent

WRITE_OPS = ["INSERT", "UPDATE", "DELETE", "DROP"]

def read_only_guard(event: BeforeToolCallEvent):
    """Block writes. This agent is read-only."""
    if event.tool_use["name"] == "query_database":
        sql = event.tool_use["input"].get("query", "")
        if any(kw in sql.upper() for kw in WRITE_OPS):
            event.cancel_tool = "Read-only access."

agent = Agent(
    tools=[query_database],
    hooks=[read_only_guard],
)
```

--------------------------------

### Define a tool and agent in TypeScript

Source: https://strandsagents.com

Use the tool function with Zod schemas for input validation and invoke the agent asynchronously.

```typescript
import { Agent, tool } from '@strands-agents/sdk'
import z from 'zod'

const searchLogs = tool({
  name: 'search_logs',
  description: 'Search logs by keyword.',
  inputSchema: z.object({
    query: z.string(),
    hours: z.number().default(24),
  }),
  callback: ({ query, hours }) =>
    logApi.search(query, hours),
})

const agent = new Agent({ tools: [searchLogs] })

await agent.invoke(
  'Find all timeout errors from the last 6 hours'
)
```

--------------------------------

### Monitor tool calls with Python hooks

Source: https://strandsagents.com

Use the AfterToolCallEvent hook to intercept and log tool execution details within the agent loop.

```python
from strands import Agent
from strands.hooks import AfterToolCallEvent

def log_tool_calls(event: AfterToolCallEvent):
    """Log every tool call."""
    print(f"Tool: {event.tool_use['name']}")
    print(f"Result: {event.result['status']}")

agent = Agent(
    tools=[search_logs, query_database],
    hooks=[log_tool_calls],
    trace_attributes={
        "service": "ops-agent",
        "env": "production",
    },
)
```

--------------------------------

### Add conversation summarization in TypeScript

Source: https://strandsagents.com

Pass a new instance of SummarizingConversationManager to the agent configuration.

```typescript
import {
  SummarizingConversationManager,
} from '@strands-agents/sdk'

// Same agent, now with summarization.
const agent = new Agent({
  tools: [searchLogs],
  conversationManager:
    new SummarizingConversationManager(),
})
```

--------------------------------

### Define a Research Agent in TypeScript

Source: https://strandsagents.com

Create an agent using the TypeScript SDK with Zod for input validation and a hook for tool call interception.

```typescript
import {
  Agent, tool, BeforeToolCallEvent
} from '@strands-agents/sdk'
import z from 'zod'
import { writeFileSync } from 'fs'

const saveReport = tool({
  name: 'save_report',
  description: 'Save a research report.',
  inputSchema: z.object({
    title: z.string(),
    content: z.string(),
  }),
  callback: ({ title, content }) => {
    writeFileSync(`reports/${title}.md`, content)
    return `Saved ${title}.md`
  },
})

const agent = new Agent({ tools: [saveReport] })

agent.addHook(BeforeToolCallEvent, (event) => {
  const inp = String(event.toolUse.input)
  if (event.toolUse.name === 'save_report') {
    if (!inp.includes('[source]')) {
      event.cancel = 'Add source citations.'
    }
  }
})

await agent.invoke('Research AI agent frameworks')
```

--------------------------------

### Monitor tool calls with TypeScript hooks

Source: https://strandsagents.com

Register an AfterToolCallEvent hook using the addHook method to log tool execution status in TypeScript.

```typescript
import {
  Agent, AfterToolCallEvent,
} from '@strands-agents/sdk'

const agent = new Agent({
  tools: [searchLogs, queryDatabase],
  traceAttributes: {
    service: 'ops-agent',
    env: 'production',
  },
})

agent.addHook(AfterToolCallEvent, (event) => {
  console.log(`Tool: ${event.toolUse.name}`)
  console.log(`Status: ${event.result.status}`)
})
```
Tokens:
Raw