Agentic Commerce: From Recommendation to Autonomous Transaction

Commerce is evolving from “search and click” to “describe and buy.” Agentic commerce represents the next evolution: AI agents that handle the entire transaction lifecycle—from understanding user intent, to recommending products, to executing payment and fulfillment—all autonomously.

This guide covers building production agentic commerce systems, including payment integration, fraud detection, idempotency, and escalation to human operators when needed.

The Agentic Commerce Pipeline

User Intent → Product Discovery → Comparison → Decision → Payment → Fulfillment
     ↓              ↓                ↓           ↓          ↓           ↓
  Natural      Search/Match     Rank/Filter  Recommend   Execute    Track
  Language     Catalog          Options      & Confirm   Charge     & Notify

Key Components

1. Intent Understanding: Parse natural language requests
2. Product Discovery: Search catalogs with semantic matching
3. Decision Making: Compare options, apply business rules
4. Payment Execution: Secure, idempotent transactions
5. Fulfillment: Order processing and tracking
6. Escalation: Human handoff for edge cases

Building the Agent

Tool Definitions

Define commerce tools with strict input validation:

from pydantic import BaseModel, Field
from enum import Enum

class ProductSearch(BaseModel):
    query: str = Field(description="Product search query")
    max_price: float | None = Field(default=None, ge=0)
    min_rating: float | None = Field(default=None, ge=0, le=5)
    category: str | None = None

class CheckoutRequest(BaseModel):
    product_id: str
    quantity: int = Field(ge=1, le=100)
    shipping_method: str = "standard"
    idempotency_key: str = Field(description="Unique key for this transaction")

@function_tool
def search_products(search: ProductSearch) -> list[dict]:
    """Search the product catalog."""
    return product_catalog.search(
        query=search.query,
        max_price=search.max_price,
        min_rating=search.min_rating,
        category=search.category,
    )

@function_tool
def checkout(request: CheckoutRequest) -> dict:
    """Process a checkout request with idempotency."""
    return payment_gateway.charge(
        product_id=request.product_id,
        quantity=request.quantity,
        shipping_method=request.shipping_method,
        idempotency_key=request.idempotency_key,
    )

Agent Configuration

commerce_agent = Agent(
    name="Shopping Assistant",
    instructions="""You are a helpful shopping assistant.

When helping users:
1. Understand their needs from natural language
2. Search for relevant products
3. Compare options and make recommendations
4. Only process payment after explicit user confirmation
5. Always use idempotency keys for payments
6. Escalate to human for orders over $500 or unusual requests

Never:
- Process payment without user confirmation
- Share payment details in messages
- Make assumptions about shipping preferences""",
    tools=[search_products, checkout, get_product_details],
    output_type=ShoppingRecommendation,
)

Payment Integration

Stripe Integration

import stripe

stripe.api_key = "sk_your_key"

class StripePaymentGateway:
    async def charge(
        self,
        product_id: str,
        quantity: int,
        idempotency_key: str,
    ) -> dict:
        """Process payment with idempotency."""
        try:
            # Create or retrieve existing payment intent
            intent = stripe.PaymentIntent.create(
                amount=calculate_total(product_id, quantity),
                currency="usd",
                idempotency_key=idempotency_key,
                metadata={
                    "product_id": product_id,
                    "quantity": quantity,
                },
            )

            return {
                "status": "success",
                "payment_intent_id": intent.id,
                "amount": intent.amount,
            }

        except stripe.error.CardError as e:
            return {"status": "failed", "error": str(e)}
        except stripe.error.IdempotencyError:
            # Return existing result for this key
            return await self.get_existing_charge(idempotency_key)

Idempotency

Idempotency prevents duplicate charges:

import uuid
from datetime import datetime

class IdempotencyStore:
    def __init__(self):
        self.store = {}  # Use Redis in production

    def generate_key(self, user_id: str, product_id: str) -> str:
        """Generate a unique idempotency key."""
        return f"{user_id}:{product_id}:{datetime.now().isoformat()}"

    async def check_and_store(self, key: str, result: dict) -> dict:
        """Check if key exists, store result if not."""
        if key in self.store:
            return self.store[key]

        self.store[key] = result
        return result

Fraud Detection

Rule-Based Detection

