Demo Building Google ADK Agent With BigQuery MCP

In this lesson we build a churn-detection agent that uses the ADK MCP Toolbox to query BigQuery. The Toolbox runs as a helper subprocess (via stdio), publishes BigQuery tools (for example execute_sql, get_table_info, list_table_ids), and the agent calls those tools while composing its own SQL. Below you’ll find a clear, step-by-step walkthrough covering:

imports and environment bootstrap
model initialization
configuring the MCP Toolbox bridge to BigQuery
agent instructions and schema guidance
the runner loop that sends a user query and streams the agent’s final reply

This pattern is reusable for other data sources (Databricks, Snowflake) with minor adjustments.

Imports and environment bootstrap

Load environment variables and suppress the ADK experimental warning before importing ADK modules. The ADK issues a one-time warning for experimental features (MCP support), so filter warnings before ADK imports:

# python
import asyncio
import os
import warnings
from dotenv import load_dotenv

# Load .env into process environment
load_dotenv()

# ADK warns once per experimental feature it touches (MCP support is one of
# them). Filter before the ADK imports -- the first warning fires at import time.
warnings.filterwarnings("ignore", message=r".*\[EXPERIMENTAL\].*")

Then import ADK, MCP, and helper types used by the runner and content model:

# python
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.genai import types
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

Quick reference:

InMemoryRunner runs the agent and maintains session state.
types and types.Part let you construct multimodal messages (text, files, audio, etc.).
McpToolset, StdioConnectionParams, and StdioServerParameters create the MCP/stdio bridge to the Toolbox subprocess.

.env example

Copy .env.example to .env and fill in the actual keys. Never commit a file containing secrets.

Copy .env.example to .env and fill in your actual keys. Never commit .env.

# --- Model (all scripts) - Claude via KodeKey ----
KODEKEY_BASE_URL=https://your-kodekey-endpoint/v1
KODEKEY_API_KEY=your-kodekey-api-key

# --- BigQuery (churn_agent_bigquery.py) ----
# Auth comes from ADC: run `gcloud auth application-default login` first.
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT=your-gcp-project-id
GOOGLE_CLOUD_LOCATION=us-xxxxx
# Optional: path to the toolbox binary (default ./toolbox)
# --- Databricks / Snowflake entries omitted for brevity ---

Environment variables summary:

Variable	Purpose	Example
`KODEKEY_BASE_URL`	KodeKey/OpenAI-compatible API base URL	`https://your-kodekey-endpoint/v1`
`KODEKEY_API_KEY`	API key for the model endpoint	`your-kodekey-api-key`
`GOOGLE_CLOUD_PROJECT`	BigQuery project used by Toolbox (ADC also required)	`your-gcp-project-id`
`TOOLBOX_BIN`	Optional path to the `toolbox` binary if not `./toolbox`	`./toolbox`

Model initialization

Create a LiteLlm model instance using your KodeKey/OpenAI-compatible endpoint and model name. Replace the model name with the model available on your endpoint.

# python
MODEL = LiteLlm(
    model="openai/claude-haiku-4-5-20251001",
    api_base=os.environ["KODEKEY_BASE_URL"],
    api_key=os.environ["KODEKEY_API_KEY"],
)

Grab the Google Cloud project and optional toolbox path from the environment:

# python
PROJECT = os.environ["GOOGLE_CLOUD_PROJECT"]
TOOLBOX = os.environ.get("TOOLBOX_BIN", "./toolbox")

Configure the MCP Toolbox bridge to BigQuery

The ADK launches the Toolbox binary in stdio mode; Toolbox connects to BigQuery and exposes a set of tools.

# python
bigquery_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command=TOOLBOX,
            # --log-level error keeps Toolbox startup chatter off your terminal.
            args=["--prebuilt", "bigquery", "--stdio", "--log-level", "error"],
            # Toolbox reads BIGQUERY_PROJECT; pass along environment for ADC (HOME, PATH, etc.)
            env={**os.environ, "BIGQUERY_PROJECT": PROJECT},
        ),
        timeout=60,
    ),
    # Give the model just the two tools it needs, not all prebuilt tools.
    tool_filter=["execute_sql", "get_table_info"],
)

Notes:

args uses --prebuilt bigquery --stdio. Toolbox connects to BigQuery using Application Default Credentials (run gcloud auth application-default login if needed).
env ensures Toolbox sees BIGQUERY_PROJECT, so it can set the project for queries.
tool_filter restricts the model to only the tools it needs for safety.

Toolbox uses Application Default Credentials (ADC). Run gcloud auth application-default login and verify ADC access before starting the agent. Always close Toolbox from the same task that started it to avoid noisy shutdown tracebacks.

Toolset parameter summary:

Parameter	Purpose
`command`	Path to the `toolbox` binary
`args`	CLI args passed to Toolbox (prebuilt, stdio, log-level, etc.)
`env`	Environment passed to Toolbox (including `BIGQUERY_PROJECT`)
`tool_filter`	Whitelist of exposed tool names (e.g., `execute_sql`)

Agent definition and instructions

Create an LlmAgent named churn_agent. Provide the task instruction plus the BigQuery table schema so the model can reason without querying the churned label:

# python
agent = LlmAgent(
    name="churn_agent",
    model=MODEL,
    instruction=(
        "You help a customer-success team spot churn risk. The data lives in the "
        "BigQuery table `analytics.customer_features` -- one row per customer with "
        "columns: customer_id, region, plan_tier, tenure_months, usage_last_30d, "
        "usage_delta_30d, support_tickets_90d, upgrade_eligible. Use the BigQuery "
        "tools to run SQL to fetch the customer named by the user. DO NOT query the "
        "`churned` column. Interpret the signals: low `usage_last_30d` or a sharply "
        "negative `usage_delta_30d` means disengagement; high `support_tickets_90d` "
        "means friction; short `tenure_months` is riskier. State the risk level "
        "(low/medium/high), cite the signals, and suggest one retention action."
    ),
    tools=[bigquery_tools],
)

Why include the schema in the instruction?

It enables the model to reason about which columns are available without repeatedly querying schema.
It explicitly forbids using the churned label so the agent cannot return ground-truth labels.

Preview of the BigQuery table (for context)

Below is a preview screenshot of analytics.customer_features from BigQuery. This demonstrates the table layout and the churned column (which the agent must not query):

A screenshot of the Google Cloud BigQuery console showing the "customer_features" table in preview mode. It lists customer records with columns like customer_id, region, usage metrics, plan_tier, tenure_months, and churned.

The schema is also visible here:

Screenshot of the Google Cloud BigQuery console showing the schema for a "customer_features" table. The table fields (customer_id, region, upgrade_eligible, usage_last_30d, usage_delta_30d, support_tickets_90d, plan_tier, tenure_months, churned) and the Explorer sidebar are visible.

Main runner and session loop

The main function runs the agent with InMemoryRunner, opens a session, sends a user question, and streams events until the agent emits a final response. It also ensures the Toolbox subprocess is closed cleanly.

# python
async def main():
    # Create a runner for the agent and a session
    runner = InMemoryRunner(agent=agent, app_name="churn")
    s = await runner.session_service.create_session(app_name="churn", user_id="me")

    # Build the user message using ADK types (allows multimodal parts)
    msg = types.Content(
        role="user",
        parts=[types.Part(text="Is customer CUST-0003115 at risk?")]
    )

    try:
        # Run the agent asynchronously and stream events
        async for event in runner.run_async(user_id="me", session_id=s.id, new_message=msg):
            # When the agent emits the final response, print its pieces
            if event.is_final_response() and event.content:
                print("".join(p.text or "" for p in event.content.parts))
    finally:
        # Shut the Toolbox subprocess down here (inside the same task that started it).
        await bigquery_tools.close()

# Run the async main loop
asyncio.run(main())

Why types.Part? ADK’s content parts support multimodal inputs (text, images, audio, etc.), so the typed Part makes messages explicit and extensible.

Example run and sample output

Run the script:

python churn_agent_bigquery.py

Sample output (formatted for clarity). The agent uses the BigQuery tools to read the customer’s features (without querying the churned column), reasons over signals, and returns a risk assessment and one recommended retention action:

**Risk Assessment for CUST-0003115**

Risk Level: HIGH 🔴

Customer Profile:
- Customer ID: CUST-0003115
- Region: NORTH
- Plan Tier: Pro
- Tenure: 20 months

Risk Signals:
1. Critical Disengagement
   - usage_last_30d: 0.11 (extremely low)
   - usage_delta_30d: -26.86% (sharp decline)
   => Shows the customer has almost stopped using the platform.

2. Support Friction
   - support_tickets_90d: 5
   => Multiple unresolved tickets indicate friction.

3. Moderate Tenure Risk
   - Tenure: 20 months (past onboarding but may not be deeply embedded)

Recommended Retention Action:
Urgent outreach with problem-solving focus: contact the customer to investigate the usage drop, review and resolve support tickets, offer hands-on help, or consider a plan adjustment to retain them.

The agent cites the signals (usage metrics and support tickets) and suggests a single concrete retention action, following the instructions.

Closing notes and best practices

Always close the MCP Toolbox subprocess from the same task that started it to avoid noisy cancel-scope tracebacks at interpreter shutdown.
Minimize the tool_filter to only the tools the agent actually needs — principle of least privilege for safety.
Use get_table_info when you want the agent to fetch schema information programmatically instead of embedding the schema in the instruction.
Provide clear instructions and a schema in the agent prompt so the model can reason without accidentally leaking labels like churned.

Links and References

Google Cloud BigQuery Documentation
Authenticate with Application Default Credentials (gcloud)
KodeKey / OpenAI-compatible endpoints (use your provider’s docs for model names and API details)
ADK and MCP tools are part of the Google ADK developer tooling (see repo/docs for the ADK you are using)

This completes the churn-agent-with-BigQuery lesson. The same approach — using MCP Toolbox to expose a small, audited set of database tools — can be applied to other backends such as Databricks or Snowflake.

Building Agent With BigQuery

Building Agent With Databricks

Building Agent With Snowflake

Demo Building Google ADK Agent With BigQuery MCP

Imports and environment bootstrap

.env example

Model initialization

Configure the MCP Toolbox bridge to BigQuery

Agent definition and instructions

Preview of the BigQuery table (for context)

Main runner and session loop

Example run and sample output

Closing notes and best practices

Links and References

Watch Video

​Imports and environment bootstrap

​.env example

​Model initialization

​Configure the MCP Toolbox bridge to BigQuery

​Agent definition and instructions

​Preview of the BigQuery table (for context)

​Main runner and session loop

​Example run and sample output

​Closing notes and best practices

​Links and References

Watch Video

Imports and environment bootstrap

.env example

Model initialization

Configure the MCP Toolbox bridge to BigQuery

Agent definition and instructions

Preview of the BigQuery table (for context)

Main runner and session loop

Example run and sample output

Closing notes and best practices

Links and References