Build an AI Data Analysis Agent with Gemini API — Combining Code Execution, Function Calling, and Structured Output

The Problem This Agent Solves

Data analysis typically involves a manual, multi-step process: data cleaning, exploratory analysis, visualization, and report generation. By combining three powerful Gemini API features — Code Execution, Function Calling, and Structured Output — you can build an AI agent that automates this entire pipeline with simple natural language instructions.

The agent below is built in Python from scratch. Upload a CSV or Excel file, ask a question in plain English, and it fetches the data, runs the statistics, draws the charts, and returns structured insights — without a single hand-written Pandas query.

What you'll learn:

How to use Gemini API's Code Execution to run Python in a sandboxed environment
Designing Function Calling tools to connect your agent to external data sources
Extracting type-safe JSON results with Structured Output
Integrating all three features into a complete data analysis pipeline

Prerequisites: Familiarity with Gemini API basics and intermediate Python development experience.

Environment Setup

Requirements

Python 3.10+
Google AI Python SDK (google-genai v1.55.0+)
A Gemini API key (available from Google AI Studio)

Installation

pip install google-genai pandas openpyxl matplotlib

Set your API key as an environment variable:

export GOOGLE_API_KEY="your-api-key-here"

Initialize the SDK in Python:

from google import genai
 
client = genai.Client()
 
# Code Execution-compatible model
MODEL_ID = "gemini-3-flash"

Model selection tip: Code Execution works with both gemini-3-flash and gemini-3-pro. Use Flash for cost efficiency and speed; choose Pro when your analysis requires complex multi-step reasoning.

Concepts and Design Philosophy

Role of Each Feature

The agent uses three Gemini API features, each serving a distinct purpose:

Feature	Role	Example
Code Execution	Data processing, statistics, visualization	Pandas aggregations, Matplotlib charts
Function Calling	External resource access	File loading, DB queries, API calls
Structured Output	Result formatting	JSON-schema-compliant insight reports

Architecture Overview

User (natural language query)
    ↓
Gemini API Agent
    ├── Function Calling → Fetch data from files/databases
    ├── Code Execution  → Analyze with Pandas/Matplotlib
    └── Structured Output → Format results as typed JSON
    ↓
Structured report + chart images

With this design, a user can simply say "Analyze last month's sales data and identify the top 5 categories by growth rate," and the agent will autonomously handle every step from data ingestion to visualization.

Step-by-Step Implementation

Step 1: Running Data Analysis with Code Execution

Let's start with the fundamentals. Code Execution lets Gemini generate and run Python code in a sandboxed environment, iterating until it reaches a solution.

from google import genai
from google.genai import types
 
client = genai.Client()
 
# Sample CSV data (in practice, you'd upload a file)
csv_data = """date,product,revenue,quantity
2026-01-01,Widget A,15000,120
2026-01-02,Widget A,18000,145
2026-01-03,Widget B,22000,90
2026-01-04,Widget A,16500,130
2026-01-05,Widget B,25000,105
2026-01-06,Widget C,8000,200
2026-01-07,Widget A,19000,155
2026-01-08,Widget B,21000,88
2026-01-09,Widget C,9500,220
2026-01-10,Widget A,17500,140"""
 
response = client.models.generate_content(
    model="gemini-3-flash",
    contents=f"""Analyze the following CSV data:
1. Calculate total revenue and average unit price per product
2. Create a daily revenue trend chart using Matplotlib
3. Identify the top-performing product and explain why
 
Data:
{csv_data}""",
    config=types.GenerateContentConfig(
        tools=[types.Tool(code_execution=types.ToolCodeExecution)]
    )
)
 
# Response contains text, code, execution results, and images
for part in response.candidates[0].content.parts:
    if part.text:
        print("📝 Text:", part.text)
    elif part.executable_code:
        print("💻 Code:", part.executable_code.code)
    elif part.code_execution_result:
        print("📊 Result:", part.code_execution_result.output)
    elif part.inline_data:
        # Matplotlib chart image (base64-encoded)
        print("🖼️ Chart image received")