class FraudDetector:
    def __init__(self):
        self.rules = [
            self.check_velocity,
            self.check_amount,
            self.check_location,
        ]

    async def evaluate(self, transaction: dict) -> dict:
        """Evaluate transaction against fraud rules."""
        risk_score = 0
        triggered_rules = []

        for rule in self.rules:
            result = await rule(transaction)
            if result["triggered"]:
                risk_score += result["score"]
                triggered_rules.append(result["rule"])

        return {
            "risk_score": risk_score,
            "triggered_rules": triggered_rules,
            "action": self.determine_action(risk_score),
        }

    async def check_velocity(self, transaction: dict) -> dict:
        """Check for rapid successive transactions."""
        recent_count = await self.get_recent_transaction_count(
            transaction["user_id"],
            minutes=5,
        )
        return {
            "rule": "velocity",
            "triggered": recent_count > 3,
            "score": 30 if recent_count > 3 else 0,
        }

    async def check_amount(self, transaction: dict) -> dict:
        """Check for unusual transaction amounts."""
        avg_amount = await self.get_average_amount(transaction["user_id"])
        if transaction["amount"] > avg_amount * 3:
            return {"rule": "amount", "triggered": True, "score": 20}
        return {"rule": "amount", "triggered": False, "score": 0}

    def determine_action(self, risk_score: int) -> str:
        if risk_score >= 50:
            return "block"
        elif risk_score >= 30:
            return "review"
        else:
            return "approve"

ML-Based Detection

class MLFraudDetector:
    def __init__(self):
        self.model = load_fraud_model()

    async def evaluate(self, transaction: dict) -> dict:
        """ML-based fraud scoring."""
        features = self.extract_features(transaction)
        prediction = self.model.predict([features])

        return {
            "fraud_probability": prediction[0],
            "action": "block" if prediction[0] > 0.8 else "approve",
        }

    def extract_features(self, transaction: dict) -> list:
        """Extract ML features from transaction."""
        return [
            transaction["amount"],
            transaction["time_of_day"],
            transaction["device_fingerprint"],
            transaction["user_account_age"],
            # ... more features
        ]

Escalation Patterns

Human Handoff

class EscalationHandler:
    def should_escalate(self, context: dict) -> bool:
        """Determine if the agent should escalate."""
        return (
            context["order_amount"] > 500 or
            context["user_risk_score"] > 30 or
            context["is_return_customer"] and context["previous_issues"] > 2 or
            context["unusual_request"]
        )

    async def escalate(self, context: dict, reason: str) -> dict:
        """Hand off to human operator."""
        ticket = await self.create_support_ticket(
            context=context,
            reason=reason,
            priority="high" if context["order_amount"] > 1000 else "medium",
        )

        return {
            "status": "escalated",
            "ticket_id": ticket.id,
            "message": "I've connected you with a specialist who can help with your order.",
        }

Conversation Flow

async def handle_commerce_conversation(user_message: str, session_id: str):
    """Handle a complete commerce conversation."""
    agent = commerce_agent
    history = await get_conversation_history(session_id)

    # Run agent with context
    result = await Runner.run(
        agent,
        user_message,
        context={
            "session_id": session_id,
            "user_id": get_user_id(session_id),
        },
    )

    # Check if payment was attempted
    if result.payment_attempted:
        # Verify payment result
        payment_result = result.payment_result

        if payment_result["status"] == "success":
            # Initiate fulfillment
            await initiate_fulfillment(payment_result)
        elif payment_result["status"] == "requires_action":
            # 3D Secure or similar
            return {"action": "3ds_required", "url": payment_result["url"]}

    # Save conversation
    await save_conversation(session_id, user_message, result.final_output)

    return {"response": result.final_output}

Best Practices

1. Always require confirmation: Never process payment without explicit user consent
2. Use idempotency keys: Prevent duplicate charges on retries
3. Implement fraud detection: Rule-based + ML for defense in depth
4. Escalate appropriately: Human handoff for high-value or unusual transactions
5. Audit everything: Complete transaction logs for compliance
6. Handle failures gracefully: Clear error messages and retry paths
7. Test edge cases: Failed payments, timeouts, partial fulfillments

Next Steps

• Explore Stripe Integration for detailed payment setup
• Read about Agentic Commerce with ADK for Google ADK patterns
• Learn about Agent Tracing for transaction monitoring

Agentic commerce represents the future of digital transactions. By building systems that handle the complete lifecycle—from discovery to fulfillment—you can create seamless buying experiences that operate 24/7 without human intervention.