Aws Strands Agents Agentcore

AWS Strands Agents & AgentCore

Overview

AWS Strands Agents SDK: Open-source Python framework for building AI agents with model-driven orchestration (minimal code, model decides tool usage)

Amazon Bedrock AgentCore: Enterprise platform for deploying, operating, and scaling agents in production

Relationship: Strands SDK runs standalone OR with AgentCore platform services. AgentCore is optional but provides enterprise features (8hr runtime, streaming, memory, identity, observability).

---

Quick Start Decision Tree

What are you building?

Single-purpose agent:

• Event-driven (S3, SQS, scheduled) → Lambda deployment
• Interactive with streaming → AgentCore Runtime
• API endpoint (stateless) → Lambda

Multi-agent system:

• Deterministic workflow → Graph Pattern
• Autonomous collaboration → Swarm Pattern
• Simple delegation → Agent-as-Tool Pattern

Tool/Integration Server (MCP):

• ALWAYS deploy to ECS/Fargate or AgentCore Runtime
• NEVER Lambda (stateful, needs persistent connections)

See [architecture.md](references/architecture.md) for deployment examples.

---

Critical Constraints

MCP Server Requirements

1. Transport: MUST use streamable-http (NOT stdio) 2. Endpoint: MUST be at 0.0.0.0:8000/mcp 3. Deployment: MUST be ECS/Fargate or AgentCore Runtime (NEVER Lambda) 4. Headers: Must accept application/json and text/event-stream

Why: MCP servers are stateful and need persistent connections. Lambda is ephemeral and unsuitable.

See [limitations.md](references/limitations.md) for details.

Tool Count Limits

• Models struggle with > 50-100 tools
• Solution: Implement semantic search for dynamic tool loading

See [patterns.md](references/patterns.md) for implementation.

Token Management

• Claude 4.5: 200K context (use ~180K max)
• Long conversations REQUIRE conversation managers
• Multi-agent costs multiply 5-10x

See [limitations.md](references/limitations.md) for strategies.

---

Deployment Decision Matrix

---

Multi-Agent Pattern Selection

Recommendation: Start with single agents, evolve as needed.

See [architecture.md](references/architecture.md) for examples.

---

When to Read Reference Files

patterns.md

• Base agent factory patterns (reusable components)
• MCP server registry patterns (tool catalogues)
• Semantic tool search (> 50 tools)
• Tool design best practices
• Security patterns
• Testing patterns

observability.md

• AWS AgentCore Observability Platform setup
• Runtime-hosted vs self-hosted configuration
• Session tracking for multi-turn conversations
• OpenTelemetry setup
• Cost tracking hooks
• Production observability patterns

evaluations.md

• AWS AgentCore Evaluations - Quality assessment with LLM-as-a-Judge
• 13 built-in evaluators (Helpfulness, Correctness, GoalSuccessRate, etc.)
• Custom evaluators with your own prompts and models
• Online (continuous) and on-demand evaluation modes
• CloudWatch integration and alerting

limitations.md

• MCP server deployment issues
• Tool selection problems (> 50 tools)
• Token overflow
• Lambda limitations
• Multi-agent cost concerns
• Throttling errors
• Cold start latency

---

#-Driven Philosophy

Key Concept: Strands Agents delegates orchestration to the model rather than requiring explicit control flow code.

# Traditional: Manual orchestration (avoid)
while not done:
    if needs_research:
        result = research_tool()
    elif needs_analysis:
        result = analysis_tool()

# Strands: Model decides (prefer)
agent = Agent(
    system_prompt="You are a research analyst. Use tools to answer questions.",
    tools=[research_tool, analysis_tool]
)
result = agent("What are the top tech trends?")
 automatically orchestrates: research_tool → analysis_tool → respond

---

Selection

Primary Provider: Anthropic Claude via AWS Bedrock

Model ID Format: anthropic.claude-{model}-{version}

Current Models (as of January 2025):

• anthropic.claude-sonnet-4-5-20250929-v1:0 - Production
• anthropic.claude-haiku-4-5-20251001-v1:0 - Fast/economical
• anthropic.claude-opus-4-5-20250514-v1:0 - Complex reasoning

Check Latest Models:

aws bedrock list-foundation-models --by-provider anthropic \
  --query 'modelSummaries[*].[modelId,modelName]' --output table

---

Quick Examples

Basic Agent

from strands import Agent
from strands.models import BedrockModel
from strands.session import DynamoDBSessionManager
from strands.agent.conversation_manager import SlidingWindowConversationManager

agent = Agent(
    agent_id="my-agent",
    model=BedrockModel(model_id="anthropic.claude-sonnet-4-5-20250929-v1:0"),
    system_prompt="You are helpful.",
    tools=[tool1, tool2],
    session_manager=DynamoDBSessionManager(table_name="sessions"),
    conversation_manager=SlidingWindowConversationManager(max_messages=20)
)

result = agent("Process this request")

See [patterns.md](references/patterns.md) for base agent factory patterns.

MCP Server (ECS/Fargate)

from mcp.server import FastMCP
import psycopg2.pool

# Persistent connection pool (why Lambda won't work)
db_pool = psycopg2.pool.SimpleConnectionPool(minconn=1, maxconn=10, host="db.internal")

mcp = FastMCP("Database Tools")