Expected output:

💻 Code: import pandas as pd
import matplotlib.pyplot as plt
...
📊 Result:
Revenue by product:
  Widget A: $86,000 (avg. unit price: $124)
  Widget B: $68,000 (avg. unit price: $240)
  Widget C: $17,500 (avg. unit price: $42)
🖼️ Chart image received
📝 Text: Widget A leads in total revenue...

Key point: The Code Execution sandbox comes pre-loaded with Pandas, NumPy, Matplotlib, scikit-learn, and 50+ other libraries. You cannot install additional packages. Only Matplotlib is supported for chart rendering.

Step 2: Connecting Data Sources with Function Calling

Next, add Function Calling so the agent can access external data sources — files, databases, or APIs.

from google.genai import types
import pandas as pd
import json
 
# --- Tool implementations ---
 
def load_csv_file(file_path: str) -> str:
    """Load a CSV file and return a preview with statistics"""
    df = pd.read_csv(file_path)
    summary = {
        "rows": len(df),
        "columns": list(df.columns),
        "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
        "preview": df.head(20).to_csv(index=False),
        "describe": df.describe().to_json()
    }
    return json.dumps(summary, ensure_ascii=False)
 
def query_database(sql: str) -> str:
    """Execute a SQL query and return results as CSV"""
    import sqlite3
    conn = sqlite3.connect("sales.db")
    df = pd.read_sql_query(sql, conn)
    conn.close()
    return df.to_csv(index=False)
 
def save_report(filename: str, content: str) -> str:
    """Save an analysis report as a Markdown file"""
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"Report saved to {filename}"
 
# --- Function declarations for Gemini ---
 
tools = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="load_csv_file",
            description="Load a CSV file and return statistics and a preview",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "file_path": types.Schema(
                        type="STRING",
                        description="Path to the CSV file"
                    )
                },
                required=["file_path"]
            )
        ),
        types.FunctionDeclaration(
            name="query_database",
            description="Execute a SQL query against an SQLite database",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "sql": types.Schema(
                        type="STRING",
                        description="The SQL query to execute"
                    )
                },
                required=["sql"]
            )
        ),
        types.FunctionDeclaration(
            name="save_report",
            description="Save an analysis report to a file",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "filename": types.Schema(
                        type="STRING",
                        description="Output filename"
                    ),
                    "content": types.Schema(
                        type="STRING",
                        description="Report content in Markdown format"
                    )
                },
                required=["filename", "content"]
            )
        )
    ]
)

Step 3: Getting Type-Safe Results with Structured Output

Add Structured Output to ensure analysis results conform to a well-defined JSON schema.

from google.genai import types
 
# Define the analysis report schema
analysis_schema = types.Schema(
    type="OBJECT",
    properties={
        "summary": types.Schema(
            type="STRING",
            description="Executive summary of the analysis (3 sentences max)"
        ),
        "key_metrics": types.Schema(
            type="ARRAY",
            items=types.Schema(
                type="OBJECT",
                properties={
                    "name": types.Schema(type="STRING"),
                    "value": types.Schema(type="STRING"),
                    "trend": types.Schema(
                        type="STRING",
                        enum=["up", "down", "stable"]
                    )
                }
            ),
            description="List of key performance metrics"
        ),
        "insights": types.Schema(
            type="ARRAY",
            items=types.Schema(
                type="OBJECT",
                properties={
                    "title": types.Schema(type="STRING"),
                    "description": types.Schema(type="STRING"),
                    "confidence": types.Schema(
                        type="STRING",
                        enum=["high", "medium", "low"]
                    ),
                    "action_item": types.Schema(type="STRING")
                }
            ),
            description="Discovered insights with recommended actions"
        ),
        "anomalies": types.Schema(
            type="ARRAY",
            items=types.Schema(type="STRING"),
            description="Detected anomalies or outliers"
        )
    },
    required=["summary", "key_metrics", "insights"]
)
 
