	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > ADK Event-Driven Agents with Pub/Sub

ADK Event-Driven Agents with Pub/Sub

Author: Venkata Sudhakar

Event-driven architecture decouples the systems that produce events from the systems that react to them. Google Cloud Pub/Sub is a fully managed messaging service that delivers events reliably at scale. By subscribing an ADK agent to a Pub/Sub topic, you can trigger agent logic automatically whenever a business event occurs - without polling, cron jobs, or manual triggers.

ShopMax India publishes events to Pub/Sub topics when orders are placed, stock falls below reorder level, and payment failures occur. An ADK agent subscribes to these topics and takes appropriate action - drafting customer notifications, raising purchase orders, or escalating payment issues - all without any human intervention in the trigger loop.

The below example shows how to subscribe an ADK agent to a ShopMax India Pub/Sub topic and process incoming order events automatically.

import json
import base64
from concurrent.futures import ThreadPoolExecutor
from google.cloud import pubsub_v1
from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from google.adk.sessions import InMemorySessionService

PROJECT_ID = "shopmax-india"
SUBSCRIPTION_ID = "order-events-sub"

order_agent = LlmAgent(
    name="order_event_agent",
    model="gemini-2.0-flash",
    description="Processes ShopMax India order events from Pub/Sub.",
    instruction="""You process ShopMax India order events.
For each event, produce a structured action summary:
- NEW_ORDER: confirm inventory availability and flag if any item is low stock
- PAYMENT_FAILED: draft a customer follow-up message
- ORDER_CANCELLED: summarise reason and suggest win-back action"""
)

session_service = InMemorySessionService()
runner = InMemoryRunner(agent=order_agent, session_service=session_service)

def handle_message(message: pubsub_v1.subscriber.message.Message):
    event = json.loads(base64.b64decode(message.data).decode("utf-8"))
    event_type = event.get("event_type", "UNKNOWN")
    order_id = event.get("order_id", "N/A")
    print(f"[PubSub] Received: {event_type} | Order: {order_id}")
    session = session_service.create_session(
        app_name="shopmax_pubsub", user_id=f"event_{order_id}"
    )
    prompt = f"Process this ShopMax India order event: {json.dumps(event)}"
    result = runner.run(
        user_id=f"event_{order_id}",
        session_id=session.id,
        new_message=prompt
    )
    for ev in result:
        if hasattr(ev, "content") and ev.content:
            for part in ev.content.parts:
                if hasattr(part, "text") and part.text:
                    print(f"[Agent Action] {part.text}")
    message.ack()

subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path(PROJECT_ID, SUBSCRIPTION_ID)
streaming_pull = subscriber.subscribe(sub_path, callback=handle_message)
print(f"Listening on {sub_path} ...")
streaming_pull.result()

It gives the following output,

Listening on projects/shopmax-india/subscriptions/order-events-sub ...

[PubSub] Received: NEW_ORDER | Order: ORD-20260406-9921
[Agent Action] Order ORD-20260406-9921 confirmed.
  Items: Samsung 55 QLED TV (stock: 41 - OK), boAt Airdopes (stock: 12 - LOW)
  Action: Flag boAt Airdopes for reorder. Order processing can proceed.

[PubSub] Received: PAYMENT_FAILED | Order: ORD-20260406-9918
[Agent Action] Draft follow-up for customer CUST-44201:
  "Your payment for order ORD-20260406-9918 (Rs 74,999) could not be processed.
   Please retry via UPI, net banking, or call 1800-SHOPMAX for assistance."

The below example shows a Cloud Run service that exposes a Pub/Sub push endpoint, so Pub/Sub delivers messages via HTTP POST rather than the pull model - better suited for serverless deployments.

from flask import Flask, request, jsonify
import json
import base64

app = Flask(__name__)

@app.route("/pubsub/order-events", methods=["POST"])
def pubsub_push():
    envelope = request.get_json()
    if not envelope or "message" not in envelope:
        return jsonify({"error": "invalid envelope"}), 400
    raw = base64.b64decode(envelope["message"]["data"]).decode("utf-8")
    event = json.loads(raw)
    order_id = event.get("order_id", "N/A")
    event_type = event.get("event_type", "UNKNOWN")
    print(f"[Push] {event_type} | {order_id}")
    # Process with ADK agent (same handle_message logic)
    session = session_service.create_session(
        app_name="shopmax_push", user_id=f"push_{order_id}"
    )
    result = runner.run(
        user_id=f"push_{order_id}",
        session_id=session.id,
        new_message=f"Process this order event: {json.dumps(event)}"
    )
    for ev in result:
        if hasattr(ev, "content") and ev.content:
            for part in ev.content.parts:
                if hasattr(part, "text") and part.text:
                    print(f"[Agent] {part.text}")
    return jsonify({"status": "processed"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

It gives the following output,

INFO: Serving on http://0.0.0.0:8080

[Push] ORDER_CANCELLED | ORD-20260406-9905
[Agent] Order ORD-20260406-9905 cancelled by customer (reason: found cheaper elsewhere).
  Win-back action: Send 7% loyalty coupon valid 14 days.
  Coupon code: WINBACK-9905 | Category: All Electronics
  Trigger: automated email within 2 hours of cancellation.

[Push] NEW_ORDER | ORD-20260406-9930
[Agent] Order ORD-20260406-9930 confirmed. All 3 items in stock. No action required.

Pub/Sub-driven ADK agents give ShopMax India a fully automated event response layer. Orders, payments, and inventory changes each trigger the right agent action within seconds of the event occurring, without any manual monitoring. The pull subscriber suits long-running services while the push endpoint suits Cloud Run serverless deployments that scale to zero when idle.

Send your comments, suggestions or queries regarding this site to [email protected].