@mcp.tool()
def query_database(sql: str) -> dict:
    conn = db_pool.getconn()
    try:
        cursor = conn.cursor()
        cursor.execute(sql)
        return {"status": "success", "rows": cursor.fetchall()}
    finally:
        db_pool.putconn(conn)

# CRITICAL: streamable-http mode
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

See [architecture.md](references/architecture.md) for deployment details.

Tool Error Handling

from strands import tool

@tool
def safe_tool(param: str) -> dict:
    """Always return structured results, never raise exceptions."""
    try:
        result = operation(param)
        return {"status": "success", "content": [{"text": str(result)}]}
    except Exception as e:
        return {"status": "error", "content": [{"text": f"Failed: {str(e)}"}]}

See [patterns.md](references/patterns.md) for tool design patterns.

Observability

AgentCore Runtime (Automatic):

# Install with OTEL support
# pip install 'strands-agents[otel]'
# Add 'aws-opentelemetry-distro' to requirements.txt

from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()
agent = Agent(...)  # Automatically instrumented

@app.entrypoint
def handler(payload):
    return agent(payload["prompt"])

Self-Hosted:

export AGENT_OBSERVABILITY_ENABLED=true
export OTEL_PYTHON_DISTRO=aws_distro
export OTEL_RESOURCE_ATTRIBUTES="service.name=my-agent"

opentelemetry-instrument python agent.py

General OpenTelemetry:

from strands.observability import StrandsTelemetry

# Development
telemetry = StrandsTelemetry().setup_console_exporter()

# Production
telemetry = StrandsTelemetry().setup_otlp_exporter()

See [observability.md](references/observability.md) for detailed patterns.

---

Session Storage Selection

Local dev         → FileSystem
Lambda agents     → S3 or DynamoDB
ECS agents        → DynamoDB
Interactive chat  → AgentCore Memory
Knowledge bases   → AgentCore Memory

See [architecture.md](references/architecture.md) for storage backend comparison.

---

When to Use AgentCore Platform vs SDK Only

Use Strands SDK Only

• Simple, stateless agents
• Tight cost control required
• No enterprise features needed
• Want deployment flexibility

Use Strands SDK + AgentCore Platform

• Need 8-hour runtime support
• Streaming responses required
• Enterprise security/compliance
• Cross-session intelligence needed
• Want managed infrastructure

See [architecture.md](references/architecture.md) for platform service details.

---

Common Anti-Patterns

1. ❌ Overloading agents with > 50 tools → Use semantic search 2. ❌ No conversation management → Implement SlidingWindow or Summarising 3. ❌ Deploying MCP servers to Lambda → Use ECS/Fargate 4. ❌ No timeout configuration → Set execution limits everywhere 5. ❌ Ignoring token limits → Implement conversation managers 6. ❌ No cost monitoring → Implement cost tracking from day one

See [patterns.md](references/patterns.md) and [limitations.md](references/limitations.md) for details.

---

Production Checklist

Before deploying:

• [ ] Conversation management configured
• [ ] AgentCore Observability enabled or OpenTelemetry configured
• [ ] AgentCore Evaluations configured for quality monitoring
• [ ] Observability hooks implemented
• [ ] Cost tracking enabled
• [ ] Error handling in all tools
• [ ] Security permissions validated
• [ ] MCP servers deployed to ECS/Fargate
• [ ] Timeout limits set
• [ ] Session backend configured (DynamoDB for production)
• [ ] CloudWatch alarms configured

---

Reference Files Navigation

• [architecture.md](references/architecture.md) - Deployment patterns, multi-agent orchestration, session storage, AgentCore services
• [patterns.md](references/patterns.md) - Foundation components, tool design, security, testing, performance optimisation
• [limitations.md](references/limitations.md) - Known constraints, workarounds, mitigation strategies, challenges
• [observability.md](references/observability.md) - AgentCore Observability platform, ADOT, GenAI dashboard, OpenTelemetry, hooks, cost tracking
• [evaluations.md](references/evaluations.md) - AgentCore Evaluations, built-in evaluators, custom evaluators, quality monitoring

---

Key Takeaways

1. MCP servers MUST use streamable-http, NEVER Lambda 2. Use semantic search for > 15 tools 3. Always implement conversation management 4. Multi-agent costs multiply 5-10x (track from day one) 5. Set timeout limits everywhere 6. Error handling in tools is non-negotiable 7. Lambda for stateless, AgentCore for interactive 8. AgentCore Observability and Evaluations for production 9. Start simple, evolve complexity 10. Security by default 11. Separate config from code

Architecture & Deployment Patterns

What is Strands Agents SDK?

Open-source Python SDK for building AI agents with model-driven orchestration (minimal code).

Core Components:

• Agent: Model + tools + system prompt
• @tool: Decorator for agent-callable functions
• Multi-Agent Patterns: Swarm, Graph, Agent-as-Tool
• Session Management: FileSystem, S3, DynamoDB, AgentCore Memory
• Conversation Managers: SlidingWindow, Summarising
• Hooks: Lifecycle event interception
• Metrics: Automatic tracking (tokens, latency, tools)

---

What is Amazon Bedrock AgentCore?

Enterprise platform providing production infrastructure for deploying and scaling agents.

AgentCore Platform Services:

---

Deployment Architectures

Lambda Serverless (Stateless Agents Only)

When to Use:

• Event-driven workloads (S3, SQS, EventBridge)
• Stateless request/response (< 10 minutes)
• Asynchronous background jobs

When NOT to Use:

• Interactive chat (no streaming)
• Long-running tasks (> 15 minutes)
• Hosting MCP servers (stateful)

Example:

def lambda_handler(event, context):
    tools = MCPRegistry.load_servers(["database-query", "aws-tools"])

    agent = Agent(
        agent_id=None,  # Stateless
        system_prompt="Process this task.",
        tools=tools,
        session_backend=None
    )

    result = agent(event["query"])
    return {"statusCode": 200, "body": json.dumps(result)}

---

ECS/Fargate (MCP Servers)

When to Use:

• Always for MCP servers (24/7 availability)
• Connection pooling to databases, APIs

Why Not Lambda for MCP:

• ❌ Ephemeral (15-minute max)
• ❌ Connection pools don't persist
• ❌ Cold starts add latency

Example:

from mcp.server import FastMCP
import psycopg2.pool

# Persistent connection pool
db_pool = psycopg2.pool.SimpleConnectionPool(minconn=1, maxconn=10, host="db.internal")

mcp = FastMCP("Database Tools")

@mcp.tool()
def query_database(sql: str) -> dict:
    conn = db_pool.getconn()
    try:
        cursor = conn.cursor()
        cursor.execute(sql)
        return {"status": "success", "rows": cursor.fetchall()}
    finally:
        db_pool.putconn(conn)

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

---

AgentCore Runtime (Interactive Agents)

When to Use:

• Long-running tasks (up to 8 hours)
• Real-time streaming required
• Complex multi-agent orchestration
• Enterprise security requirements

Example:

from fastapi import FastAPI

app = FastAPI()

@app.post("/agent/invoke")
async def invoke_agent(request: dict):
    agent = Agent(
        agent_id=request["agent_id"],
        system_prompt=request["system_prompt"],
        tools=load_tools(),
        session_backend="agentcore-memory"
    )

    result = agent(request["input"])
    return {"response": result.message["content"][0]["text"]}

---

Hybrid Architecture (Recommended)

Combine Lambda agents with ECS-hosted MCP servers:

S3/SQS/EventBridge → Lambda Agents → HTTP → ECS MCP Servers
API Gateway → Lambda Agents → HTTP → ECS MCP Servers
Web Client → AgentCore Runtime → HTTP → ECS MCP Servers

---

Agent Execution Flow

1. User Input → Agent
2. Agent → Model (system prompt + tools + context)
3. Model Decision:
   - Generate Response → Return to user
   - Call Tool → Execute → Return to model → Repeat step 2
4. Final Response → User

Metrics Tracked: Token usage, latency, tool statistics, cycle count

---

Session Storage Options

---

Tool Integration Options

Direct MCP Integration

Simple tool requirements, < 10 MCP servers:

from strands.tools.mcp import MCPClient
from mcp import streamablehttp_client

client = MCPClient(lambda: streamablehttp_client("http://mcp:8000/mcp"))
with client:
    tools = client.list_tools_sync()
agent = Agent(tools=tools)

AgentCore Gateway

Multiple protocols, frequent tool changes, centralised governance:

• Protocol abstraction (MCP + Lambda + REST)
• Runtime discovery (dynamic tool loading)
• Automatic authentication

Limitations: OpenAPI specs > 2MB cannot be loaded, discovery adds 50-200ms latency

---

Multi-Agent Patterns

Agent-as-Tool (Simple Delegation)

# Specialist agents
researcher = Agent(system_prompt="Research specialist.", tools=[web_search])
writer = Agent(system_prompt="Content writer.", tools=[grammar_check])

# Wrap as tools
@tool
def research_topic(topic: str) -> str:
    result = researcher(f"Research: {topic}")
    return result.message["content"][0]["text"]

@tool
def write_article(data: str, topic: str) -> str:
    result = writer(f"Write article about {topic} using: {data}")
    return result.message["content"][0]["text"]

# Orchestrator
orchestrator = Agent(
    system_prompt="Coordinate research and writing.",
    tools=[research_topic, write_article]
)

Graph (Deterministic Workflow)

from strands.multiagent import GraphBuilder

builder = GraphBuilder()
builder.add_node("collector", data_collector_agent)
builder.add_node("analyser", analyser_agent)
builder.add_node("reporter", reporter_agent)

builder.add_edge("collector", "analyser")
builder.add_edge("analyser", "reporter")

builder.set_execution_timeout(300)  # 5 minutes
builder.set_max_node_executions(10)

graph = builder.build(entry_point="collector")
result = graph.run({"task": "Analyse Q4 sales data"})

Swarm (Autonomous Collaboration)

from strands.multiagent import Swarm

swarm = Swarm(
    nodes=[researcher, writer, reviewer],
    entry_point=researcher,
    max_handoffs=10,
    execution_timeout=300.0
)

result = swarm.run("Create and review an article")

---

Regional Considerations

Data Residency: Bedrock processes data in-region (Australian data sovereignty, etc.)

Best Practice:

model = BedrockModel(
    model_id="anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name="eu-west-1"  # GDPR-compliant
)

AgentCore Evaluations

LLM-as-a-Judge quality assessment for agents. Monitors AgentCore Runtime endpoints or CloudWatch LogGroups. Integrates with Strands and LangGraph via OpenTelemetry/OpenInference.

Modes: Online (continuous sampling) or on-demand Results: CloudWatch GenAI dashboard, CloudWatch Metrics, configurable alerts

---

Built-in Evaluators