# Get structured analysis results
response = client.models.generate_content(
    model="gemini-3-flash",
    contents="Structure the sales data analysis results from above",
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=analysis_schema
    )
)
 
import json
result = json.loads(response.text)
print(json.dumps(result, indent=2))

Expected output:

{
  "summary": "Widget A leads in total revenue at 49% market share. Widget B shows the highest unit price strategy. Widget C captures the high-volume, low-price segment.",
  "key_metrics": [
    {"name": "Total Revenue", "value": "$171,500", "trend": "up"},
    {"name": "Daily Average", "value": "$17,150", "trend": "stable"},
    {"name": "Peak Day", "value": "Jan 5 ($25,000)", "trend": "up"}
  ],
  "insights": [
    {
      "title": "Widget B's premium pricing strategy is effective",
      "description": "Despite the lowest unit count, Widget B commands the highest unit price, suggesting strong profit margins",
      "confidence": "high",
      "action_item": "Conduct a detailed margin analysis on Widget B to optimize pricing strategy"
    }
  ],
  "anomalies": ["Widget C revenue spike on Jan 9 (+18.75% day-over-day)"]
}

Step 4: The Complete Agent Class

Now let's bring everything together into a production-ready agent.

from google import genai
from google.genai import types
import pandas as pd
import json
import base64
from pathlib import Path
 
class DataAnalysisAgent:
    """AI Data Analysis Agent powered by Gemini API"""
 
    def __init__(self, model: str = "gemini-3-flash"):
        self.client = genai.Client()
        self.model = model
        self._setup_tools()
 
    def _setup_tools(self):
        """Configure Function Calling + Code Execution tools"""
        self.function_tool = types.Tool(
            function_declarations=[
                types.FunctionDeclaration(
                    name="load_csv_file",
                    description="Load a CSV file and return stats",
                    parameters=types.Schema(
                        type="OBJECT",
                        properties={
                            "file_path": types.Schema(
                                type="STRING",
                                description="Path to the CSV file"
                            )
                        },
                        required=["file_path"]
                    )
                )
            ]
        )
        self.code_exec_tool = types.Tool(
            code_execution=types.ToolCodeExecution
        )
 
    def _handle_function_call(self, fn_call) -> str:
        """Dispatch function calls to implementations"""
        name = fn_call.name
        args = dict(fn_call.args)
 
        if name == "load_csv_file":
            df = pd.read_csv(args["file_path"])
            return json.dumps({
                "rows": len(df),
                "columns": list(df.columns),
                "preview": df.head(20).to_csv(index=False),
                "stats": df.describe().to_json()
            })
        return '{"error": "Unknown function"}'
 
    def analyze(self, file_path: str, question: str) -> dict:
        """
        Analyze a file and return structured results.
 
        Args:
            file_path: Path to a CSV/Excel file
            question: Analysis question or instruction
 
        Returns:
            dict: Structured analysis with metrics, insights, and charts
        """
        config = types.GenerateContentConfig(
            tools=[self.function_tool, self.code_exec_tool],
            system_instruction="""You are a data analysis expert.
Follow these steps:
1. Use load_csv_file to read the data
2. Use Code Execution to analyze and visualize the data
3. Provide detailed insights about your findings"""
        )
 
        prompt = f"""Analyze the following file:
File: {file_path}
Question: {question}
 
Steps:
1. Load the data using load_csv_file
2. Analyze and visualize with Python code
3. Summarize your findings"""
 
        contents = [types.Content(
            role="user",
            parts=[types.Part(text=prompt)]
        )]
 
        graphs = []
        max_turns = 5
 
        for _ in range(max_turns):
            response = self.client.models.generate_content(
                model=self.model,
                contents=contents,
                config=config
            )
 
            fn_calls = [
                p for p in response.candidates[0].content.parts
                if p.function_call
            ]
 
            if fn_calls:
                contents.append(response.candidates[0].content)
                fn_responses = []
                for fc in fn_calls:
                    result = self._handle_function_call(fc.function_call)
                    fn_responses.append(
                        types.Part(function_response=types.FunctionResponse(
                            name=fc.function_call.name,
                            response=json.loads(result)
                        ))
                    )
                contents.append(types.Content(
                    role="user", parts=fn_responses
                ))
                continue
 
            for part in response.candidates[0].content.parts:
                if part.inline_data:
                    graphs.append(part.inline_data)
            break
 
        # Phase 2: Structure the results
        analysis_text = response.text if response.text else ""
 
        structured_response = self.client.models.generate_content(
            model=self.model,
            contents=f"Structure these analysis results:\n\n{analysis_text}",
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=types.Schema(
                    type="OBJECT",
                    properties={
                        "summary": types.Schema(type="STRING"),
                        "key_metrics": types.Schema(
                            type="ARRAY",
                            items=types.Schema(
                                type="OBJECT",
                                properties={
                                    "name": types.Schema(type="STRING"),
                                    "value": types.Schema(type="STRING"),
                                    "trend": types.Schema(type="STRING")
                                }
                            )
                        ),
                        "insights": types.Schema(
                            type="ARRAY",
                            items=types.Schema(
                                type="OBJECT",
                                properties={
                                    "title": types.Schema(type="STRING"),
                                    "description": types.Schema(type="STRING"),
                                    "action_item": types.Schema(type="STRING")
                                }
                            )
                        )
                    },
                    required=["summary", "key_metrics", "insights"]
                )
            )
        )
 
        result = json.loads(structured_response.text)
        result["graphs"] = [
            {
                "mime_type": g.mime_type,
                "data": base64.b64encode(g.data).decode()
            }
            for g in graphs
        ]
        return result
 
    def save_graphs(self, result: dict, output_dir: str = "./output"):
        """Save chart images from analysis results to disk"""
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        for i, graph in enumerate(result.get("graphs", [])):
            ext = graph["mime_type"].split("/")[-1]
            path = Path(output_dir) / f"chart_{i}.{ext}"
            with open(path, "wb") as f:
                f.write(base64.b64decode(graph["data"]))
            print(f"💾 Chart saved: {path}")

