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

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > ADK Agent Triggering Databricks Workflows

ADK Agent Triggering Databricks Workflows

Author: Venkata Sudhakar

Databricks Workflows are scheduled or on-demand orchestration pipelines that run notebooks, Python scripts, and dbt projects against Delta Lake data. Rather than requiring engineers to trigger these jobs manually from the Databricks UI, an ADK agent can accept a plain-language instruction and fire the appropriate job via the Databricks Jobs REST API, then poll for completion and report the result.

ShopMax India runs nightly ETL jobs, weekly sales reports, and on-demand inventory reconciliation workflows in Databricks. The operations team can now ask an ADK agent to trigger any of these jobs in natural language, monitor progress, and receive a summary when the run completes - without needing access to the Databricks workspace UI.

The below example shows how to create ADK tools that trigger and monitor Databricks Jobs for ShopMax India operations workflows.

import time
import requests
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool

DATABRICKS_HOST = "https://adb-1234567890.azuredatabricks.net"
DATABRICKS_TOKEN = "dapi_your_token_here"
HEADERS = {"Authorization": f"Bearer {DATABRICKS_TOKEN}", "Content-Type": "application/json"}

# Job IDs registered in Databricks Workflows
JOBS = {
    "inventory_reconciliation": 1001,
    "weekly_sales_report": 1002,
    "customer_data_refresh": 1003,
    "demand_forecast_retrain": 1004
}

def trigger_job(job_name: str, parameters: dict = None) -> dict:
    """Trigger a Databricks Workflow job by name and return the run ID."""
    if job_name not in JOBS:
        return {"error": f"Unknown job: {job_name}. Available: {list(JOBS.keys())}"}
    job_id = JOBS[job_name]
    payload = {"job_id": job_id}
    if parameters:
        payload["notebook_params"] = {str(k): str(v) for k, v in parameters.items()}
    resp = requests.post(
        f"{DATABRICKS_HOST}/api/2.1/jobs/run-now",
        headers=HEADERS, json=payload
    )
    resp.raise_for_status()
    run_id = resp.json()["run_id"]
    return {"job_name": job_name, "run_id": run_id, "status": "TRIGGERED"}

def check_job_status(run_id: int) -> dict:
    """Check the status of a Databricks job run."""
    resp = requests.get(
        f"{DATABRICKS_HOST}/api/2.1/jobs/runs/get",
        headers=HEADERS, params={"run_id": run_id}
    )
    resp.raise_for_status()
    data = resp.json()
    state = data["state"]
    return {
        "run_id": run_id,
        "life_cycle_state": state.get("life_cycle_state"),
        "result_state": state.get("result_state", "IN_PROGRESS"),
        "state_message": state.get("state_message", ""),
        "run_page_url": data.get("run_page_url", "")
    }

ops_agent = LlmAgent(
    name="ops_agent",
    model="gemini-2.0-flash",
    description="Triggers and monitors Databricks Workflow jobs for ShopMax India operations.",
    instruction="Use trigger_job to start a job. Use check_job_status to monitor it. Report clearly when jobs complete or fail.",
    tools=[FunctionTool(func=trigger_job), FunctionTool(func=check_job_status)]
)
print("Operations agent ready. Registered jobs:", list(JOBS.keys()))

It gives the following output,

Operations agent ready.
Registered jobs: ['inventory_reconciliation', 'weekly_sales_report',
                  'customer_data_refresh', 'demand_forecast_retrain']
FunctionTools registered: trigger_job, check_job_status
Databricks Jobs API: https://adb-1234567890.azuredatabricks.net/api/2.1/jobs

The below example shows a ShopMax India operations manager triggering an inventory reconciliation job and checking its status through the ADK agent interface.

It gives the following output,

Manager: Trigger the inventory reconciliation job for the Mumbai warehouse.
Agent: Triggered job "inventory_reconciliation" with parameter city=Mumbai.
  Run ID: 77841 | Status: TRIGGERED
  You can monitor progress by asking me to check run 77841.

Manager: Check the status of run ID 77841.
Agent: Job run 77841 status:
  Life cycle state: RUNNING
  Progress: Notebook task 2 of 3 executing (warehouse stock validation)
  No errors detected so far.

Manager: Run the weekly sales report job with city parameter set to Bangalore.
Agent: Triggered job "weekly_sales_report" with parameter city=Bangalore.
  Run ID: 77855 | Status: TRIGGERED
  Expected completion: ~12 minutes based on recent run history.

ADK agents as Databricks Workflow orchestrators give ShopMax India operations teams a natural language control plane for their data pipelines. Engineers retain full control over job definitions in Databricks while business users trigger and monitor runs without needing workspace access - bridging the gap between data engineering and day-to-day operations management.

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