Quality Metrics: Helpfulness, Correctness, Faithfulness, ResponseRelevance, Conciseness, Coherence, InstructionFollowing

Safety Metrics: Refusal, Harmfulness, Stereotyping

Tool Performance: GoalSuccessRate, ToolSelectionAccuracy, ToolParameterAccuracy, ContextRelevance

---

Setup

IAM Role - Execution role needs:

• logs:DescribeLogGroups, logs:GetLogEvents
• bedrock:InvokeModel

Instrumentation - Requires ADOT (same as AgentCore Observability)

---

Configuration

from bedrock_agentcore_starter_toolkit import Evaluation

eval_client = Evaluation()

config = eval_client.create_online_config(
    config_name="my_agent_quality",
    agent_id="agent_myagent-ABC123xyz",
    sampling_rate=10.0,  # Evaluate 10% of interactions
    evaluator_list=["Builtin.Helpfulness", "Builtin.GoalSuccessRate", "Builtin.ToolSelectionAccuracy"],
    enable_on_create=True
)

Data sources:

• Agent endpoint (AgentCore Runtime)
• CloudWatch LogGroups (external agents, requires OTEL service name)

---

Custom Evaluators

custom_eval = eval_client.create_evaluator(
    evaluator_name="CustomerSatisfaction",
    model_id="anthropic.claude-sonnet-4-5-20250929-v1:0",
    evaluation_prompt="""Assess customer satisfaction based on:
1. Query resolution (0-10)
2. Response clarity (0-10)
3. Tone appropriateness (0-10)
Return average score.""",
    level="Agent"  # or "Tool" for tool-level evaluation
)

---

Results

CloudWatch GenAI Dashboard: CloudWatch → GenAI Observability → Evaluations tab

CloudWatch Metrics: AWS/BedrockAgentCore/Evaluations

Alerts:

import boto3
cw = boto3.client('cloudwatch')

cw.put_metric_alarm(
    AlarmName='AgentQualityDegradation',
    MetricName='Helpfulness',
    Namespace='AWS/BedrockAgentCore/Evaluations',
    Statistic='Average',
    Period=3600,
    EvaluationPeriods=2,
    Threshold=7.0,
    ComparisonOperator='LessThanThreshold'
)

Limitations & Considerations

1. Tool Selection at Scale

Issue: Models struggle with > 50-100 tools

Impact: Wrong tool selection, decreased accuracy

Solution: Semantic search for dynamic tool loading (see patterns.md)

Example: AWS internal agent with 6,000 tools uses semantic search

---

2. Token Context Windows

Issue: Long conversations exceed model limits

Limits:

• Claude 4.5: 200K tokens (use ~180K max)
• Nova Pro: 300K tokens (use ~250K max)

Impact: Truncated history, "forgotten" context

Solution:

from strands.agent.conversation_manager import SlidingWindowConversationManager

manager = SlidingWindowConversationManager(max_messages=20, min_messages=2)
agent = Agent(conversation_manager=manager)

---

3. Lambda Streaming

Issue: Lambda doesn't support HTTP response streaming

Impact: No real-time responses, long wait times

Solution: Use AgentCore Runtime for streaming, or implement polling pattern

---

4. Multi-Agent Cost

Issue: Each agent call consumes tokens

Multiplier:

• Agent-as-Tool: 2-3x
• Graph: 3-5x
• Swarm: 5-8x

Impact: Unexpected bills at scale

Solution: Cost tracking hooks, budget alerts, model selection (Haiku for simple tasks)

---

5. Bedrock API Throttling

Issue: ConverseStream API has rate limits

Default: 50-100 TPS (varies by region/account)

Solution: Request quota increases, exponential backoff retry:

def invoke_with_retry(agent: Agent, query: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return agent(query)
        except ClientError as e:
            if e.response['Error']['Code'] == 'ThrottlingException':
                wait = (2 ** attempt) + random.uniform(0, 1)
                time.sleep(wait)
            else:
                raise
    raise Exception("Max retries exceeded")

---

AgentCore Platform Limitations

Runtime Constraints

---

Gateway Limitations

API Spec Size: OpenAPI specs > 2MB cannot be loaded

Workaround: Split into multiple registrations or create facade APIs with only agent-relevant operations

Tool Discovery: Large catalogues (> 50 tools) slow initialisation

Latency: 50-200ms added for discovery

---

Browser Tool Issues

CAPTCHA Blocking: Cannot automate Google, LinkedIn, banking sites

Solution: Use official APIs instead, human-in-the-loop for CAPTCHA sites, or enterprise API partnerships

CORS Errors: Web applications calling AgentCore encounter CORS errors

Solution:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-domain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

---

Memory Service Limitations

Scale Limits: > 100K graph entries degrade performance

Query Latency: 50-200ms per retrieval

Consistency: Eventual, not transactional

Best Practice: Use for high-value data, not transactional state. Use DynamoDB for critical transactional data.

---

Multi-Agent System Challenges

Swarm Pattern Unpredictability

Issue: Swarm agents make autonomous handoff decisions

Symptoms: Agents loop unnecessarily, handoffs don't follow expected paths

Mitigation:

from strands.multiagent import Swarm

swarm = Swarm(
    nodes=[researcher, writer, reviewer],
    entry_point=researcher,
    max_handoffs=10,  # Prevent infinite loops
    execution_timeout=300.0
)

---

Graph Pattern Complexity

Issue: Complex graphs become difficult to maintain

Best Practice: Keep graphs simple (< 10 nodes), document with diagrams, use sub-graphs for complex workflows

---

Cost Accumulation

---

Production Deployment Challenges

Cold Start Latency

Issue: 30-60 seconds for first invocation

Causes: Model loading, MCP client initialisation, dependencies

Solutions:

1. Warm Agent Pools:

class AgentPool:
    def __init__(self, pool_size: int = 5):
        self.agents = queue.Queue(maxsize=pool_size)
        for _ in range(pool_size):
            self.agents.put(BaseAgentFactory.create_agent(...))

    def get_agent(self) -> Agent:
        return self.agents.get()

    def return_agent(self, agent: Agent):
        agent.clear_messages()
        self.agents.put(agent)

2. Lambda Provisioned Concurrency 3. AgentCore Runtime (eliminates cold starts)

---

State Management Complexity

Challenges: Concurrent access to shared sessions, race conditions, state corruption

Solution: DynamoDB with optimistic locking

from strands.session import DynamoDBSessionManager

session_manager = DynamoDBSessionManager(
    table_name="agent-sessions",
    region_name="us-east-1",
    use_optimistic_locking=True
)

---

Observability Gaps

Common Gaps: Why did agent choose specific tool? What was the model's reasoning? Why did multi-agent handoff occur?

Solutions: 1. Structured Logging (see observability.md) 2. Model Reasoning Traces (Claude 4):

model = BedrockModel(
    model_id="anthropic.claude-4-20250228-v1:0",
    enable_thinking=True
)

3. AgentCore Observability (automatic metrics)

---

Security Considerations

Tool Permission Management

Risk: Agents with broad permissions, hallucinations cause unintended actions

Mitigation: Principle of least privilege

@tool
def query_database(sql: str) -> dict:
    # Assume read-only role before executing
    assume_role("arn:aws:iam::account:role/ReadOnlyDatabaseRole")
    # Execute query

---

Data Residency and Compliance

Consideration: LLM providers process data in different regions (GDPR, HIPAA)

Solution: Enforce regional processing

model = BedrockModel(
    model_id="anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name="eu-west-1"  # GDPR-compliant
)

session_manager = DynamoDBSessionManager(
    table_name="agent-sessions",
    region_name="eu-west-1"
)

---

Integration Challenges

Legacy System Integration

Common Issues: APIs lack semantic descriptions, complex multi-step authentication, non-standard data formats

Pattern: Facade for legacy APIs

@tool
def get_customer_data(customer_email: str) -> dict:
    """
    Get customer data from legacy CRM.

    Internally handles session tokens, multi-step API calls, and data transformation.
    """
    session = legacy_crm.authenticate()
    customer = legacy_crm.find_customer(session, email=customer_email)
    orders = legacy_crm.get_orders(session, customer.id)

    return {
        "status": "success",
        "content": [{"text": json.dumps({
            "name": customer.name,
            "orders": [order.to_dict() for order in orders]
        })}]
    }

---

Real-Time Requirements

Limitation: Agents have inherent latency (1-10 seconds)

Not Suitable For: High-frequency trading, real-time control systems, sub-second response requirements

Suitable For: Customer support, content generation, data analysis, workflow automation

---

Summary: Priorities

Must Address

1. Tool Discovery at Scale: Semantic search for > 50 tools 2. Cost Monitoring: Cost tracking from day one 3. Observability: Logging, metrics, tracing 4. Security: Tool-level permissions, human-in-the-loop 5. MCP Servers: Deploy in streamable-http mode, NOT Lambda

Nice to Have

1. Warm Agent Pools: Reduce cold starts 2. Response Caching: Avoid duplicate LLM calls 3. Multi-Region: Deploy close to users

Can Defer

1. Advanced Multi-Agent: Start single agents first 2. Custom Models: Use Bedrock initially 3. Complex Graphs: Begin with linear workflows

Observability & Tracing

Overview

Two observability approaches:

1. AWS AgentCore Observability - Managed service with GenAI dashboard, automatic instrumentation (for agents using AgentCore services) 2. OpenTelemetry - Standard OTLP configuration for self-hosted agents or third-party platforms

For quality assessment, see [evaluations.md](evaluations.md).

---

AWS AgentCore Observability

One-Time Setup: Enable Transaction Search

Via Console: CloudWatch → Application Signals → Transaction Search → Enable (select 1-10% sampling)

Via CLI:

# Configure X-Ray → CloudWatch permissions
aws logs put-resource-policy --policy-name AgentCoreObs --policy-document '{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "xray.amazonaws.com"},
    "Action": "logs:PutLogEvents",
    "Resource": ["arn:aws:logs:REGION:ACCOUNT:log-group:aws/spans:*"]
  }]
}'

# Enable CloudWatch destination
aws xray update-trace-segment-destination --destination CloudWatchLogs

---

Runtime-Hosted Agents (Automatic)

Installation:

pip install 'strands-agents[otel]'
echo "aws-opentelemetry-distro" >> requirements.txt

Agent code (automatic instrumentation):

from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()
agent = Agent(
    model=BedrockModel(model_id="anthropic.claude-sonnet-4-5-20250929-v1:0"),
    tools=[weather_tool],
    system_prompt="You are helpful."
)

@app.entrypoint
def handler(payload):
    return agent(payload["prompt"]).message['content'][0]['text']

Deploy:

