Data Engineering Orchestrator Agent (Google ADK & A2A)

A production-grade, multi-agent data orchestrator built using Google’s Agent Development Kit (ADK) and the A2A SDK. This application implements a powerful Orchestrator-Worker architecture that separates a central "brain" (Gemini 3.5 Flash) from specialized GCP database-specific executables and a remote Data Engineering Specialist agent.

---

🌟 Key Features

• Orchestrator-Worker Pattern: The agent is split into the Orchestrator (the natural language UX, managing conversation flow) and the Workers (the tools doing the heavy lifting).
• Secure Agent-to-Agent (A2A) Delegation: Integrates with the remote GCP-hosted Data Engineering Specialist (dataengineeringagent) over authorized REST/HTTP.
• Dynamic MCP Registry Toolsets: Automatically retrieves full MCP toolsets (BigQuery, GCS, Composer) directly from Google Cloud via the AgentRegistry resource configurations.
• Pre-configured BigQuery Analytics: Possesses deep semantic understanding of e-commerce databases (e.g., table structure, primary keys, metric formulas, and join logic for thelook_ecommerce).
• Strict Access Control & Guardrails: Enforces semantic checks on table access for data-engineering workflows, rejecting unauthorized tables while routing appropriate commands to styling templates (forcing uppercase SQL).

---

📂 Project Structure

Data-engineering-orchestrator-agent/
├── agent.py            # Main entrypoint: Agent configuration, tools, and orchestration logic
├── .adk/
│   └── session.db      # Local SQLite database for conversation state and session history
└── README.md           # Project documentation and guide

---

⚙️ Core Configuration Deep Dive

1. The Orchestrator (agent.py)

Configured via the ADK LlmAgent framework, using gemini-3.5-flash. The agent uses structured, robust base instructions containing join paths, metrics calculations, and safety boundaries:

root_agent = LlmAgent(
    model="gemini-3.5-flash",
    name="unicorn_orchestrator",
    instruction=base_instruction + "...",
    tools=[execute_data_engineering_task, execute_analytics_query, execute_composer_task, execute_gcs_task],
)

2. Remote A2A Tooling (execute_data_engineering_task)

Uses local Google Cloud Application Default Credentials (ADC) to request a short-lived bearer token, connects to the A2ACardResolver, and establishes an asynchronous stream-based command session. It attaches specific conversation metadata and a style-guide instruction payload to enforce code styling standards on the remote agent.

3. Google Cloud Agent Registry

Directly fetches server-side tool schemas via AgentRegistry for three distinct GCP database/storage domains:

• BigQuery MCP (execute_analytics_query)
• Composer MCP (execute_composer_task)
• Google Cloud Storage MCP (execute_gcs_task)

---

🚀 Getting Started

Prerequisites

• Python 3.10 or higher
• Google Cloud CLI (gcloud) installed and configured
• Google AI Studio API Key

Installation

1. Clone this repository to your local machine.

2. Install dependencies:

   pip install -r requirements.txt

   (Ensure you have packages like google-adk, a2a-sdk, mcp, and toolbox-core installed).

3. Authenticate with Google Cloud: Ensure your local environment has active credentials to allow the A2A tool to call GCP endpoints:

   gcloud auth application-default login

4. Set Your AI Studio API Key: Open agent.py and replace PUT_YOUR_API_KEY_HERE on line 28 with your actual Gemini API key, or set it via your shell environment:

   export GEMINI_API_KEY="your_api_key_here"

---

🖥️ Running the Agent

Launch the standard ADK local web dashboard to interact with the orchestrator:

adk web --allow_origins="regex:.*"

Once running, navigate to the local address displayed in your console (usually http://localhost:8000) to access the browser-based chat environment.

---

💬 Sample Prompts to Try

📈 Data Analytics (BigQuery)

"Which product categories brought in the highest revenue last month in theLook eCommerce, and what is our unique visitor count for those categories?"

🛠️ Data Engineering & Dataform (A2A)

"Create a new Dataform repository workspace flow to deduplicate and load our sales table into BigQuery."

🔗 Cloud Composer & Orchestration

"Trigger the daily ETL DAG on our us-central1 Airflow environment."