Accelerate Deep Research Workflows with Amazon Bedrock AgentCore Application

git clone https://github.com/aws-samples/sample-agentcore-deep-research.git
cd sample-agentcore-deep-research

cd infra-cdk
cp .config_example.yaml config.yaml

stack_name_base: deep-research
admin_user_email: your-email@example.com  # auto-creates user and emails credentials

backend:
  pattern: strands-deep-research
  deployment_type: docker
  model_id: global.anthropic.claude-sonnet-4-6

tools:
  arxiv:
    enabled: true
    default_on: true
  nova:
    enabled: true
    default_on: true
  tavily:
    enabled: true          # set to true if you have a Tavily API key
    default_on: true
    required:
      api_key: tvly-xxx    # get your key from https://tavily.com/
  # ... other tools configured similarly

npm install
npx cdk bootstrap    # only required once per account/region
npm run deploy

App URL: https://main.d1234567abcdef.amplifyapp.com

import json
import boto3

client = boto3.client("bedrock-agentcore-runtime", region_name="us-east-1")

response = client.invoke_agent_runtime(
    agentRuntimeArn=runtime_arn,
    payload=json.dumps({
        "prompt": "What are the latest advances in quantum error correction?",
        "runtimeSessionId": session_id,
        "enabledSources": ["arxiv", "nova"],
    }),
    payloadContentType="application/json",
    sessionId=session_id,
    authContext={
        "authToken": access_token,
        "authTokenType": "ACCESS_TOKEN",
    },
)

for event in response["payloadStream"]:
    if "chunk" in event:
        chunk = json.loads(event["chunk"]["bytes"].decode("utf-8"))
        if "data" in chunk:
            print(chunk["data"], end="", flush=True)

gateway/tools/your_tool/
  your_tool_lambda.py    # Lambda handler
  tool_spec.json         # MCP tool schema

[
  {
    "name": "internal_api_search",
    "description": "Search the company internal knowledge base for policies, procedures, and technical documentation.",
    "inputSchema": {
      "type": "object",
      "properties": {
        "query": {
          "type": "string",
          "description": "Search query for finding relevant internal documents"
        },
        "max_results": {
          "type": "integer",
          "description": "Maximum number of results to return (default 10)"
        }
      },
      "required": ["query"]
    }
  }
]

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def handler(event, context):
    """Lambda handler for AgentCore Gateway tool."""
    logger.info(f"Received event: {json.dumps(event)}")

    try:
        query = event.get("query", "")
        max_results = event.get("max_results", 10)

        # Your tool logic here: call an API, query a database, etc.
        results = your_search_function(query, max_results)

        return {"content": [{"type": "text", "text": results}]}

    except Exception as e:
        logger.error(f"Error: {str(e)}")
        return {"error": f"Internal server error: {str(e)}"}

cd docker
docker compose up

cd infra-cdk
npx cdk destroy --force