Build an AI Research Agent with Web Search on Amazon Bedrock AgentCore

"""
Demo: Web Search on Amazon Bedrock AgentCore

A Strands Agents demo showing how an AI agent uses the fully managed
Web Search tool on AgentCore Gateway to ground responses in current,
cited web knowledge — with zero data egress from AWS.
"""

import httpx
import botocore.session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp import MCPClient
from strands.models.bedrock import BedrockModel

# =============================================================================
# CONFIGURATION
# =============================================================================
GATEWAY_URL = "https://<your-gateway-id>.gateway.bedrock-agentcore.<region>.amazonaws.com/mcp"
AWS_REGION = "us-east-1"
AWS_SERVICE = "bedrock-agentcore"

MODEL = BedrockModel(
model_id="us.anthropic.claude-sonnet-4-6",
region_name="us-east-1",
)

# =============================================================================
# SigV4 Auth
# =============================================================================
class SigV4HttpxAuth(httpx.Auth):
def __init__(self):
session = botocore.session.get_session()
self._credentials = session.get_credentials()

def auth_flow(self, request: httpx.Request):
creds = self._credentials.get_frozen_credentials()
aws_request = AWSRequest(
method=request.method,
url=str(request.url),
headers={k: v for k, v in request.headers.items()},
data=request.content,
)
SigV4Auth(creds, AWS_SERVICE, AWS_REGION).add_auth(aws_request)
for key, value in aws_request.headers.items():
request.headers[key] = value
yield request

def create_signed_client():
return streamablehttp_client(url=GATEWAY_URL, auth=SigV4HttpxAuth())

# =============================================================================
# SYSTEM PROMPT
# =============================================================================
SYSTEM_PROMPT = """You are a research assistant with access to web search.

## How You Work
1. ALWAYS search before answering. Never rely on training data for facts.
2. For comparisons, make MULTIPLE searches and synthesize.
3. Use maxResults strategically: 5 for factual lookups, 10-15 for research, 20-25 for surveys.

## Response Format
- Keep answers concise: max 5-8 bullet points.
- Cite claims inline as: [Title](URL)
- No separate "Sources" section at the end.
"""

# =============================================================================
# DEMO QUERIES
# =============================================================================
DEMO_QUERIES = [
{
"label": "🔴 REAL-TIME GROUNDING",
"query": "What new features did AWS announce this week?",
"feature": "Freshness — index refreshed within minutes of publication",
},
{
"label": "🟡 KNOWLEDGE GRAPH",
"query": "Who is the CEO of Amazon and when was the company founded?",
"feature": "Built-in knowledge graph for high-confidence entity facts",
},
{
"label": "🟢 MULTI-QUERY SYNTHESIS",
"query": "What are the top 3 programming languages in 2025 and why?",
"feature": "Agent searches multiple times and synthesizes across sources",
},
{
"label": "🔵 SEMANTIC SNIPPETS",
"query": "How does Amazon Bedrock AgentCore Web Search work?",
"feature": "Returns relevant passages optimized for LLM context, not raw HTML",
},
{
"label": "🟣 STALE KNOWLEDGE FIX",
"query": "What is the latest stable version of Python and its key features?",
"feature": "Current facts beyond model training cutoff",
},
]

def main():
web_search = MCPClient(create_signed_client)

agent = Agent(
model=MODEL,
tools=[web_search],
system_prompt=SYSTEM_PROMPT,
)

print("""
╔══════════════════════════════════════════════════════════════╗
║ ║
║ Web Search on Amazon Bedrock AgentCore ║
║ ────────────────────────────────────── ║
║ Strands Agents SDK Demo ║
║ ║
║ Features: ║
║ • Real-time web index (refreshed within minutes) ║
║ • Knowledge graph for entity facts ║
║ • Semantic snippet extraction ║
║ • Domain governance (server-side denylist) ║
║ • Zero data egress — queries stay in AWS ║
║ • MCP-compliant tool discovery ║
║ ║
╚══════════════════════════════════════════════════════════════╝
""")

print("Choose mode:")
print(" [1] Guided demo (pre-built queries showcasing each feature)")
print(" [2] Interactive (ask anything)")
print()
mode = input("Select (1/2): ").strip()

if mode == "1":
run_guided_demo(agent)
else:
run_interactive(agent)

def run_guided_demo(agent: Agent):
print("\n" + "=" * 60)
print("GUIDED DEMO — Each query highlights a Web Search feature")
print("=" * 60)

for i, demo in enumerate(DEMO_QUERIES, 1):
print(f"\n{'─' * 60}")
print(f" Demo {i}/5: {demo['label']}")
print(f" Feature: {demo['feature']}")
print(f"{'─' * 60}")
print(f"\n Query: \"{demo['query']}\"\n")
input(" Press Enter to run... ")
print()
response = agent(demo["query"])
print(f"\n{response}")
if i < len(DEMO_QUERIES):
input("\n Press Enter for next demo... ")

print("\n" + "=" * 60)
print("DEMO COMPLETE")
print("=" * 60)
print("""
Features demonstrated:
✓ Real-time grounding ✓ Knowledge graph facts
✓ Multi-query synthesis ✓ Semantic snippets
✓ Stale knowledge fix ✓ Fact verification
✓ Zero data egress ✓ Domain filtering (infra)
✓ MCP tool discovery ✓ Framework-agnostic
""")

def run_interactive(agent: Agent):
print("\n Ready. Type 'quit' to exit.\n")
while True:
query = input("You: ").strip()
if query.lower() in ("quit", "exit", "q"):
break
if not query:
continue
response = agent(query)
print(f"\n{response}\n")

if __name__ == "__main__":
main()