Building Your First AI Shopping Agent: A Safe, 3-Step Pilot Plan
Why Most AI Shopping Pilots Crash and Burn (And How Yours Won’t)
The idea of AI-driven shopping is undeniably exciting. Google’s Universal Commerce Protocol (UCP) promises a future where AI agents can browse, compare, and purchase products on behalf of users. But here’s the uncomfortable truth: most early pilots are failing spectacularly.
The pattern is painfully predictable. Teams get swept up in the vision. They rush to build fully integrated checkout experiences from day one. They try connecting every backend system simultaneously: inventory, order management, loyalty programs, tax services. The result? An integration nightmare that’s brittle, unreliable, and doomed before it ever truly begins.
I’ve watched this happen repeatedly. And I’ve learned that success comes from a radically different philosophy: minimal initial capabilities with maximum observability.
Think of it like building a house. You must pour a solid foundation and erect a sturdy frame before thinking about the roof or interior design. This phased approach prevents system-wide meltdowns by breaking the problem into staged milestones, each with clear go/no-go criteria and rigorous testing.
Let me walk you through exactly how to do this in 90 days.
The Golden Rule: Start Small, See Everything
Before diving into the blueprint, you need to understand two fundamental concepts that will save your pilot from disaster.
Understanding Concurrency
Concurrency is what happens when multiple actions occur simultaneously. Imagine two AI agents trying to buy the last item in stock at the exact same moment. Without proper controls, one of two bad things happens: either one user gets an error after thinking their purchase succeeded, or worse, the item sells twice.
Proper concurrency control ensures only one agent successfully claims the item while the other receives a graceful notification.
Understanding Observability
Observability is your ability to understand what’s happening inside your system by observing its outputs. Think of it as a flight recorder for your AI agent. It gives you a detailed record of everything the agent did, saw, and decided during a transaction.
When a purchase fails (and it will), observability lets you replay the entire sequence to understand precisely what went wrong. For every agent decision, you need to answer three questions fast: what happened, why, and based on what.
This transforms mysterious failures into solvable engineering problems.
The 3-Step Blueprint for a Successful 90-Day Pilot
Here’s the overall architecture of what we’re building:
Each phase builds on the previous one. The strategy is to defer complexity until core components are proven stable and reliable.
Step 1: Laying the Foundation (Making Your Catalog AI-Ready)
Goal: Get your product catalog “agent-ready” so an AI can reliably discover products and receive accurate, real-time information. No purchases happen at this stage. We’re laying the track before running the train.
Why this matters: Without this foundation, your AI will hallucinate products, recommend out-of-stock items, or show wrong prices. Agent-driven commerce only works when product data is accurate, fresh, and standardized. Getting this right eliminates a massive source of future failures.
What to Build
Implement two read-only endpoints:
# Endpoint 1: Product Discovery
@app.route('/search_products', methods=['GET'])
def search_products():
"""
Allow agents to query products by keywords or filters.
Returns structured results with all attributes an agent needs.
"""
query = request.args.get('q')
results = catalog_service.search(
query=query,
include_fields=['name', 'id', 'price', 'availability', 'variants']
)
return jsonify({
'products': results,
'total': len(results),
'query': query
})
# Endpoint 2: Offer Quote
@app.route('/quote_offer', methods=['GET'])
def quote_offer():
"""
Given a product ID, return personalized offer details:
availability, final price, and applicable promotions.
"""
product_id = request.args.get('product_id')
quantity = request.args.get('quantity', 1, type=int)
offer = pricing_service.get_quote(
product_id=product_id,
quantity=quantity
)
return jsonify({
'product_id': product_id,
'price': offer.price,
'in_stock': offer.available,
'quantity_available': offer.stock_count,
'promotions': offer.applicable_promotions
})Testing Phase 1
Your test suite should cover three critical areas:
Data Accuracy Tests: Verify API responses match your real catalog. If a product shows $19.99 on your website, the API must return the same. Test variants, edge cases, and discontinued items.
Performance Tests: Set an SLA like p95 latency under 200ms. Use load testing to simulate bursts of searches. Slow responses kill the agent experience.
Get Micheal Lanham’s stories in your inbox
Join Medium for free to get updates from this writer.
Chaos Tests: Simulate your search backend going down. Does the API return proper errors that an agent can handle gracefully?
Go/No-Go Criteria
Proceed only when:
- Data quality exceeds 99% accuracy
- Performance meets your SLA
- You can trace any query through the system easily
Step 2: Adding a “Hold” Button (The Reservation System)
Goal: Introduce inventory reservation without processing payments. The system can place temporary holds on items, preventing the nightmare of selling something you don’t have.
Why this matters: This step digitizes a common retail practice. When a customer places a “buy online, pick up in store” order, a store employee physically pulls that item off the shelf and sets it aside. The order is only finalized after confirming the item is in hand. A reservation system does exactly this digitally.
What to Build
Add two new endpoints backed by a state store (Redis or a database table):
@app.route('/reserve_items', methods=['POST'])
def reserve_items():
"""
Place a temporary hold on items.
Uses idempotency keys to prevent double-reservations.
"""
data = request.json
items = data.get('items') # List of {product_id, quantity}
idempotency_key = request.headers.get('Idempotency-Key')
# Check for existing reservation with same key
existing = reservation_store.get_by_idempotency_key(idempotency_key)
if existing:
return jsonify(existing.to_dict())
# Attempt reservation with optimistic locking
try:
reservation = reservation_service.reserve(
items=items,
idempotency_key=idempotency_key,
expires_in_minutes=15
)
return jsonify({
'reserve_id': reservation.id,
'items': reservation.items,
'expires_at': reservation.expires_at.isoformat(),
'status': 'reserved'
})
except OutOfStockError as e:
return jsonify({
'error': 'out_of_stock',
'unavailable_items': e.items
}), 409
@app.route('/confirm_order', methods=['POST'])
def confirm_order():
"""
Convert reservation to order record.
Generates consent artifact and order ID.
"""
data = request.json
reserve_id = data.get('reserve_id')
reservation = reservation_store.get(reserve_id)
if not reservation or reservation.is_expired():
return jsonify({'error': 'reservation_expired'}), 410
order = order_service.create_from_reservation(
reservation=reservation,
user_consent_timestamp=datetime.utcnow()
)
return jsonify({
'order_id': order.id,
'status': 'pending_payment',
'items': order.items,
'total': order.total
})The Power of Idempotency
Idempotency is your secret weapon against duplicate orders. Even if an agent retries a request due to network glitches, identical requests have a single effect on the backend.
Problem How Reservation Solves It Concurrency Issues Prevents two agents from buying the last item simultaneously Out-of-Stock Surprises Catches inventory issues before payment Duplicate Orders Idempotency keys ensure retries don’t cause double bookings
Critical Tests for Phase 2
Concurrent Reserve Tests: Simulate two agents reserving the same item within milliseconds. One must succeed, the other must fail gracefully.
Expiration Tests: Let a reservation expire without confirmation. Verify inventory is released and late confirmations are rejected.
Failure Mode Tests: Force out-of-stock scenarios. Ensure the system handles partial fulfillment decisions consistently.
Step 3: Turning on the Power (Limited, Safe Payments)
Goal: Enable real but limited checkout. Think of this as a soft opening for a restaurant: test the full experience with a small, manageable group before opening to the public.
Why this matters: Because discovery and reservation are now proven reliable, this phase can safely focus on the only new variable: payment processing.
Safety Nets
Limited Scope: Start with one payment method (like Google Pay) and a small group of test users.
Step-Up Authentication: Handle scenarios where banks require extra verification like one-time passcodes.
Clean Rollback Logic: If payment fails, automatically undo everything and release inventory.
@app.route('/cancel_order', methods=['POST'])
def cancel_order():
"""
Clean rollback: cancel order and release inventory.
Idempotent: multiple calls have same effect.
"""
data = request.json
order_id = data.get('order_id')
order = order_store.get(order_id)
if not order:
return jsonify({'error': 'order_not_found'}), 404
if order.status == 'cancelled':
return jsonify({'status': 'already_cancelled', 'order_id': order_id})
# Release inventory
reservation_service.release(order.reservation_id)
# Refund if payment was captured
if order.payment_captured:
payment_service.refund(order.payment_id)
# Update order status
order.status = 'cancelled'
order_store.save(order)
logger.info(f"Order {order_id} cancelled and inventory released",
extra={'order_id': order_id, 'trace_id': get_trace_id()})
return jsonify({
'status': 'cancelled',
'order_id': order_id,
'inventory_released': True
})The Ultimate Test
The ultimate test of observability is this: an on-call engineer should be able to debug any failed transaction within minutes. If a checkout fails and nobody can explain why from logs and traces, that’s a no-go to proceed further.
Building the Agent with OpenAI’s Agents SDK
Here’s how to wire everything together using the OpenAI Agents SDK:
from openai_agents import Agent, tool
# Define tools that map to our API endpoints
@tool
def search_products(query: str) -> dict:
"""Search for products matching the query."""
response = requests.get(
f"{API_BASE}/search_products",
params={"q": query}
)
return response.json()
@tool
def quote_offer(product_id: str, quantity: int = 1) -> dict:
"""Get price and availability for a specific product."""
response = requests.get(
f"{API_BASE}/quote_offer",
params={"product_id": product_id, "quantity": quantity}
)
return response.json()
@tool
def reserve_items(items: list) -> dict:
"""Reserve items for purchase."""
response = requests.post(
f"{API_BASE}/reserve_items",
json={"items": items},
headers={"Idempotency-Key": generate_idempotency_key()}
)
return response.json()
@tool
def confirm_order(reserve_id: str) -> dict:
"""Confirm the order and initiate payment."""
response = requests.post(
f"{API_BASE}/confirm_order",
json={"reserve_id": reserve_id}
)
return response.json()
# Create the shopping agent
shopping_agent = Agent(
name="Shopping Assistant",
instructions="""You help users purchase items by:
1. Searching for products matching their needs
2. Getting quotes to verify availability
3. Reserving items BEFORE confirming orders
4. Handling errors gracefully (suggest alternatives if unavailable)
""",
tools=[search_products, quote_offer, reserve_items, confirm_order]
)Example Transaction Flow
Here’s what a successful transaction looks like in your logs:
[Agent] Received: "Find me a laptop bag under $50"
[Tool:search_products] Input: "laptop bag under $50" -> 3 results
[Tool:quote_offer] Input: product_id=101 -> price=$45, in_stock=true
[Tool:reserve_items] Input: [101] -> reserve_id="R-0001", expires=15min
[Tool:confirm_order] Input: "R-0001" -> order_id="O-12345", status="paid"
[Agent] Response: "Order O-12345 confirmed! Expect confirmation email shortly."Every step shares a trace ID. When something fails, you can reconstruct the entire narrative in seconds.
Your Blueprint for the Future
By the end of this 90-day pilot, you’ll have more than a tech demo. You’ll have a working agentic commerce flow built on rock-solid foundations with concrete metrics proving reliability: zero duplicate orders, high reservation success rates, and payment flows that fail gracefully.
This phased approach strategically turns potential failures into learning opportunities. Instead of a big failure followed by doubt, you get quick wins followed by controlled expansions.
From here, you can iteratively expand: plug in tax calculations in Phase 4, add loyalty point redemption in Phase 5, and so on. Each addition builds on proven foundations.
Most importantly, when someone asks “What if the agent tries to buy something out-of-stock at 3 AM?”, you’ll have an answer and a log trace to prove it.
You’ll be ready for the future of commerce.
Want to dive deeper into AI agent development? Follow me for more practical guides on building production-ready AI systems.
