AWS Bedrock AgentCore: Production Patterns for Enterprise AI Agents

from strands import Swarm, Agent

# Define specialized agents
researcher = Agent(
    name="researcher",
    model="anthropic.claude-sonnet-4",
    instructions="You research and gather information."
)

analyst = Agent(
    name="analyst",
    model="anthropic.claude-sonnet-4",
    instructions="You analyze data and find patterns."
)

# Create swarm
research_swarm = Swarm(
    agents=[researcher, analyst],
    aggregation="consensus"  # or "vote", "merge"
)

# Execute
result = research_swarm.run("Analyze Q4 sales performance")

from strands import AgentGraph, Edge

graph = AgentGraph()

# Add nodes
graph.add_agent(intake_agent, node_id="intake")
graph.add_agent(research_agent, node_id="research")
graph.add_agent(write_agent, node_id="write")
graph.add_agent(review_agent, node_id="review")

# Add edges with conditions
graph.add_edge("intake", "research")
graph.add_edge("research", "write", condition=lambda s: len(s.sources) >= 3)
graph.add_edge("write", "review")
graph.add_edge("review", "write", condition=lambda s: s.needs_revision)
graph.add_edge("review", "END", condition=lambda s: s.approved)

# Execute
result = graph.run("Create a market analysis report")

from strands import Workflow, Step

workflow = Workflow([
    Step("gather", research_agent),
    Step("analyze", analyst_agent),
    Step("synthesize", writer_agent),
    Step("validate", reviewer_agent),
])

result = workflow.run("Research quantum computing trends")

from bedrock_agentcore import Agent, Memory

agent = Agent(
    model_id="anthropic.claude-sonnet-4",
    memory=Memory(
        # Short-term: conversation context
        short_term_enabled=True,
        short_term_window=10,  # Last 10 messages

        # Long-term: semantic extraction
        long_term_enabled=True,
        long_term_extraction="auto",  # Auto-extract facts

        # Session persistence
        session_id="user-123-session-456",
        persistence="dynamodb"  # or "s3"
    )
)

# After conversation, memory contains:
{
    "user_preferences": {
        "communication_style": "technical",
        "timezone": "PST",
        "role": "engineering_manager"
    },
    "learned_facts": [
        "User's team uses Python for backend",
        "Deployment target is EKS",
        "Budget constraint: $10k/month"
    ],
    "interaction_patterns": {
        "peak_hours": "9am-11am",
        "avg_session_length": "15min"
    }
}

from bedrock import Guardrails

guardrails = Guardrails(
    # Content filters
    content_policy={
        "hate": {"strength": "HIGH"},
        "insults": {"strength": "MEDIUM"},
        "sexual": {"strength": "HIGH"},
        "violence": {"strength": "HIGH"},
    },

    # Topic blocking
    denied_topics=[
        "competitor_products",
        "internal_financials",
        "unreleased_features"
    ],

    # PII handling
    pii_config={
        "action": "ANONYMIZE",  # or "BLOCK"
        "types": ["EMAIL", "PHONE", "SSN", "CREDIT_CARD"]
    },

    # Word filters
    word_filters={
        "profanity": "BLOCK",
        "custom_words": ["secret_project", "codename_x"]
    },

    # Grounding check (hallucination prevention)
    grounding={
        "enabled": True,
        "threshold": 0.7,
        "source_required": True
    }
)

# Attach to agent
agent = Agent(
    model_id="anthropic.claude-sonnet-4",
    guardrails=guardrails
)

from bedrock_agentcore import Agent, CloudWatchMetrics

agent = Agent(
    model_id="anthropic.claude-sonnet-4",
    metrics=CloudWatchMetrics(
        namespace="MyApp/Agents",
        dimensions={"AgentName": "ResearchAssistant"},
        metrics=[
            "InvocationCount",
            "Latency",
            "TokensUsed",
            "ErrorRate",
            "GuardrailTriggered"
        ]
    )
)

from aws_xray_sdk.core import xray_recorder

@xray_recorder.capture("agent_execution")
async def run_agent(query: str):
    with xray_recorder.in_subsegment("tool_calls"):
        result = await agent.run(query)
    return result

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["bedrock:InvokeModel", "bedrock:InvokeAgent"],
      "Resource": "arn:aws:bedrock:*:*:agent/*"
    },
    {
      "Effect": "Allow",
      "Action": ["dynamodb:GetItem", "dynamodb:Query"],
      "Resource": "arn:aws:dynamodb:*:*:table/AgentMemory",
      "Condition": {
        "ForAllValues:StringEquals": {
          "dynamodb:LeadingKeys": ["${aws:PrincipalTag/AgentId}"]
        }
      }
    }
  ]
}

