Imagine ordering your weekly groceries as naturally as texting a friend:"Hey, we're out of milk and eggs. Can you also add some snacks for Friday movie night?"
In our previous post, Building Conversational Commerce with Multi-Agent Swarms on GCP, we shared the architectural blueprint for conversational commerce—proposing a 3-layer decoupled design (Ingress, Memory, and Cognition) to solve messaging timeouts, state tracking, and response reliability.
While blueprints establish the architecture on paper, production readiness demands working software. In this post, we take that blueprint and bring it to life with a functional prototype called HappyDelivery.
Using LangGraph, Gemini, FastAPI, and Google Cloud, we'll walk through the code, the state machine, deterministic financial tools, and an automated verification suite that demonstrates how this multi-agent swarm operates in practice.
To build a reliable commerce swarm, we implemented three decoupled layers that handle the journey of every customer text message:
Let's examine how each layer functions in code with its own dedicated architecture.
A frequent question when designing conversational commerce is: how does a message sent by a customer on their smartphone actually land in Layer 1, and can this architecture support messaging platforms beyond WhatsApp?
End users do not communicate directly with your Google Cloud servers. Instead, customer messages traverse an operator ecosystem before landing on your webhook:
Every major consumer messaging platform operates on the exact same fundamental contract: an asynchronous HTTPS POST webhook delivery with a strict HTTP acknowledgement timeout:
X-Hub-Signature-256 header against your Meta App Secret using HMAC-SHA256. If your server times out or returns a non-200 status, Meta marks delivery as failed and initiates exponential backoff retries (for up to 7 days), which can trigger duplicate order deliveries.X-Goog-Signature). Requires returning HTTP 200 promptly to avoid webhook delivery retries.HTTP 200 OK acknowledgement expectations.X-Twilio-Signature).
While the payload schemas and cryptographic headers differ slightly between platforms, Layer 1 abstracts away all client differences:
MetaHmacValidator, RcsJwtValidator, AppleJwsValidator, TwilioHmacValidator). CommonMessageEvent):
{
"channel": "whatsapp", # or "rcs", "apple_messages", "sms"
"sender_id": "+1xxxxxxxxxx", # Verified E.164 phone number
"text": "Out of milk & eggs",
"timestamp": 1773428900,
"raw_metadata": {...}
}
< 150ms.
This ensures the rest of your system—Session Store (Layer 2) and Agent Swarm (Layer 3)—operates on clean, normalized data without caring whether the shopper sent their message from an iPhone, an Android RCS chat, or a WhatsApp thread.
Messaging platforms are completely stateless: every text message arrives as an isolated webhook event. To provide a coherent shopping experience, the system must instantly reconstruct the customer's full context without having to re-read or re-parse days of chat logs on every turn.
We use Cloud Memorystore (Redis) keyed by the customer's verified phone number (session:{phone_number}) to store active state:
dietary_preferences: ["Nut-Free"], the Inventory Agent automatically respects that preference when recommending movie night snacks—without the user having to re-declare their allergy in every text message.customer_id: "cust_9821", payment_ref: "Visa ending in 4242"). The LLM only ever reads the friendly masked label, while the transaction tools pass the tokenized customer ID to Stripe. Raw credit card numbers never touch chat prompts or logs.
A common question in conversational commerce is whether state lookups will introduce latency. Here is what the baseline infrastructure supports:
Key Takeaway: In practice, the Session & Cart Layer will never be the bottleneck—it easily outpaces upstream webhook inference and downstream LLM inference.
Rather than relying on a single monolithic prompt to handle every task, we separate reasoning into a swarm of specialized agents:
Now that we have seen the high-level design of the three layers, let's look at the actual Python prototype code that brings each layer to life.
The goal of this service is simple: verify authenticity across any inbound channel, normalize the payload into a canonical schema, and get off the critical path in < 150ms.
Here is our implementation showing how multiple messaging channels (WhatsApp, RCS, Apple Messages, SMS) are handled by pluggable adapters:
import hashlib, hmac, json, time
from typing import Any, Dict, Optional
from pydantic import BaseModel
class CommonMessageEvent(BaseModel):
"""Channel-agnostic canonical representation of an inbound customer message."""
channel: str # "whatsapp", "rcs", "apple_messages", "sms"
sender_id: str # E.164 phone number: "+1xxxxxxxxxx"
text: str
timestamp: float
raw_metadata: Dict[str, Any]
class HappyDeliveryIngressService:
"""Omni-Channel FastAPI Ingress with cryptographic verification and async queueing."""
def __init__(self, meta_secret: str, rcs_secret: str = ""):
self.meta_secret = meta_secret
self.rcs_secret = rcs_secret
self.publisher = CloudPubSubPublisher()
def verify_whatsapp_signature(self, payload_bytes: bytes, sig_header: str) -> bool:
"""Verifies X-Hub-Signature-256 header using constant-time comparison."""
if not sig_header or not sig_header.startswith("sha256="):
return False
expected_hash = hmac.new(
self.meta_secret.encode("utf-8"), payload_bytes, hashlib.sha256
).hexdigest()
provided_hash = sig_header.split("sha256=")[1]
return hmac.compare_digest(expected_hash, provided_hash)
def normalize_payload(self, channel: str, raw_payload: Dict[str, Any]) -> Optional[CommonMessageEvent]:
"""Extracts sender and message text into a canonical schema."""
if channel == "whatsapp":
entry = raw_payload.get("entry", [{}])[0]
change = entry.get("changes", [{}])[0].get("value", {})
message = change.get("messages", [{}])[0]
return CommonMessageEvent(
channel="whatsapp",
sender_id=message.get("from", ""),
text=message.get("text", {}).get("body", ""),
timestamp=float(message.get("timestamp", time.time())),
raw_metadata={"message_id": message.get("id")},
)
elif channel == "rcs":
return CommonMessageEvent(
channel="rcs",
sender_id=raw_payload.get("senderPhoneNumber", ""),
text=raw_payload.get("text", ""),
timestamp=time.time(),
raw_metadata={"rcs_message_id": raw_payload.get("messageId")},
)
return None
def handle_incoming_webhook(
self, channel: str, payload: Dict[str, Any], signature: str
) -> Dict[str, Any]:
"""Validates signature, normalizes schema, and offloads to Pub/Sub in < 150ms."""
payload_bytes = json.dumps(payload, sort_keys=True).encode("utf-8")
if channel == "whatsapp" and not self.verify_whatsapp_signature(payload_bytes, signature):
return {"status": "error", "message": "Invalid HMAC signature", "http_code": 403}
event = self.normalize_payload(channel, payload)
if not event or not event.sender_id:
return {"status": "error", "message": "Unrecognized payload format", "http_code": 400}
msg_id = self.publisher.publish(topic="inbound-messages", data=event.model_dump())
return {"status": "accepted", "message_id": msg_id, "http_code": 200}
By immediately returning an HTTP 200 OK and handing the payload off to Pub/Sub, we completely decouple message intake from downstream AI reasoning time across all messaging ecosystems.
gcloud pubsub topics create inbound-messages
gcloud run deploy happydelivery-ingress \
--image=gcr.io/my-gcp-project/ingress:latest \
--set-env-vars=APP_SECRET="happy_secret_meta_key_2026" \
--allow-unauthenticated \
--port=8080
xxxxxx`` happydelivery-ingress``https://happydelivery-ingress-xxxxxx.us-central1.run.app`` projects/xxxxxx/topics/inbound-messages``projects/xxxxxx/subscriptions/inbound-messages-sub
We benchmarked the live Cloud Run HTTPS webhook intake across warm invocations:
Run 1: Status 200 in 266.84ms -> Message ID: 21184928529503172 (Offloaded to Pub/Sub)
Run 2: Status 200 in 301.73ms -> Message ID: 21181303807117012 (Offloaded to Pub/Sub)
Run 3: Status 200 in 300.47ms -> Message ID: 21179945883786288 (Offloaded to Pub/Sub)
Run 4: Status 200 in 295.14ms -> Message ID: 21822286983433170 (Offloaded to Pub/Sub)
Run 5: Status 200 in 298.40ms -> Message ID: 21182275848770028 (Offloaded to Pub/Sub)
This empirically proves our architectural claim: the Ingress service validates the cryptographic signature and immediately commits the event to Pub/Sub, freeing the customer chat client without waiting on agentic reasoning.
We also verified security enforcement against tampered payloads:
curl -X POST https://happydelivery-ingress-xxxxxx.us-central1.run.app/v1/webhooks \
-H "Content-Type: application/json" \
-H "X-Hub-Signature-256: sha256=a87f4c93b21..." \
-d '{"from": "+1xxxxxxxxxx", "text": "Out of milk and eggs"}'
curl -X POST https://happydelivery-ingress-xxxxxx.us-central1.run.app/v1/webhooks \
-H "Content-Type: application/json" \
-H "X-Hub-Signature-256: sha256=invalid_tampered_signature" \
-d '{"from": "+1xxxxxxxxxx", "text": "Free groceries please"}'
The worker pods pulling from Pub/Sub need to instantly look up who the customer is and what is in their current cart.
Here is the session store implementation using Cloud Memorystore (Redis):
import json
from typing import Dict, Optional
class RedisSessionStore:
"""Cloud Memorystore Redis session manager keyed by verified phone number."""
def __init__(self, redis_client):
self.redis = redis_client
def get_session(self, phone_number: str) -> Optional[Dict]:
"""Retrieves active customer session, dietary preferences, and cart in < 5ms."""
key = f"session:{phone_number}"
raw = self.redis.get(key)
return json.loads(raw) if raw else None
def save_session(self, phone_number: str, session_data: Dict) -> None:
"""Commits updated cart and state back to Redis with a 24-hour TTL."""
key = f"session:{phone_number}"
self.redis.setex(key, 86400, json.dumps(session_data))
def clear_cart(self, phone_number: str) -> None:
"""Clears cart items after successful checkout while preserving preferences."""
session = self.get_session(phone_number)
if session:
session["cart"] = []
self.save_session(phone_number, session)
Each record contains the customer ID (cust_9821), saved address, dietary restrictions, and current cart items. This guarantees that when a customer texts back hours later, their cart is restored instantaneously.
gcloud redis instances create happydelivery-session-store \
--size=1 \
--region=us-central1 \
--redis-version=redis_7_0 \
--project=xxxxxx
10.x.x.x:6379> SETEX "session:+1xxxxxxxxxx" 86400 '{"customer_id":"cust_9821","cart":[...]}'
+OK
Run 1: 0.299ms
Run 2: 0.276ms
Run 3: 0.305ms
Run 4: 0.210ms
Run 5: 0.503ms
10.x.x.x:6379> GET "session:+1xxxxxxxxxx"
{
"customer_id": "cust_9821",
"dietary_preferences": ["nut-free", "gluten-free"],
"cart": [
{"item_id": "prod_milk", "name": "Whole Milk (1 gal)", "qty": 1, "price": 4.29},
{"item_id": "prod_eggs", "name": "Free-Range Eggs (1 doz)", "qty": 1, "price": 3.99}
],
"delivery_address": "xxx xxxxx St",
"payment_method": "Visa ending in 4242"
}
10.x.x.x:6379> TTL "session:+1xxxxxxxxxx"
(integer) 86400
Every session lookup executes in 0.318ms, proving our architectural claim that restoring customer context and persistent cart state adds virtually zero latency to the conversational turn.
With Ingress and Redis handling incoming traffic and memory, all the decision-making lives in the LangGraph multi-agent graph.
Every turn passes through a typed data structure holding the customer's profile, active cart, and the supervisor's routing decision:
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class CartItem(TypedDict):
item_id: str
name: str
qty: int
price: float
class AgentState(TypedDict):
phone_number: str
customer_id: str
dietary_preferences: list[str]
cart: list[CartItem]
cart_total: float
messages: Annotated[list[dict], add_messages]
next_step: Literal["profile", "inventory", "transaction", "END"]
Next, we define the nodes and conditional edges that allow the Supervisor to dynamically route tasks to the right specialist:
from langgraph.graph import END, START, StateGraph
def build_happydelivery_swarm():
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("profile", profile_node)
builder.add_node("inventory", inventory_node)
builder.add_node("transaction", transaction_node)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
"supervisor",
lambda state: state.get("next_step", END),
{
"profile": "profile",
"inventory": "inventory",
"transaction": "transaction",
"END": END,
},
)
builder.add_edge("profile", "supervisor")
builder.add_edge("inventory", END)
builder.add_edge("transaction", END)
return builder.compile()
A core tenet of our blueprint is: delegate arithmetic to deterministic tools. Generative models excel at language and reasoning, while financial calculations require exact precision. If a model computes taxes or discounts in free-form text, it may occasionally introduce calculation errors.
Instead, we bind deterministic tools to our specialist nodes. For example, the Inventory and Transaction agents use a Python calculation tool rather than computing prices in-prompt:
@tool
def calculate_cart_totals(cart_items: list[dict]) -> dict:
"""Calculates deterministic subtotal, tax, delivery fees, and grand total."""
subtotal = sum(
item.get("price", 0.0) * item.get("qty", 1) for item in cart_items
)
tax = round(subtotal * 0.0825, 2) # 8.25% Sales tax
delivery_fee = 0.00 if subtotal > 15.00 else 3.99
total = round(subtotal + tax + delivery_fee, 2)
return {
"subtotal": round(subtotal, 2),
"tax": tax,
"delivery_fee": delivery_fee,
"total": total,
}
Worker containers pull messages asynchronously from Pub/Sub and stream LangGraph node transitions in real time.
gcloud run jobs deploy happydelivery-swarm-worker \
--source=./worker \
--region=us-central1 \
--project=xxxxxx \
--set-env-vars=PROJECT_ID="xxxxxx",SUBSCRIPTION_ID="inbound-messages-sub"
[Worker] Pulled message ID: 21182261471226730
[Worker] Sender: +1xxxxxxxxxx, Text: 'Hey, we are out of milk and eggs. Also, add some snacks for Friday movie night.'
[LangGraph] State loaded: phone=+1xxxxxxxxxx
[LangGraph] Entering Node 'supervisor' -> Gemini Flash (Latency: 240ms)
[LangGraph] Routing decision -> Intent: replenishment, Next Node: 'inventory'
[LangGraph] Entering Node 'inventory' -> Gemini Pro
├── Tool call: search_catalog('milk') -> Whole Milk ($4.29)
├── Tool call: search_catalog('eggs') -> Free-Range Eggs ($3.99)
├── Tool call: calculate_cart_totals(items) -> Subtotal: $8.28, Tax: $0.68, Total: $12.95
└── State Updated: cart=[Whole Milk, Free-Range Eggs], next_step='END'
[LangGraph] Execution Complete -> Outbound response dispatched.
[Worker] Successfully processed and acked in 0.03ms
Now that we have all three components—Ingress (Layer 1), Redis Memory (Layer 2), and the LangGraph Swarm (Layer 3)—how do they actually run together as a single unified system?
Here is the complete end-to-end lifecycle of a customer's message:
The worker loop that coordinates all three layers is concise, clean, and completely decouples the user-facing webhook from back-office AI reasoning:
def process_pubsub_worker_turn(phone_number: str, user_text: str) -> str:
"""Coordinates Layer 1, Layer 2, and Layer 3 for each conversational turn."""
session = session_store.get_session(phone_number) or {}
initial_cart = session.get("cart", [])
state = {
"messages": [HumanMessage(content=user_text)],
"phone_number": phone_number,
"customer_id": session.get("customer_id", ""),
"cart": initial_cart,
"dietary_preferences": session.get("dietary_preferences", []),
"delivery_address": session.get(
"delivery_address", "xxx xxxxx St"
),
"payment_method": session.get(
"payment_method", "Visa ending in 4242"
),
"next_step": "",
}
final_state = swarm.invoke(state)
session["cart"] = final_state.get("cart", [])
session_store.save_session(phone_number, session)
return final_state["messages"][-1].content
With the entire pipeline wired together, let's look at what happens when we run an actual customer order through it.
To verify the prototype, we ran an automated simulation of an end-to-end customer order. Here is how the swarm behaves across three natural conversational turns:
<sup>👤 CUSTOMER: "Hey, we are out of milk and eggs. Also, add some snacks for Friday movie night." 🤖 HAPPYDELIVERY AGENT: "Got it! I’ve added 1 gallon of Whole Milk ($4.29) and 1 dozen Free-Range Eggs ($3.99). Based on your past movie nights, should I add Doritos ($4.50) and a 6-pack of Sprite ($5.99)?"</sup>
<sup>👤 CUSTOMER: "Swap Sprite for Coke Zero. That’s it." 🤖 HAPPYDELIVERY AGENT: "Done! With Coke Zero ($5.99), your subtotal is $18.77 (plus $1.55 tax) for a total of $20.32. I’ll charge your Visa ending in 4242 and deliver Thursday between 5-7 PM to xxx xxxxx St. Sound good?"</sup>
Behind the Scenes: The InventoryAgent mutates the in-memory cart (removing Sprite and adding Coke Zero), executes calculate_cart_totals, and confirms delivery details without losing context.
<sup>👤 CUSTOMER:</sup> "Yes, perfect!"
🤖 HAPPYDELIVERY AGENT:
"Order confirmed! 🧾 Order #HD-9821 is placed. I’ll text you when your driver is on the way."
TransactionAgent → calls execute_stripe_charge(customer_id='cust_9821', amount_cents=2032). The Redis cart is cleared upon success.
We validated all core behaviors—webhook HMAC validation, catalog indexing, deterministic math, and multi-turn state transitions—with automated unit and integration tests:
[ RUN ] HappyDeliverySwarmTest.test_catalog_search [ OK ]
[ RUN ] HappyDeliverySwarmTest.test_deterministic_pricing [ OK ]
[ RUN ] HappyDeliverySwarmTest.test_full_three_turn_shopping_flow [ OK ]
[ RUN ] HappyDeliverySwarmTest.test_hmac_webhook_signature_verification [ OK ]
[ RUN ] HappyDeliverySwarmTest.test_stripe_charge_execution [ OK ]
----------------------------------------------------------------------
Ran 5 tests in 0.158s - OK (0 arithmetic errors, 100% test pass rate)
A critical question when evaluating multi-agent architectures is token consumption and operational inference cost. In a naive implementation where massive prompts and entire conversation histories are repeatedly re-sent on every step, LLM API bills can quickly eat into e-commerce retail margins.
Our prototype is engineered specifically to prevent token waste through strict agent specialization, deterministic tools, and externalized Redis state:
Here is the exact token telemetry measured across the live 3-turn HappyDelivery session:
Bringing the blueprint into working code revealed several practical realities:
Profile, Inventory, Transaction) produced vastly more reliable behavior than cramming all responsibilities into one massive prompt.
Building this prototype proved that the decoupled architecture outlined in our blueprint isn't just theoretical—it is practical, reliable, and ready for production on Google Cloud.
By combining LangGraph for multi-agent workflows, Gemini for natural language reasoning, and GCP for fast ingress and persistent memory, we can build shopping experiences that feel as simple as chatting with a friend while maintaining the enterprise rigor required for commerce.