from bedrock_agentcore_starter_toolkit import Runtime
runtime = Runtime()
runtime.configure(entrypoint="agent.py", requirements_file="requirements.txt")
runtime.launch()

Traces appear automatically in GenAI Observability Dashboard.

---

Non-Runtime Hosted (Self-Hosted with AgentCore Observability)

Environment variables:

export AGENT_OBSERVABILITY_ENABLED=true
export OTEL_PYTHON_DISTRO=aws_distro
export OTEL_PYTHON_CONFIGURATOR=aws_configurator
export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
export OTEL_RESOURCE_ATTRIBUTES="service.name=my-agent"

Run:

pip install strands-agents[otel] aws-opentelemetry-distro
opentelemetry-instrument python agent.py

---

Session Tracking

Associate traces across multi-turn conversations:

from opentelemetry import baggage, context

session_id = "user-123"
ctx = baggage.set_baggage("session.id", session_id)

with context.attach(ctx):
    response = agent("First question")
    response = agent("Follow-up question")

---

Viewing Data

GenAI Dashboard: CloudWatch → GenAI Observability → Bedrock AgentCore tab

• Agents View: All agents, metrics per agent
• Sessions View: Browse sessions, filter by agent/date
• Traces View: Inspect trajectories and timelines

CloudWatch Logs:

• Runtime logs: /aws/bedrock-agentcore/runtimes/<agent_id>-<endpoint>/[runtime-logs]
• OTEL logs: /aws/bedrock-agentcore/runtimes/<agent_id>-<endpoint>/otel-rt-logs

Transaction Search: CloudWatch → Transaction Search → /aws/spans/default

---

OpenTelemetry (General)

Setup Options

Recommended: StrandsTelemetry Fluent API

from strands.observability import StrandsTelemetry

# Development: Console output
telemetry = StrandsTelemetry().setup_console_exporter()

# Production: OTLP endpoint (reads OTEL_EXPORTER_OTLP_ENDPOINT from env)
telemetry = StrandsTelemetry().setup_otlp_exporter()

# Both
telemetry = StrandsTelemetry() \
    .setup_console_exporter() \
    .setup_otlp_exporter() \
    .setup_meter(enable_otlp_exporter=True)

Environment Variables:

export OTEL_EXPORTER_OTLP_ENDPOINT="http://collector:4318"
export OTEL_EXPORTER_OTLP_HEADERS="x-api-key=your-key"

---

Custom Attributes

agent = Agent(
    system_prompt="...",
    tools=[...],
    trace_attributes={
        "environment": "production",
        "customer_tier": "enterprise",
        "region": "us-east-1"
    }
)

---

Automatic Metrics

result = agent("Query")
metrics = result.metrics

# Token usage
print(metrics.accumulated_usage['totalTokens'])
print(metrics.accumulated_usage['inputTokens'])
print(metrics.accumulated_usage['outputTokens'])

# Performance
print(metrics.cycle_count)
print(metrics.cycle_durations)

# Tool usage
for tool_name, data in metrics.tool_metrics.items():
    print(f"{tool_name}: {data['call_count']} calls")
    print(f"  Success: {data['success_count'] / data['call_count']:.1%}")
    print(f"  Avg duration: {data['total_duration'] / data['call_count']:.2f}s")

---

Custom Observability Hooks

Production Hook Pattern

from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeToolCallEvent, AfterToolCallEvent, AfterInvocationEvent
import logging, time

class ObservabilityHook(HookProvider):
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.logger = logging.getLogger(f"agent.{agent_name}")
        self.tool_timings = {}

    def register_hooks(self, registry: HookRegistry, **kwargs):
        registry.add_callback(BeforeToolCallEvent, self.before_tool)
        registry.add_callback(AfterToolCallEvent, self.after_tool)
        registry.add_callback(AfterInvocationEvent, self.log_completion)

    def before_tool(self, event: BeforeToolCallEvent):
        tool_id = event.tool_use["toolUseId"]
        self.tool_timings[tool_id] = time.time()
        self.logger.info("Tool invoked", extra={
            "tool": event.tool_use["name"],
            "input": event.tool_use["input"]
        })

    def after_tool(self, event: AfterToolCallEvent):
        tool_id = event.tool_use["toolUseId"]
        duration = time.time() - self.tool_timings[tool_id]

        self.logger.info("Tool completed", extra={
            "tool": event.tool_use["name"],
            "duration_ms": duration * 1000,
            "status": event.result.get("status")
        })

        if duration > 5.0:
            self.logger.warning(f"Slow tool: {event.tool_use['name']} took {duration:.2f}s")

    def log_completion(self, event: AfterInvocationEvent):
        metrics = event.result.metrics.get_summary()
        self.logger.info("Agent completed", extra={
            "cycles": metrics["total_cycles"],
            "tokens": metrics["accumulated_usage"]["totalTokens"]
        })

# Usage
agent = Agent(hooks=[ObservabilityHook("my-agent")])

---

Cost Tracking Hook

from strands.hooks import HookProvider
from strands.hooks.events import AfterInvocationEvent
import boto3

PRICING = {"input": 0.003 / 1000, "output": 0.015 / 1000}

