Event-Driven Multi-Agent Pipelines on Google Cloud with ADK, Gemma 3 and Eventarc

This article provides a step by step architectural and deployment guide for StreamForge, a real-time...

This article provides a step by step architectural and deployment guide for StreamForge, a real-time event-driven multi-agent pipeline built with the Google Agent Development Kit (ADK), open-weights Gemma 3 (gemma3:12b), Google Cloud Eventarc, Cloud Pub/Sub, and Google Cloud Run.

Repository: github.com/RobinaMirbahar/streamforge-event-adk

Enterprise data platforms operate on real-time state changes: a product engineering team uploads a release specification (vertex-vector-sync-v4.2.json) to a Google Cloud Storage bucket, a streaming job commits a partition to BigQuery, or a partner service publishes a schema event to Cloud Pub/Sub. Connecting Google Cloud Eventarc directly to a multi-agent Google ADK workflow runtime on Cloud Run triggers autonomous workflows within 200 ms of an object landing in storage while scaling to 0 instances when the bucket is idle.

Why Combine Eventarc with ADK ParallelAgent and LoopAgent?

Processing a multi-regional product release requires three distinct tasks: extracting technical benchmarks from the raw JSON payload, translating launch messaging across target locales (en_US, de_DE, ja_JP), and simulating 12-month Annual Recurring Revenue (ARR) expansion under regional B2B currency rules. Running those three steps sequentially increases end-to-end latency, while running them in parallel without a verification step risks publishing unrounded foreign-exchange prices.

End-to-End Event-Driven Architecture on Google Cloud

1. Tier 1: Event Ingestion and Routing: Uploading releases/2026-10/vertex-vector-sync-v4.2.json to gs://streamforge-launch-inbox emits a native google.cloud.storage.object.v1.finalized event. Google Cloud Eventarc wraps the notification in a CNCF CloudEvents v1.0 envelope (ce-id, ce-type, ce-source, ce-time) and delivers it over an OIDC-authenticated Cloud Pub/Sub push subscription.
2. Tier 2: Serverless Multi-Agent Runtime (Cloud Run): A private FastAPI receiver checks the incoming ce-id header (ce-98104271-gcs-finalized) against Cloud Firestore. New events execute the Google ADK workflow (ParallelAgent → LoopAgent → LlmAgent) powered by Gemma 3 12B via LiteLlm.
3. Tier 3: State and Downstream Sink: The final publisher agent validates the output against the Pydantic EventDrivenLaunchManifest schema and commits both the manifest (MANIFEST-SF-VECTOR-SYNC-4.2) and the ce-id deduplication record to Cloud Firestore.

Orchestrating ParallelAgent Fan-Out and LoopAgent Self-Correction

In Google ADK, all branches inside a ParallelAgent share the same InvocationContext.session.state. Assigning a distinct output_key to each branch isolates their writes cleanly:

• TechnicalSpecExtractorAgent writes engineering benchmarks to output_key="extracted_specs".
• GlobalLocalizationAgent writes localized headlines and draft currency tiers to output_key="localized_copy".
• RevenuePricingSimulatorAgent writes the 12-month ARR forecast and regional FX compliance rules to output_key="pricing_simulation".

Once all 3 parallel branches complete, QualityCritiqueRefinementLoop (LoopAgent with max_iterations=3) invokes LaunchReadinessCriticAgent to audit the combined state. When GlobalLocalizationAgent converts the $499/mo Growth Tier at the raw 0.92 EUR exchange rate (€459.08/mo), Iteration 1 of the LoopAgent flags the violation against the de_DE rule requiring VAT-exclusive B2B prices rounded to the nearest €5 (quality_score = 0.72) and updates the pricing tier to €460/mo. On Iteration 2, the audit reaches quality_score = 1.0 and invokes exit_quality_loop(), setting tool_context.actions.escalate = True to exit the loop immediately.

At This Point You Should Have

• Python 3.10+ (3.11 recommended) with google-adk>=1.0.0, litellm, fastapi, and pydantic installed.
• A Google Cloud project with Cloud Run, Eventarc, Cloud Pub/Sub, Cloud Storage, and Cloud Firestore APIs enabled.
• Access to Gemma 3 12B (ollama_chat/gemma3:12b locally via Ollama or google/gemma-3-12b-it on a Cloud Run NVIDIA L4 24 GB GPU service).

