Natural Language to BigQuery SQL Agent

Natural Language to BigQuery SQL Agent

This tutorial builds an ADK agent that translates plain English questions into BigQuery SQL, executes the query through an MCP server, and returns the result. The agent uses Gemini's code generation capability to write the SQL and the MCP tool to run it.

MCP Server with Dynamic SQL Execution

# bq_sql_server.py
from mcp.server.fastmcp import FastMCP
from google.cloud import bigquery
import os

mcp = FastMCP("bq-sql-runner")
client = bigquery.Client(project=os.environ["GCP_PROJECT_ID"])

SCHEMA = """
Table: shopmax.orders
Columns: order_id (INT64), customer_id (INT64), city (STRING),
         product_category (STRING), amount_rs (FLOAT64),
         order_date (DATE), status (STRING)

Table: shopmax.customers
Columns: customer_id (INT64), name (STRING), email (STRING),
         city (STRING), joined_date (DATE)
"""

@mcp.tool()
def get_schema() -> str:
    """Return the database schema so the agent can write correct SQL."""
    return SCHEMA

@mcp.tool()
def run_sql(query: str) -> dict:
    """Execute a read-only BigQuery SQL query and return the results."""
    if any(kw in query.upper() for kw in ["INSERT", "UPDATE", "DELETE", "DROP", "CREATE"]):
        return {"error": "Only SELECT queries are allowed"}
    try:
        rows = client.query(query).result()
        return {"rows": [dict(row) for row in rows], "count": rows.total_rows}
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    mcp.run(transport="stdio")

ADK Agent with Two-Step Reasoning

# nl_to_bq_agent.py
import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

async def main():
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="python",
            args=["bq_sql_server.py"],
        )
    )

    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="NL to SQL Agent",
        instruction="""
You convert natural language questions into BigQuery SQL and execute them.

Workflow:
1. Call get_schema to understand the available tables and columns.
2. Write a correct SELECT query based on the user question.
3. Call run_sql with the query.
4. Present the results in a readable format.

Never guess column names - always check the schema first.
""",
        tools=tools,
    )

    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name="nl-bq", user_id="u1", session_id="s1"
    )
    runner = Runner(agent=agent, app_name="nl-bq", session_service=session_service)

    questions = [
        "How many orders were placed in Mumbai last month?",
        "Who are the top 3 customers by total spending?",
        "What is the average order value per city?",
    ]

    for q in questions:
        print(f"\nQuestion: {q}")
        content = types.Content(role="user", parts=[types.Part(text=q)])
        async for event in runner.run_async(user_id="u1", session_id="s1", new_message=content):
            if event.is_final_response():
                print(event.content.parts[0].text)

    await exit_stack.aclose()

asyncio.run(main())

How the Agent Reasons

1. Receives the natural language question
2. Calls get_schema to learn table structures
3. Generates a parameterized SELECT query
4. Calls run_sql with the generated SQL
5. Formats and returns the result to the user

Safety Guardrails in the MCP Server

• Keyword blocklist rejects INSERT, UPDATE, DELETE, DROP, CREATE
• BigQuery runs in a service account with read-only IAM permissions
• Results are capped by BigQuery default row limits