from bedrock_agentcore import Tool, Permission

# Define tool with explicit permissions
database_tool = Tool(
    name="query_database",
    description="Query the sales database",
    permissions=Permission(
        allowed_tables=["sales", "products"],
        denied_tables=["users", "credentials"],
        row_filter="department = ${user.department}",
        max_rows=1000
    )
)

from bedrock_agentcore import Agent
import boto3

# Use Secrets Manager for API keys
secrets = boto3.client('secretsmanager')

agent = Agent(
    model_id="anthropic.claude-sonnet-4",
    tools=[
        Tool(
            name="external_api",
            api_key=secrets.get_secret_value(
                SecretId="prod/external-api-key"
            )["SecretString"]
        )
    ]
)

# Reserve 100 model units for Claude Sonnet
# Saves ~30% vs on-demand for consistent usage

# System prompt (cached after first call)
SYSTEM_PROMPT = """You are a research assistant...
[2000 tokens of instructions]
"""

# User-specific context (not cached)
user_context = f"User: {user.name}, Role: {user.role}"

# This call benefits from caching
response = agent.invoke(
    system=SYSTEM_PROMPT,  # Cached
    context=user_context,  # Not cached
    query=user_query       # Not cached
)

from bedrock_agentcore import Agent, TokenBudget

agent = Agent(
    model_id="anthropic.claude-sonnet-4",
    token_budget=TokenBudget(
        max_input_tokens=10000,
        max_output_tokens=4000,
        max_total_per_session=50000,
        on_exceed="graceful_stop"  # or "error"
    )
)

# buildspec.yml
version: 0.2
phases:
  install:
    commands:
      - pip install -r requirements.txt
  test:
    commands:
      - pytest tests/agent_tests.py -v
      - python scripts/evaluate_agent.py --threshold 0.85
  deploy:
    commands:
      - aws bedrock update-agent --agent-id $AGENT_ID --agent-config file://agent-config.json

# Deploy new version alongside existing
new_agent = deploy_agent(config, version="v2")

# Gradual traffic shift
traffic_config = {
    "v1": 90,  # 90% to existing
    "v2": 10   # 10% to new
}

# Monitor metrics, then shift
if new_agent.metrics.error_rate < 0.01:
    traffic_config = {"v1": 0, "v2": 100}

# Automatic rollback on error spike
if agent.metrics.error_rate > 0.05:
    rollback_to_version("v1")
    alert_team("Agent v2 rolled back due to high error rate")

"""
Research Assistant Agent - AWS Bedrock/AgentCore Implementation
Demonstrates all 7 pillars of production agent systems.
"""

import boto3
from bedrock_agentcore import Agent, Tool, Memory, Guardrails
from strands import Workflow, Step

# Pillar 5: Security - IAM-based configuration
session = boto3.Session(profile_name="production")

# Pillar 3: Guardrails
guardrails = Guardrails(
    content_policy={"hate": {"strength": "HIGH"}},
    pii_config={"action": "ANONYMIZE"},
    grounding={"enabled": True, "threshold": 0.7}
)

# Pillar 2: Memory
memory = Memory(
    short_term_enabled=True,
    long_term_enabled=True,
    persistence="dynamodb",
    table_name="AgentMemory"
)

# Pillar 1: Orchestration - Tools
search_tool = Tool(
    name="web_search",
    description="Search the web for information",
    handler=lambda q: search_api.search(q)
)

fetch_tool = Tool(
    name="fetch_url",
    description="Fetch content from a URL",
    handler=lambda url: fetch_and_parse(url)
)

# Create agent
agent = Agent(
    model_id="anthropic.claude-sonnet-4",
    instructions="You are a research assistant...",
    tools=[search_tool, fetch_tool],
    memory=memory,
    guardrails=guardrails,
    # Pillar 6: Cost management
    token_budget={"max_total": 50000},
    # Pillar 4: Observability
    metrics_namespace="ResearchApp/Agents"
)

# Pillar 7: Lifecycle - Versioning
agent.version = "1.2.0"
agent.deploy(environment="production")