Tools: Build Production-ready AI Agents with AWS Bedrock & Agentcore

What We're Building: The Product Hunt Launch Assistant ## The AWS AI Agent Stack ## AI Agent Architecture ## 1. The Agent Layer (Strands SDK) ## 2. Custom Tools with the @tool Decorator ## 3. The Memory System (AgentCore Memory) ## Building the API Layer ## Lessons Learned: What I'd Do Differently ## 1. Start with AgentCore Runtime for Production ## 2. Use Infrastructure as Code ## 3. Build Modular Tools ## 4. Plan for Multi-Agent Systems So you've heard about AI agents, right? They're everywhere now… automating workflows, answering customer queries, and even planning product launches. But here's the thing: building one that actually works in production is a whole different game compared to throwing together a ChatGPT wrapper. I recently built an "AI-powered Product Hunt launch assistant" during the AWS AI Agent Hackathon at AWS Startup Loft, Tokyo. And honestly? It taught me a ton about what it takes to build production-ready AI agents on AWS. In this article, I'll try to walk you through the process of how to build on AWS, think about the architecture, the tools, and share the lessons that I learnt, so you can build your own AI agent-based projects without the trial-and-error pain. The hackathon crew at AWS AI Agent Hackathon Before diving into the tech, let me give you context. The Product Hunt Launch Assistant is an AI agent that helps entrepreneurs plan and execute their Product Hunt launches. It can: The interface to prepare your project info to launch The chat interface with real-time streaming responses The cool part? All of this runs on AWS Bedrock using the Strands Agents SDK and AgentCore Memory. Let me break down how it all works. If you wanna follow along or check out the code, head over to Github. The core components that comprise the stack are going to be: Let's look at how the Product Hunt Launch Assistant is structured: Full system architecture The heart of the application is the ProductHuntLaunchAgent class, where we create the agent. What I love about Strands is how minimal the code is. You give it: And that's it! The framework handles all the reasoning, tool selection, and response generation. No complex prompt chains or hardcoded workflows. Tools are where our agent gets its superpowers. Strands makes it dead simple with the @tool decorator: The docstring is super important here!!! as it tells the AI model what the tool does and when to use it. The model reads this and decides autonomously when to invoke each tool based on user queries. AI-generated launch timeline with detailed task breakdown This is where things get interesting. Most AI chatbots are stateless. They forget everything after each conversation. But for a SaaS product, we need persistence. We need the agent to remember: AgentCore Memory solves this with two types of memory: Here's how I implemented memory hooks with Strands: The key insight here is the hook system. Before each message is processed, we retrieve relevant memories and inject them as context. After the agent responds, we save the interaction for future reference. The agent remembering product context across sessions I set up two memory strategies: The memories expire after 90 days (configurable), and the memory ID is stored in AWS SSM Parameter Store for persistence. For the web interface, I used FastAPI with Server-Sent Events (SSE) for streaming responses: The streaming experience is crucial for UX. Nobody wants to stare at a loading spinner for 10 seconds. And, that's actually it! It's that simple and straightforward. You can simply add a UI to the backend (or vibe code it), and you've built a fully functional, scalable, and production-ready SaaS for yourself. But, to truly make it production-ready, there are a few things that I'd do differently. When I built this, I ran the agent locally. For production, I'd use AgentCore Runtime from day one. It gives you: The hackathon version has no Terraform/CDK. Big mistake for production. I'd always go with IaC to keep things consistent and manageable: Probably wanna check this official AWS article, which outlines building AI Agents with CloudFormation. My tools are pretty monolithic. In hindsight, I'd break them into smaller, composable pieces. For example: This makes the agent more flexible and easier to test. The Product Hunt assistant is a single agent. But as your SaaS grows, you'll want multiple agents working together: Also, apart from these, some best practices that I'd put into place would be to: Alright, that's it! If you've made it this far, you now know more about building AI agents on AWS than most developers out there. The stack is still evolving fast, but the fundamentals of tools, memory, and agents aren't going anywhere. If you wanna try it out yourself, find the code for the assistant on Github. So go ahead, clone the repo, break things, and build something cool. And hey, if you end up launching on Product Hunt using this assistant, let me know, I'd love to see what you ship! Templates let you quickly answer FAQs or store snippets for re-use. Are you sure you want to hide this comment? It will become hidden in your post, but will still be visible via the comment's permalink. Hide child comments as well For further actions, you may consider blocking this person and/or reporting abuse COMMAND_BLOCK: from strands import Agent from strands.models import BedrockModel class ProductHuntLaunchAgent: def __init__(self, region_name: str = None, user_id: str = None, session_id: str = None): # Initialize the Bedrock model (Claude 3.5 Haiku) self.model = BedrockModel( model_id="anthropic.claude-3-5-haiku-20241022-v1:0", temperature=0.3, region_name=self.region, stream=True # Enable streaming ) # Create the agent with Product Hunt tools and memory hooks self.agent = Agent( model=self.model, tools=[ generate_launch_timeline, generate_marketing_assets, research_top_launches, ], system_prompt=self.system_prompt, hooks=[self.memory_hooks], ) Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: from strands import Agent from strands.models import BedrockModel class ProductHuntLaunchAgent: def __init__(self, region_name: str = None, user_id: str = None, session_id: str = None): # Initialize the Bedrock model (Claude 3.5 Haiku) self.model = BedrockModel( model_id="anthropic.claude-3-5-haiku-20241022-v1:0", temperature=0.3, region_name=self.region, stream=True # Enable streaming ) # Create the agent with Product Hunt tools and memory hooks self.agent = Agent( model=self.model, tools=[ generate_launch_timeline, generate_marketing_assets, research_top_launches, ], system_prompt=self.system_prompt, hooks=[self.memory_hooks], ) COMMAND_BLOCK: from strands import Agent from strands.models import BedrockModel class ProductHuntLaunchAgent: def __init__(self, region_name: str = None, user_id: str = None, session_id: str = None): # Initialize the Bedrock model (Claude 3.5 Haiku) self.model = BedrockModel( model_id="anthropic.claude-3-5-haiku-20241022-v1:0", temperature=0.3, region_name=self.region, stream=True # Enable streaming ) # Create the agent with Product Hunt tools and memory hooks self.agent = Agent( model=self.model, tools=[ generate_launch_timeline, generate_marketing_assets, research_top_launches, ], system_prompt=self.system_prompt, hooks=[self.memory_hooks], ) COMMAND_BLOCK: from strands import tool @tool def generate_launch_timeline( product_name: str, product_type: str, launch_date: str, additional_notes: str = "" ) -> Dict[str, Any]: """ Generate a comprehensive launch timeline and checklist for Product Hunt launch. Args: product_name: Name of the product to launch product_type: Type of product (SaaS, Mobile App, Chrome Extension, etc.) launch_date: Target launch date (e.g., "next Tuesday", "December 15, 2024") additional_notes: Any additional requirements or constraints """ # Parse launch date, calculate timeline, return structured data parsed_date = parse_launch_date(launch_date) days_until_launch = calculate_timeline_days(parsed_date) timeline = create_timeline(product_name, product_type, parsed_date, days_until_launch) return { "success": True, "timeline": timeline, "total_days": days_until_launch, "launch_date": format_date_for_display(parsed_date), "key_milestones": extract_milestones(timeline) } Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: from strands import tool @tool def generate_launch_timeline( product_name: str, product_type: str, launch_date: str, additional_notes: str = "" ) -> Dict[str, Any]: """ Generate a comprehensive launch timeline and checklist for Product Hunt launch. Args: product_name: Name of the product to launch product_type: Type of product (SaaS, Mobile App, Chrome Extension, etc.) launch_date: Target launch date (e.g., "next Tuesday", "December 15, 2024") additional_notes: Any additional requirements or constraints """ # Parse launch date, calculate timeline, return structured data parsed_date = parse_launch_date(launch_date) days_until_launch = calculate_timeline_days(parsed_date) timeline = create_timeline(product_name, product_type, parsed_date, days_until_launch) return { "success": True, "timeline": timeline, "total_days": days_until_launch, "launch_date": format_date_for_display(parsed_date), "key_milestones": extract_milestones(timeline) } COMMAND_BLOCK: from strands import tool @tool def generate_launch_timeline( product_name: str, product_type: str, launch_date: str, additional_notes: str = "" ) -> Dict[str, Any]: """ Generate a comprehensive launch timeline and checklist for Product Hunt launch. Args: product_name: Name of the product to launch product_type: Type of product (SaaS, Mobile App, Chrome Extension, etc.) launch_date: Target launch date (e.g., "next Tuesday", "December 15, 2024") additional_notes: Any additional requirements or constraints """ # Parse launch date, calculate timeline, return structured data parsed_date = parse_launch_date(launch_date) days_until_launch = calculate_timeline_days(parsed_date) timeline = create_timeline(product_name, product_type, parsed_date, days_until_launch) return { "success": True, "timeline": timeline, "total_days": days_until_launch, "launch_date": format_date_for_display(parsed_date), "key_milestones": extract_milestones(timeline) } COMMAND_BLOCK: from bedrock_agentcore.memory import MemoryClient from strands.hooks import HookProvider, HookRegistry, MessageAddedEvent, AfterInvocationEvent class ProductHuntMemoryHooks(HookProvider): def __init__(self, memory_id: str, client: MemoryClient, actor_id: str, session_id: str): self.memory_id = memory_id self.client = client self.actor_id = actor_id self.session_id = session_id def retrieve_product_context(self, event: MessageAddedEvent): """Retrieve product and user context BEFORE processing the query.""" user_query = messages[-1]["content"][0]["text"] # Get relevant memories from both namespaces for context_type, namespace in self.namespaces.items(): memories = self.client.retrieve_memories( memory_id=self.memory_id, namespace=namespace.format(actorId=self.actor_id), query=user_query, top_k=3, ) # Inject context into the user's message # ... def save_launch_interaction(self, event: AfterInvocationEvent): """Save the interaction AFTER the agent responds.""" self.client.create_event( memory_id=self.memory_id, actor_id=self.actor_id, session_id=self.session_id, messages=[ (user_query, "USER"), (agent_response, "ASSISTANT"), ], ) Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: from bedrock_agentcore.memory import MemoryClient from strands.hooks import HookProvider, HookRegistry, MessageAddedEvent, AfterInvocationEvent class ProductHuntMemoryHooks(HookProvider): def __init__(self, memory_id: str, client: MemoryClient, actor_id: str, session_id: str): self.memory_id = memory_id self.client = client self.actor_id = actor_id self.session_id = session_id def retrieve_product_context(self, event: MessageAddedEvent): """Retrieve product and user context BEFORE processing the query.""" user_query = messages[-1]["content"][0]["text"] # Get relevant memories from both namespaces for context_type, namespace in self.namespaces.items(): memories = self.client.retrieve_memories( memory_id=self.memory_id, namespace=namespace.format(actorId=self.actor_id), query=user_query, top_k=3, ) # Inject context into the user's message # ... def save_launch_interaction(self, event: AfterInvocationEvent): """Save the interaction AFTER the agent responds.""" self.client.create_event( memory_id=self.memory_id, actor_id=self.actor_id, session_id=self.session_id, messages=[ (user_query, "USER"), (agent_response, "ASSISTANT"), ], ) COMMAND_BLOCK: from bedrock_agentcore.memory import MemoryClient from strands.hooks import HookProvider, HookRegistry, MessageAddedEvent, AfterInvocationEvent class ProductHuntMemoryHooks(HookProvider): def __init__(self, memory_id: str, client: MemoryClient, actor_id: str, session_id: str): self.memory_id = memory_id self.client = client self.actor_id = actor_id self.session_id = session_id def retrieve_product_context(self, event: MessageAddedEvent): """Retrieve product and user context BEFORE processing the query.""" user_query = messages[-1]["content"][0]["text"] # Get relevant memories from both namespaces for context_type, namespace in self.namespaces.items(): memories = self.client.retrieve_memories( memory_id=self.memory_id, namespace=namespace.format(actorId=self.actor_id), query=user_query, top_k=3, ) # Inject context into the user's message # ... def save_launch_interaction(self, event: AfterInvocationEvent): """Save the interaction AFTER the agent responds.""" self.client.create_event( memory_id=self.memory_id, actor_id=self.actor_id, session_id=self.session_id, messages=[ (user_query, "USER"), (agent_response, "ASSISTANT"), ], ) CODE_BLOCK: from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() @app.post("/api/chat-stream") async def chat_stream(request: ChatRequest): async def event_generator(): agent = get_or_create_agent(request.user_id, request.session_id) for chunk in agent.chat_stream(request.message): yield f"data: {json.dumps({'content': chunk})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() @app.post("/api/chat-stream") async def chat_stream(request: ChatRequest): async def event_generator(): agent = get_or_create_agent(request.user_id, request.session_id) for chunk in agent.chat_stream(request.message): yield f"data: {json.dumps({'content': chunk})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") CODE_BLOCK: from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() @app.post("/api/chat-stream") async def chat_stream(request: ChatRequest): async def event_generator(): agent = get_or_create_agent(request.user_id, request.session_id) for chunk in agent.chat_stream(request.message): yield f"data: {json.dumps({'content': chunk})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") - Generate comprehensive launch timelines with task dependencies - Create marketing assets (taglines, tweets, descriptions) - Research successful launches in our category - Recommend hunters and outreach strategies - Remember our product context across sessions (yes, it has memory!) - A model (Claude via Bedrock in this case) - A list of tools the agent can use - A system prompt with domain expertise - Optional hooks for things like memory - What product is the user launching - Their preferences and communication style - Previous recommendations and decisions - Short-term memory: Keeps track of the current conversation - Long-term memory: Stores key insights across multiple sessions - USER_PREFERENCE: Stores user preferences, communication style, strategic approaches, etc. - SEMANTIC: Stores factual information about products, launch strategies, recommendations, etc. - Session isolation (no state leaking between users) - 8-hour execution windows (for long-running tasks) - Pay-per-use pricing - Built-in security with identity management - Memory resources defined in CloudFormation - Lambda functions for tools (if needed) - Proper IAM roles and policies - parse_date tool - calculate_timeline tool - format_output tool - A research agent that finds competitor data - A content agent that writes marketing copy - A scheduling agent that optimizes launch timing - An orchestrator agent that coordinates everything - Collect ground truth data: building a dataset of user queries and expected responses for testing - Use Bedrock Guardrails: adding safety rails to prevent harmful outputs - Monitor with AgentCore Observability: integrating with CloudWatch, Datadog, or LangSmith for clear insights and observability - Test tool selection: making sure the agent picks the right tool for each query