Step 1: Define the CloudEvent and Launch Manifest Schemas

The pipeline enforces typed contracts at both boundaries: CloudEventEnvelope validates the incoming Eventarc HTTP headers and payload, while EventDrivenLaunchManifest constrains the final ReleasePublisherAgent output in streamforge_agent/schemas.py.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

Step 2: Implement the Parallel Enrichment and Loop Escalation Tools

In streamforge_agent/tools.py, each specialist tool exposes typed parameters and docstrings for Gemma 3 function calling, while exit_quality_loop sets tool_context.actions.escalate = True on the injected ADK ToolContext.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

Step 3: Wire ParallelAgent, LoopAgent, and SequentialAgent

In streamforge_agent/agent.py, build_streamforge_event_pipeline() composes the 3 parallel enrichment agents, the iterative quality critic loop, and the schema-constrained publisher agent into a single parent SequentialAgent.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

Step 4: Verify the Pipeline and CloudEvent Deduplication Locally

Run scripts/simulate_cloudevent_pipeline.py to execute the 3-branch ParallelAgent fan-out, inspect the 2-iteration LoopAgent self-correction trace, and verify the ce-id idempotency guard against a duplicate Pub/Sub delivery.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

Step 5: Deploy the Private Cloud Run Receiver and Eventarc Trigger

Run deploy/deploy_eventarc_adk.sh to deploy the container to Google Cloud Run (2 vCPU, 2 GiB memory, --no-allow-unauthenticated, --min-instances=0, --max-instances=10) and bind the Eventarc trigger to google.cloud.storage.object.v1.finalized.

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

==> [1/6] Enabling Cloud Run, Eventarc, Pub/Sub, Storage, and Firestore APIs...
==> [2/6] Creating Cloud Storage Inbox Bucket & Dead-Letter Pub/Sub Topic...
==> [3/6] Creating Least-Privilege Service Accounts...
==> [4/6] Deploying StreamForge ADK Service to Cloud Run (private ingress)...
Service [streamforge-event-adk] revision [streamforge-event-adk-00001-x8k] has been deployed and is serving 100 percent of traffic.
==> [5/6] Granting Cloud Run Invoker + Eventarc EventReceiver to Eventarc SA...
==> [6/6] Creating Eventarc Trigger for google.cloud.storage.object.v1.finalized...
Created trigger [streamforge-gcs-release-trigger] in [us-central1].

Official Google Cloud and ADK Documentation

• Google Agent Development Kit (ADK) Documentation
• Google ADK Workflow Agents (SequentialAgent, ParallelAgent, LoopAgent)
• Google ADK Python SDK Repository (google/adk-python)
• Google Gemma Open Models Documentation
• Google Cloud Eventarc Documentation
• Google Cloud Run Documentation
• Google Cloud Pub/Sub Push Subscriptions Guide
• Google Cloud Firestore Documentation

Summary

The goal of this article was to build and verify a real-time, event-driven multi-agent pipeline on Google Cloud using Eventarc, Cloud Pub/Sub, Cloud Run, open-weights Gemma 3, and the Google Agent Development Kit. The key to the solution was pairing ParallelAgent for concurrent 3-branch fan-out with LoopAgent for iterative cross-branch self-correction and a Cloud Firestore ce-id deduplication ledger. The verification results were:

• 🟢 Concurrent Fan-Out (ParallelAgent): Extracted engineering specs (73.8% latency drop from 145 ms to 38 ms at 250,000 rows/sec), localized copy across 3 locales (en_US, de_DE, ja_JP), and forecasted $2,856,000 in 12-month Net New ARR lift across 340 upgraded accounts (18.5% of 1,840 active Growth Tier accounts at $8,400 annual expansion).
• 🟢 Self-Correcting Quality Loop (LoopAgent): Detected the €459.08/mo and ¥75,848/mo currency formatting violations on Iteration 1 (quality_score = 0.72), repaired them to €460/mo and ¥75,800/mo, and exited cleanly on Iteration 2 (quality_score = 1.0) via exit_quality_loop().
• 🟢 At-Least-Once Idempotency: Short-circuited duplicate ce-98104271-gcs-finalized redelivery in < 5 ms with 0 redundant parallel branches or LLM calls.

The strategy for orchestrating event-driven multi-agent pipelines with Google ADK and Gemma 3 on Cloud Run was validated with an incremental step by step approach.