AI Agents – Shopping Agent using Amazon Bedrock AgentCore and Strands Agents Python SDK

model_id = "eu.amazon.nova-micro-v1:0"
model = BedrockModel(
    model_id=model_id,
)

agent = Agent(
    model=model,
    tools=[
        list_products,
        get_product,
        create_order,
        list_orders,
        get_order,
        update_order,
        cancel_order,
    ],
    system_prompt="""You're a helpful shopping assistant. You can help users browse products and manage their orders.

You have access to the following tools:

Product Catalog:
- list_products: List all products from the catalog
- list_products_by_category: List products from the catalog filtered by category (Mice, Keyboards, Monitors, Headsets, Cameras, Accessories, Laptops)
- get_product: Get details of a specific product by its ID

Order Management:
- create_order: Create a new order with items (requires product_id, name, price, quantity for each item)
- list_orders: List all orders
- list_orders_filtered_by_status: List orders, filtered by status
- get_order: Get details of a specific order by its ID
- update_order: Update an order's items or status
- cancel_order: Cancel an existing order

When helping users place orders, first look up products to get accurate product_id, name, and price information.
"""
)

@tool(description="List all products from the catalog")
def list_products() -> dict:
    """
    Tool description - List all products from the catalog

    #Returns:
        A dictionary containing 'products' list and 'count' of products found.
    """
    url = f"{PRODUCTS_CATALOG_BASE_URL}/api/v1/products"

    response = requests.get(url)
    response.raise_for_status()
    return response.json()


@tool(description="List products from the catalog filtered by category")
def list_products_by_category(category: str) -> dict:
    """
    Tool description - List products from the catalog filtered by category.

    #Args:
        category: category to filter products (e.g., "Mice", "Keyboards", "Monitors",
                  "Headsets", "Cameras", "Accessories", "Laptops")

    #Returns:
        A dictionary containing 'products' list and 'count' of products found.
    """
    url = f"{PRODUCTS_CATALOG_BASE_URL}/api/v1/products"
    params = {"category": category}

    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()


@tool(description="Get a single product by its ID")
def get_product(product_id: int) -> dict:
    """
    Tool description - Get a single product by its ID.

    #Args:
        product_id: The unique identifier of the product.

    #Returns:
        A dictionary containing product details (id, name, price, category, stock),
        or an error message if the product is not found.
    """
    url = f"{PRODUCTS_CATALOG_BASE_URL}/api/v1/products/{product_id}"

    response = requests.get(url)
    if response.status_code == 404:
        return {"error": f"Product with ID {product_id} not found"}
    response.raise_for_status()
    return response.json()

@tool(description="Create a new order with the specified items")
def create_order(items: list[dict]) -> dict:
    """
    Tool description - Create a new order with the specified items.

    #Args:
        items: List of order items. Each item must have:
               - product_id: The product ID (string)
               - name: Product name (string)
               - price: Product price (float)
               - quantity: Number of items to order (int)

    #Returns:
        The created order with id, items, total_value, status, created_at, and updated_at.
    """
    url = f"{ORDERS_BASE_URL}/api/v1/orders"

    response = requests.post(url, json={"items": items})
    if response.status_code == 400:
        return {"error": response.json().get("error", "Invalid order data")}
    response.raise_for_status()
    return response.json()


@tool(description="List all orders")
def list_orders() -> dict:
    """
    Tool description - List all orders

    #Returns:
        A dictionary containing 'orders' list and 'count' of orders found.
    """
    url = f"{ORDERS_BASE_URL}/api/v1/orders"

    response = requests.get(url)
    if response.status_code == 400:
        return {"error": response.json().get("error", "Invalid status")}
    response.raise_for_status()
    return response.json()

@app.route("/api/v1/products", methods=["GET"])
def list_products():
    category_filter = request.args.get("category")
    products = get_all_products(category_filter)
    return jsonify({"products": [p.to_dict() for p in products], "count": len(products)})


@app.route("/api/v1/products/<int:product_id>", methods=["GET"])
def get_product_by_id(product_id: int):
    product = get_product(product_id)
    if not product:
        return jsonify({"error": "Product not found"}), 404
    return jsonify(product.to_dict())

@app.route("/api/v1/orders", methods=["POST"])
def create_order_route():
    data = request.get_json()

    if not data or "items" not in data or not data["items"]:
        return jsonify({"error": "Order must contain at least one item"}), 400

    try:
        items = parse_order_items(data["items"])
    except (KeyError, ValueError) as e:
        return jsonify({"error": f"Invalid item data: {str(e)}"}), 400

    order = create_order(items)
    return jsonify(order.to_dict()), 201


@app.route("/api/v1/orders/<order_id>", methods=["GET"])
def get_order_route(order_id: str):
    order = get_order(order_id)

    if not order:
        return jsonify({"error": "Order not found"}), 404

    return jsonify(order.to_dict())

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python agent_runner_local.py

cd ../microservice-products-catalog
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
FLASK_RUN_PORT=5001 python application.py

cd ../microservice-orders
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
FLASK_RUN_PORT=5002 python application.py

#!/bin/bash

AWS_REGION=$(aws configure get region)

if [ ! -d ".elasticbeanstalk" ]; then
	echo "⚙️  Initializing Elastic Beanstalk Environment ..."
	eb init -p python-3.12 microservice-orders --region ${AWS_REGION} || {
		echo "❌ Error occurred while initializing Elastic Beanstalk Environment"
		exit 1
	}
	echo "✅ Done"
else
	echo "⏭️  Elastic Beanstalk already initialized, skipping ..."
fi

if ! eb status microservice-orders &>/dev/null; then
	echo "📦 Creating Elastic Beanstalk Environment ..."
	eb create microservice-orders || {
		echo "❌ Error occurred while creating Elastic Beanstalk Environment"
		exit 1
	}
	echo "✅ Done"
else
	echo "⏭️  Elastic Beanstalk environment already exists, skipping ..."
fi

echo "🚀 Deploying to Elastic Beanstalk Environment ..."
eb deploy || {
  echo "❌ Error occurred while deploying to Elastic Beanstalk Environment"
  exit 1
}
echo "✅ Done"

def configure_agentcore_runtime():
    print("⚙️  Configuring AgentCore Runtime...")
    agentcore_runtime = Runtime()
    agentcore_runtime.configure(
        entrypoint="agent_runner_aws.py",
        auto_create_execution_role=True,
        auto_create_ecr=True,
        requirements_file="requirements.txt",
        region=region,
        agent_name=agent_name
    )
    print("✅ AgentCore Runtime configured successfully.")

    return agentcore_runtime


def launch_agentcore_runtime(agentcore_runtime, env_vars: dict):
    print("🚀 Launching Agent to AgentCore Runtime...")
    print("⏳ This may take several minutes...")
    print(f"🔧 Environment variables: {env_vars}")
    launch_result = agentcore_runtime.launch(env_vars=env_vars)
    print("✅ Launch completed")
    print(f"🤖 Agent ARN: {launch_result.agent_arn}")
    print(f"🆔 Agent ID: {launch_result.agent_id}")

    return launch_result

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python deploy.py