Building AI Agents with Strands and AWS Bedrock AgentCore Runtime
So in the previous tutorial, we talked about how we can build agentic framework using ADK, Google ADK. And in this tutorial, we will expand the same thing and understand how we can do the same in Strands Agent. This is just to give you a contrast between how different agentic frameworks work and how you can pick one based on your requirements and understand AWS agent code runtime. So let's get started.
›AWS Bedrock AgentCore Features
Here's the list of top features provided by AWS Bedrock AgentCore:
✅ Top 6 Features of AWS Bedrock AgentCore
›1. Agent Runtime (Serverless, Secure Execution)
- Fully managed runtime for deploying agents with no infra management.
- Supports low-latency real-time interactions and 8-hour long-running async tasks.
- Provides complete session isolation to prevent data leakage.
›2. Gateway (Easy Tool & API Integration)
- Converts your APIs, Lambdas, and MCP servers into agent-ready tools with minimal code.
- Adds semantic search so agents can discover and pick the right tool intelligently.
›3. Memory (Short- & Long-term Context Persistence)
- Agents can remember across interactions automatically.
- Gives full control over what gets stored (no hidden memory).
- Zero infrastructure needed for storage or retrieval.
›4. Identity (Secure Access & Permissions)
- Provides agent identity so agents can securely access AWS resources or third-party APIs.
- Supports user-delegated access and pre-authorized consent flows.
- Ensures fine-grained IAM-level control.
›5. Observability (Monitoring, Debugging, Compliance)
- Built-in dashboards to trace agent behavior, steps, tool calls, and failures.
- OpenTelemetry compatible for integrating with Grafana, Datadog, etc.
- Helps with compliance, debugging, and production visibility.
›6. Built-in Tools (Code Interpreter + Browser Runtime)
- Code Interpreter: run Python/JS/etc securely in sandboxed environments.
- Browser Tool: serverless, fast browser automation so agents can navigate web apps.
- Enables rich workflows like scraping, form filling, or data visualization.
With that being said, let's quickly look how everything sticks together here.
We would use the same example what we used in the previous adk tutorial.
Prerequisites
- uv package manager (You can install it here)
- Python 3.13+ (Other versions may work but would recommend to be on Python 3.13+)
- AWS Agents Access (You may need AWS account with AWS Bedrock access for this. You can get access here)
- Previous Blog about ADK Agents (You can find it here) for understanding folder structure and all.
So we can set our agents directory like this:
├── strands_agents
│ ├── README.md
│ └── agents
Inside agents we will create agents and inside strands_agents we will have our runner file - run_agent.py.
You can use uv to create a virtual environment and install strands package:
uv add bedrock-agentcore boto3 google-adk pydantic uuid6 strands-agents strands-agents-tools
›Defining a model to be used by the agent
First let's start by defining a model. This is going to be the brain of our agent.
Create a new file model.py inside aws_strands_agents folder.
from strands.models import BedrockModel
AGENT_MODEL = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
orchestrator_model = BedrockModel(model_id=AGENT_MODEL, streaming=True)
Claude Haiku 4.5 is a great model for agentic workloads since it has been optimized for following complex instructions and has better reasoning capabilities. It is also very inexpensive compared to other anthropic models.
Once done, you can create your agent file inside agents folder. Let's call it news_agent.py.
Since we don't have inbuilt google search tool we will keep our agent simple for understanding purpose. Here's how our news_agent.py would look like:
from strands import Agent
from aws_strands_agents.model import orchestrator_model
news_agent = Agent(
model=orchestrator_model,
system_prompt="You need to provide accurate and concise news summaries based on the latest information available.",
callback_handler=None
)
Please note this is example and agent could return made up data or even refuse to answer since it doesn't have access to latest news data.
Now we need to write our agent runner:
from typing import Any
import traceback
from bedrock_agentcore import BedrockAgentCoreApp
from agents.news_agent import news_agent
app = BedrockAgentCoreApp()
@app.entrypoint # type: ignore[misc]
async def invoke(event: dict[str, Any]): # type: ignore[misc]
"""Entrypoint for Bedrock AgentCoreApp to invoke the agent."""
try:
user_message = event.get("inputText") or event.get("prompt", "")
# For streaming response, use agent.stream_async() and yield events
stream = news_agent.stream_async(user_message)
async for event in stream:
print("Streaming event: %s", event)
yield event
except Exception as e:
error_traceback = traceback.format_exc()
print(f"Agent invocation failed: {str(e)}\nTraceback:\n{error_traceback}")
# Yield error event instead of return in async generator
yield {
"error_message": f"Agent invocation failed: {str(e)}",
"tool_payload": {},
"summarised_markdown": "",
"suggested_next_actions": []
}
if __name__ == "__main__":
print("🤖 Compliance Copilot Agent starting on port 8080...")
print("📋 System prompt loaded - Agent will intelligently route queries")
print("🔧 Available tools: compliance_check, comprehensive_check, doc_status, show_doc")
print("🧠 Agent Mode: Smart parameter extraction from user prompts")
app.run() # type: ignore[misc]
The above code is just to show how you can integrate strands agents with AWS Bedrock AgentCore runtime.
For better explanation of the code, check out the video tutorial linked in the beginning of this blog.
Setting up Pre and Post Agent Hooks
Now lets create another agent to show usage of hooks in strands agents. Similar to ADK its fairly simple to set up hooks in strands agents.
Here's an example agent:
from strands import Agent
from model import orchestrator_model
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import AfterInvocationEvent, AfterToolCallEvent, MessageAddedEvent
from typing import Any
class CleanupHook(HookProvider):
"""Hook to validate input and clean up model responses."""
def register_hooks(self, registry: HookRegistry) -> None: # type: ignore
registry.add_callback(MessageAddedEvent, self.validate_user_input)
registry.add_callback(AfterInvocationEvent, self.after_model_call)
registry.add_callback(AfterToolCallEvent, self.after_tool_call)
def validate_user_input(self, event: MessageAddedEvent) -> None:
"""Validate user input before the agent processes it."""
# Check if this is a user message
if event.message.get('role') == 'user':
content_blocks = event.message.get('content', [])
for content_block in content_blocks:
if "text" in content_block:
text_upper = content_block["text"].upper()
# Check if uppercase text contains "DEATH"
if "DEATH" in text_upper:
raise ValueError("Topic contains prohibited content: 'death' is not allowed")
def after_tool_call(self, event: AfterToolCallEvent) -> None:
"""No-op for tool calls."""
print("tool result received")
print("------------------------------------------------------------")
def after_model_call(self, event: AfterInvocationEvent) -> None:
"""
Modify the model's response after it completes generation.
This fires after EVERY model call (including tool-use and final response).
"""
complete_message:dict[str, Any] = event.agent.messages[-1] if event.agent.messages else {} # type: ignore
if complete_message['role'] != 'assistant':
return # Only process assistant messages
messages: list[dict[str, Any]] = complete_message.get('content', [])
if isinstance(messages, list): # type: ignore
for each_message in messages:
if "text" in each_message and isinstance(each_message["text"], str):
each_message["text"] = each_message["text"].strip().replace("```json", "").replace("```", "")
jokes_agent = Agent(
model=orchestrator_model,
system_prompt="You share jokes and funny stories to lighten the mood of users on specific topic. Share a joke with the user based on the topic they provide.",
hooks=[CleanupHook()]
)
Now let's break down the code.
›Agent Hook Lifecycle
The agent uses hooks to validate input and clean responses at different stages:
User Input → MessageAddedEvent → Model Call → AfterInvocationEvent → Response
↓ ↓
validate_user_input() after_model_call()
›1. CleanupHook Class
Custom hook provider that registers three callbacks:
| Hook Event | When It Fires | Purpose |
| ---------------------- | ------------------------------ | ---------------------------- |
| MessageAddedEvent | When message is added to agent | Validate user input |
| AfterInvocationEvent | After model generates response | Clean up response formatting |
| AfterToolCallEvent | After tool execution | Log tool usage |
›2. Input Validation (validate_user_input)
What it does: Blocks prohibited topics before processing
# 1. Check if message role is 'user'
# 2. Extract text from content blocks
# 3. Convert text to uppercase
# 4. Check if "DEATH" is present
# 5. Raise error if found → stops agent execution
Example:
- ✅ "Tell me a joke about cats" → Passes
- ❌ "Tell me about death metal" → Blocked
›3. Response Cleanup (after_model_call)
What it does: Removes markdown code blocks from AI responses
# 1. Get the last assistant message
# 2. Strip markdown formatting (```json, ```)
# 3. Clean whitespace
Example:
- Before: "
json\n{\"joke\": \"Why...\"}\n" - After: "{"joke": "Why..."}"
›4. Tool Logging (after_tool_call)
Simple callback that prints when tools return results (for debugging).
›Key Takeaways
- MessageAddedEvent = Input validation gate (triggers when messages are added)
- AfterInvocationEvent = Output formatter (triggers after agent completes)
- Hooks run automatically on every agent interaction
- ValueError in hooks stops execution immediately
Now let's make our run_agent.py to use this new agent with hooks:
from typing import Any
import traceback
from bedrock_agentcore import BedrockAgentCoreApp
from agents.safety_agent import jokes_agent
app = BedrockAgentCoreApp()
@app.entrypoint # type: ignore[misc]
async def invoke(event: dict[str, Any]): # type: ignore[misc]
"""Entrypoint for Bedrock AgentCoreApp to invoke the agent."""
try:
user_message = event.get("inputText") or event.get("prompt", "")
# For streaming response, use agent.stream_async() and yield events
stream = jokes_agent.stream_async(user_message)
async for event in stream:
# print("Streaming event: %s", event)
yield event
except Exception as e:
error_traceback = traceback.format_exc()
print(f"Agent invocation failed: {str(e)}\nTraceback:\n{error_traceback}")
# Yield error event instead of return in async generator
yield {
"error_message": f"Agent invocation failed: {str(e)}",
"tool_payload": {},
"summarised_markdown": "",
"suggested_next_actions": []
}
if __name__ == "__main__":
print("🤖 Compliance Copilot Agent starting on port 8080...")
print("📋 System prompt loaded - Agent will intelligently route queries")
print("🔧 Available tools: compliance_check, comprehensive_check, doc_status, show_doc")
print("🧠 Agent Mode: Smart parameter extraction from user prompts")
app.run() # type: ignore[misc]
Notice the only part we changed is the import statement and the agent instance used. Rest everything remains same.
And that's a wrap! This was a short tutorial on how you can build agents using AWS Strands Agents framework. You can explore more about strands agents and its capabilities in the official documentation.
Additionally for deploying the strands agents with AWS Bedrock AgentCore, its pretty simple and one liner: uv run agentcore launch and you're good to go! For more details, check out the official documentation.