class CostTrackingHook(HookProvider):
    def __init__(self, budget_limit: float = 100.0):
        self.budget_limit = budget_limit
        self.total_cost = 0.0
        self.cloudwatch = boto3.client('cloudwatch')

    def register_hooks(self, registry, **kwargs):
        registry.add_callback(AfterInvocationEvent, self.track_cost)

    def track_cost(self, event):
        usage = event.result.metrics.accumulated_usage
        cost = (usage["inputTokens"] * PRICING["input"] +
                usage["outputTokens"] * PRICING["output"])
        self.total_cost += cost

        self.cloudwatch.put_metric_data(
            Namespace='AgentCosts',
            MetricData=[{
                'MetricName': 'InvocationCost',
                'Value': cost,
                'Unit': 'None'
            }]
        )

        if self.total_cost > self.budget_limit * 0.9:
            logger.warning(f"Budget: ${self.total_cost:.2f} of ${self.budget_limit:.2f}")

agent = Agent(hooks=[CostTrackingHook(budget_limit=100.0)])

---

Third-Party Platforms

Arize Phoenix

import phoenix as px
from phoenix.trace.opentelemetry import OpenInferenceTracer

session = px.launch_app()
tracer = OpenInferenceTracer()

from strands.observability import StrandsTelemetry
telemetry = StrandsTelemetry(tracer_provider=tracer.tracer_provider)
# View at http://localhost:6006

Langfuse

from langfuse.opentelemetry import LangfuseSpanExporter
from opentelemetry.sdk.trace import TracerProvider, SimpleSpanProcessor

exporter = LangfuseSpanExporter(
    public_key="pk-xxx",
    secret_key="sk-xxx",
    host="https://cloud.langfuse.com"
)
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))

from strands.observability import StrandsTelemetry
telemetry = StrandsTelemetry(tracer_provider=provider)

---

Production Checklist

AgentCore Platform

• [ ] Transaction Search enabled
• [ ] ADOT installed (aws-opentelemetry-distro)
• [ ] Session tracking enabled
• [ ] Sampling configured (1-10%)

Essential

• [ ] OpenTelemetry tracing enabled
• [ ] Cost tracking implemented
• [ ] CloudWatch dashboards created
• [ ] Error alerting configured

Metrics

• [ ] Latency tracked (p50, p90, p99)
• [ ] Token usage monitored
• [ ] Tool success rate tracked
• [ ] Error rate alerts (> 2%)

Security

• [ ] Sensitive data redacted from traces
• [ ] Access logs enabled
• [ ] Retention policies configured

Implementation Patterns & Best Practices

Foundation Component Patterns

Pattern 1: Base Agent Factory

Build reusable agent factories with organisational defaults:

# foundation/agent_factory.py
from strands import Agent
from strands.models import BedrockModel
from strands.session import DynamoDBSessionManager
from strands.agent.conversation_manager import SlidingWindowConversationManager
import os

class BaseAgentFactory:
    """Platform standard agent factory."""

    @staticmethod
    def create_agent(agent_id: str, system_prompt: str, tools: list) -> Agent:
        """Create agent with organisational defaults."""
        return Agent(
            agent_id=agent_id,
            model=BedrockModel(
                model_id=os.getenv("DEFAULT_MODEL_ID", "anthropic.claude-sonnet-4-5-20250929-v1:0"),
                region_name=os.getenv("AWS_REGION", "us-east-1")
            ),
            system_prompt=system_prompt,
            tools=tools,
            session_manager=DynamoDBSessionManager(table_name=os.getenv("SESSION_TABLE")),
            conversation_manager=SlidingWindowConversationManager(max_messages=20)
        )

# Usage
agent = BaseAgentFactory.create_agent(
    agent_id="customer-support",
    system_prompt="You are helpful.",
    tools=[tool1, tool2]
)

---

Pattern 2: MCP Server Registry

Organisation-wide MCP server catalogue:

# foundation/mcp_loader.py
from strands.tools.mcp import MCPClient
from mcp import streamablehttp_client

class MCPRegistry:
    """Load MCP servers from catalogue."""

    MCP_ENDPOINTS = {
        "database-query": "http://mcp-database.internal:8000/mcp",
        "aws-tools": "http://mcp-aws-tools.internal:8000/mcp",
        "notification": "http://mcp-notification.internal:8000/mcp"
    }

    @staticmethod
    def load_servers(server_names: list[str]) -> list:
        """Load tools from specified MCP servers."""
        all_tools = []
        for name in server_names:
            endpoint = MCPRegistry.MCP_ENDPOINTS[name]
            client = MCPClient(lambda e=endpoint: streamablehttp_client(e))
            with client:
                all_tools.extend(client.list_tools_sync())
        return all_tools

# Usage
tools = MCPRegistry.load_servers(["database-query", "aws-tools"])
agent = BaseAgentFactory.create_agent("support-agent", "You are helpful.", tools)

---

Pattern 3: Semantic Tool Search (> 50 Tools)

For large tool sets, use semantic search to dynamically load relevant tools:

# foundation/tool_search.py
from sentence_transformers import SentenceTransformer
import numpy as np

class DynamicToolLoader:
    """Load relevant tools based on query."""

    def __init__(self, all_tools: list):
        self.all_tools = all_tools
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.tool_embeddings = self.model.encode([tool.__doc__ for tool in all_tools])

    def get_relevant_tools(self, query: str, top_k: int = 10) -> list:
        """Find top-k relevant tools."""
        query_embedding = self.model.encode([query])
        similarities = np.dot(self.tool_embeddings, query_embedding.T).flatten()
        top_indices = np.argsort(similarities)[-top_k:]
        return [self.all_tools[i] for i in top_indices]