Usage example:

agent = DataAnalysisAgent(model="gemini-3-flash")
 
result = agent.analyze(
    file_path="sales_2026_q1.csv",
    question="Analyze quarterly sales trends and identify the fastest-growing product categories"
)
 
print("📊 Summary:", result["summary"])
for metric in result["key_metrics"]:
    print(f"  {metric['name']}: {metric['value']} ({metric['trend']})")
 
for insight in result["insights"]:
    print(f"\n💡 {insight['title']}")
    print(f"   {insight['description']}")
    print(f"   → Action: {insight['action_item']}")
 
agent.save_graphs(result, output_dir="./reports/charts")

Advanced Patterns

Pattern 1: Multi-File Comparative Analysis

Load and compare multiple datasets in a single analysis session.

result = agent.analyze(
    file_path="sales_q1.csv",
    question="""Compare Q1 and Q2 sales data.
    Additional data: sales_q2.csv
    Compare:
    - Revenue growth rates
    - Category market share shifts
    - Seasonal patterns"""
)

Pattern 2: Automated Scheduled Reports

Combine the agent with a scheduler for recurring reports.

import schedule
import time
from datetime import datetime
 
def daily_report():
    agent = DataAnalysisAgent()
    today = datetime.now().strftime("%Y-%m-%d")
 
    result = agent.analyze(
        file_path=f"/data/daily/{today}.csv",
        question="Generate a daily sales summary with day-over-day comparisons"
    )
 
    report = f"# Daily Report ({today})\n\n"
    report += f"## Summary\n{result['summary']}\n\n"
    report += "## Key Metrics\n"
    for m in result["key_metrics"]:
        report += f"- **{m['name']}**: {m['value']} ({m['trend']})\n"
 
    with open(f"reports/{today}.md", "w") as f:
        f.write(report)
 