# Usage
tool_loader = DynamicToolLoader(all_organisational_tools)  # 100+ tools
relevant_tools = tool_loader.get_relevant_tools(user_query, top_k=10)
agent = Agent(tools=relevant_tools)  # Only 10 tools, not 100+

Why: Models struggle with > 50-100 tools. AWS internal agents with 6,000 tools use semantic search.

---

Tool Design Patterns

Rule 1: User-Task Oriented (Not API-Oriented)

# ❌ BAD: API-style granularity
@tool
def get_user_by_id(user_id: str) -> dict:
    """Get user by ID."""
    pass

@tool
def get_user_orders(user_id: str) -> list:
    """Get orders for user."""
    pass

# ✅ GOOD: Task-oriented
@tool
def get_customer_profile(customer_email: str) -> dict:
    """
    Get complete customer profile including orders, preferences, and history.

    Args:
        customer_email: Customer's email address

    Returns:
        Comprehensive customer data
    """
    user = _get_user_by_email(customer_email)
    orders = _get_user_orders(user["id"])
    preferences = _get_user_preferences(user["id"])

    return {"user": user, "orders": orders, "preferences": preferences}

---

Rule 2: Structured Error Handling

Always return structured results, never raise exceptions to the model.

# ✅ GOOD
@tool
def query_database(sql: str) -> dict:
    """Execute SQL query and return results."""
    try:
        results = database.execute(sql)
        return {"status": "success", "content": [{"text": json.dumps(results)}]}
    except DatabaseError as e:
        return {"status": "error", "content": [{"text": f"Query failed: {str(e)}"}]}

---

Rule 3: Response Size Management

For large datasets, use pagination:

@tool
def query_large_dataset(query: str, page: int = 1, page_size: int = 10) -> dict:
    """Query dataset with pagination."""
    results = database.query(query, offset=(page-1)*page_size, limit=page_size)

    return {
        "status": "success",
        "content": [{"text": json.dumps(results)}],
        "pagination": {
            "page": page,
            "page_size": page_size,
            "has_more": len(results) == page_size
        }
    }

---

Security Patterns

Tool-Level Permissions

from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry

TOOL_PERMISSIONS = {
    "delete_database": "admin:delete_records",
    "send_company_email": "user:send_email",
    "query_database": "user:read_data"
}

class PermissionValidator(HookProvider):
    def __init__(self, user_permissions: list[str]):
        self.user_permissions = user_permissions

    def register_hooks(self, registry: HookRegistry, **kwargs):
        registry.add_callback(BeforeToolCallEvent, self.validate_permissions)

    def validate_permissions(self, event: BeforeToolCallEvent):
        tool_name = event.tool_use["name"]
        required = TOOL_PERMISSIONS.get(tool_name)
        if required and required not in self.user_permissions:
            event.cancel_tool = f"Permission denied: {required} required"

agent = Agent(
    tools=[delete_database, query_database],
    hooks=[PermissionValidator(["user:read_data"])]
)

---

Human-in-the-Loop for Sensitive Actions

class ApprovalHook(HookProvider):
    SENSITIVE_TOOLS = ["delete_database", "send_company_email", "transfer_funds"]

    def register_hooks(self, registry: HookRegistry, **kwargs):
        registry.add_callback(BeforeToolCallEvent, self.require_approval)

    def require_approval(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] in self.SENSITIVE_TOOLS:
            approval = event.interrupt("approval-required", reason={
                "action": event.tool_use["name"],
                "params": event.tool_use["input"]
            })
            if approval.lower() != "approved":
                event.cancel_tool = "Action denied by user"

agent = Agent(hooks=[ApprovalHook()], tools=[delete_database])

---

Performance Optimisation

Concurrent Tool Execution

from strands.tools.executors import ConcurrentToolExecutor

agent = Agent(
    tools=[fetch_api_data, query_database, check_cache],
    tool_executor=ConcurrentToolExecutor()
)

When to Use: Tools are I/O bound, thread-safe, order doesn't matter

When to Avoid: Tools modify shared state or depend on each other

---

Tool Caching

from functools import lru_cache

@tool
@lru_cache(maxsize=100)
def get_product_catalogue() -> dict:
    """Get product catalogue (cached)."""
    return database.get_all_products()

---

Conversation Management

For Short Sessions (< 10 Exchanges)

from strands.agent.conversation_manager import SlidingWindowConversationManager

manager = SlidingWindowConversationManager(max_messages=15, min_messages=2)
agent = Agent(conversation_manager=manager)

---

For Long Sessions (Need History)

from strands.agent.conversation_manager import SummarizingConversationManager

manager = SummarizingConversationManager(max_messages=30, summarize_messages_count=25)
agent = Agent(conversation_manager=manager)

---

Testing Patterns

Unit Testing Tools

from myapp.tools import analyse_data

def test_analyse_data():
    result = analyse_data(data="sample data")
    assert result["status"] == "success"
    assert "insights" in result["content"][0]["text"]

---

Integration Testing Agents

from foundation.agent_factory import BaseAgentFactory
from unittest.mock import Mock

def test_agent_with_mocked_tools():
    mock_tool = Mock(return_value={
        "status": "success",
        "content": [{"text": "mocked result"}]
    })

    agent = BaseAgentFactory.create_agent(
        agent_id="test-agent",
        system_prompt="You are a test agent.",
        tools=[mock_tool]
    )

    result = agent("Test query")
    assert mock_tool.called
    assert "mocked result" in str(result.message)