schedule.every().day.at("09:00").do(daily_report)

Pattern 3: Anomaly Detection Pipeline

Leverage scikit-learn within Code Execution for automated anomaly detection.

result = agent.analyze(
    file_path="sensor_data.csv",
    question="""Detect anomalies in this sensor data using:
    - Isolation Forest for outlier detection
    - Moving average deviation analysis
    - List the top 10 highest anomaly scores
    Include a chart with anomalies highlighted in red."""
)

Troubleshooting

Common Issues and Solutions

1. Code Execution timeout (30-second limit)

Large datasets can exceed the execution time limit.

# ❌ Processing millions of rows directly
"Analyze all 1 million rows in this CSV"
 
# ✅ Sample or aggregate first
"Randomly sample 10,000 rows and perform statistical analysis"

2. Conflicts between Function Calling and Code Execution

When both tools are enabled, the model may be uncertain which to use.

# Use System Instructions to clarify roles
system_instruction = """
Always use load_csv_file for data loading.
Always use Code Execution for analysis, calculations, and charts.
Never confuse these two roles.
"""

3. Structured Output schema violations

Overly complex schemas may cause incomplete results.

# ❌ Deeply nested schemas with many required fields
# ✅ Keep schemas flat, add complexity incrementally

Cost and Performance Considerations

Token Cost Estimates

Operation	Input Tokens	Output Tokens	Flash Cost (est.)
CSV load (1,000 rows)	~5,000	~500	~$0.001
Statistical analysis + code execution	~2,000	~3,000	~$0.003
Chart generation (Matplotlib)	~1,000	~2,000	~$0.002
Structured Output	~3,000	~1,000	~$0.002
Total (single analysis)	~11,000	~6,500	~$0.008

Performance Optimization Tips

Two-phase approach: Start with a data preview to understand the structure, then run targeted detailed analysis only where needed.
Context Caching: When asking multiple questions about the same dataset, use Context Caching to avoid re-sending input tokens.
Flash vs. Pro: Use Flash for routine analysis (lower cost, faster response). Reserve Pro for complex reasoning tasks.

# Efficient multi-question analysis with Context Caching
from google.genai import types
 
cache = client.caches.create(
    model="gemini-3-flash",
    contents=[types.Content(
        role="user",
        parts=[types.Part(text=f"Dataset for analysis:\n{large_csv_data}")]
    )],
    config=types.CreateCachedContentConfig(
        display_name="sales-data-q1",
        ttl="3600s"
    )
)
 
# Ask multiple questions using the cache (saves input tokens)
for question in ["What are the sales trends?", "Any anomalies?", "Build a forecast model"]:
    response = client.models.generate_content(
        model="gemini-3-flash",
        contents=question,
        config=types.GenerateContentConfig(
            cached_content=cache.name,
            tools=[types.Tool(code_execution=types.ToolCodeExecution)]
        )
    )

Conclusion

The agent we just built combines Gemini API's Code Execution, Function Calling, and Structured Output to automatically analyze CSV/Excel data from natural language instructions.

Key takeaways:

Code Execution runs Pandas and Matplotlib in a sandboxed environment, ideal for data processing and visualization
Function Calling abstracts external data access, extending the agent's capabilities to files, databases, and APIs
Structured Output ensures analysis results are returned as type-safe JSON, ready for downstream processing
Combining all three features enables sophisticated automated analysis pipelines that would be difficult to build with any single feature alone

To take this further, explore [Gemini API Streaming]((/articles/gemini-api/streaming-and-chat) for real-time analysis progress updates, the [Interactions API]((/articles/gemini-api/gemini-interactions-api-guide) for long-running research tasks, and [Function Calling production patterns]((/articles/gemini-api/gemini-function-calling-production) to improve agent reliability.