What is a Data Analysis Agent?
A data analysis agent is an AI-powered assistant that can autonomously explore, query, analyze, and visualize data based on natural language instructions. Built with the OpenAI Agents SDK, it combines the reasoning capabilities of large language models with concrete tool execution — such as running SQL queries, executing Python code, generating charts, and performing statistical computations. Unlike a simple chatbot that merely discusses data, an analysis agent actively does the work: it connects to databases, inspects schemas, formulates hypotheses, runs analyses, and iterates based on results until it reaches meaningful conclusions.
The OpenAI Agents SDK provides the scaffolding to build these agents with minimal boilerplate. It handles conversation state, tool calling loops, handoffs between specialized sub-agents, and built-in tracing for debugging. This lets you focus on defining the tools your agent needs and the instructions that guide its analytical reasoning.
Why Data Analysis Agents Matter
Traditional data analysis workflows are bottlenecked by human bandwidth. An analyst must manually write queries, export data, switch contexts between SQL editors and Python notebooks, and piece together insights across multiple tools. A data analysis agent collapses this entire loop into a single conversational interface where the agent handles the mechanical execution while the human stays in the driver's seat for strategic decisions.
Key benefits include:
- Speed: The agent can explore a dozen hypotheses in the time it takes a human to write one query
- Accessibility: Non-technical stakeholders can ask questions in plain English and get real answers backed by actual data
- Consistency: The agent follows a rigorous analytical methodology every time, reducing ad-hoc errors
- Auditability: Every query, computation, and chart generated is traced and logged, creating a reproducible analysis record
- Scalability: A single agent can serve multiple users concurrently, each exploring different facets of the same dataset
Setting Up Your Environment
Before building your agent, install the OpenAI Agents SDK along with supporting data libraries. Create a new project directory and set up a virtual environment:
mkdir data-analysis-agent
cd data-analysis-agent
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activateInstall the required packages:
pip install openai-agents pandas matplotlib seaborn sqlalchemy python-dotenvSet your OpenAI API key as an environment variable. Create a .env file:
OPENAI_API_KEY=your-api-key-hereLoad it in your application entry point:
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY not set in environment")Building Your First Data Analysis Agent
We'll start with a single agent that can perform basic data analysis using Python and pandas. The agent receives a dataset (as a CSV file path or a database connection string), understands the user's question, and executes the appropriate analysis code.
Defining Tool Functions
Tools are the executable capabilities your agent can invoke. For data analysis, we need tools that can load data, inspect its structure, run pandas operations, and generate visualizations. Each tool is a Python function decorated with @function_tool from the SDK:
# tools/data_tools.py
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path
from typing import Optional
from openai_agents import function_tool
# Global cache for loaded dataframes to avoid repeated I/O
_dataframe_cache: dict[str, pd.DataFrame] = {}
@function_tool
def load_csv(file_path: str) -> str:
"""Load a CSV file into memory and return a summary of its structure.
Args:
file_path: Path to the CSV file to load.
Returns:
A string describing the dataframe shape, columns, and dtypes.
"""
path = Path(file_path)
if not path.exists():
return f"Error: File not found at {file_path}"
df = pd.read_csv(path)
cache_key = path.stem
_dataframe_cache[cache_key] = df
summary = f"Loaded '{cache_key}' successfully.\n"
summary += f"Shape: {df.shape[0]} rows × {df.shape[1]} columns\n"
summary += f"Columns: {list(df.columns)}\n"
summary += f"Data types:\n{df.dtypes.to_string()}\n"
summary += f"First 5 rows:\n{df.head().to_string()}\n"
summary += f"Missing values per column:\n{df.isna().sum().to_string()}"
return summary
@function_tool
def get_dataframe_info(dataset_name: str) -> str:
"""Get detailed information about an already-loaded dataset.
Args:
dataset_name: The name of the dataset (stem of the loaded file).
Returns:
Statistical summary and info about the dataframe.
"""
if dataset_name not in _dataframe_cache:
return f"Error: Dataset '{dataset_name}' not found in memory. Available: {list(_dataframe_cache.keys())}"
df = _dataframe_cache[dataset_name]
buffer = []
buffer.append(f"Dataset: {dataset_name}")
buffer.append(f"Shape: {df.shape}")
buffer.append(f"\nStatistical Summary:\n{df.describe(include='all').to_string()}")
buffer.append(f"\nMissing Values:\n{df.isna().sum().to_string()}")
buffer.append(f"\nDuplicate rows: {df.duplicated().sum()}")
return "\n".join(buffer)
@function_tool
def run_pandas_query(dataset_name: str, code: str) -> str:
"""Execute a pandas query on the specified dataset and return the result.
Args:
dataset_name: The name of the dataset to query.
code: Python code that uses the variable 'df' to perform operations.
Must be a single expression or assignment that produces a result.
Returns:
String representation of the query result.
"""
if dataset_name not in _dataframe_cache:
return f"Error: Dataset '{dataset_name}' not found. Available: {list(_dataframe_cache.keys())}"
df = _dataframe_cache[dataset_name].copy()
# Create a restricted execution environment
local_vars = {'df': df, 'pd': pd, 'result': None}
try:
# Execute the code and capture the result
exec(f"result = {code}", {"__builtins__": {}}, local_vars)
result = local_vars.get('result')
if isinstance(result, pd.DataFrame):
return f"DataFrame ({result.shape[0]}×{result.shape[1]}):\n{result.head(20).to_string()}"
elif isinstance(result, pd.Series):
return f"Series ({len(result)}):\n{result.head(20).to_string()}"
elif isinstance(result, (int, float, str, bool)):
return str(result)
else:
return str(result)
except Exception as e:
return f"Error executing query: {type(e).__name__}: {str(e)}"
@function_tool
def create_visualization(dataset_name: str, plot_type: str, x_column: str,
y_column: Optional[str] = None, title: Optional[str] = None,
output_path: str = "plot.png") -> str:
"""Generate a visualization from the dataset and save it to a file.
Args:
dataset_name: Name of the loaded dataset.
plot_type: Type of plot ('histogram', 'scatter', 'bar', 'box', 'line', 'heatmap').
x_column: Column name for the x-axis.
y_column: Column name for the y-axis (required for scatter, bar, line).
title: Optional chart title.
output_path: File path to save the chart image.
Returns:
Confirmation message with the saved file path.
"""
if dataset_name not in _dataframe_cache:
return f"Error: Dataset '{dataset_name}' not found."
df = _dataframe_cache[dataset_name]
plt.figure(figsize=(10, 6))
sns.set_style("whitegrid")
try:
if plot_type == 'histogram':
sns.histplot(data=df, x=x_column, kde=True)
elif plot_type == 'scatter':
if y_column is None:
return "Error: y_column required for scatter plot"
sns.scatterplot(data=df, x=x_column, y=y_column)
elif plot_type == 'bar':
if y_column is None:
df[x_column].value_counts().plot(kind='bar')
else:
sns.barplot(data=df, x=x_column, y=y_column)
elif plot_type == 'box':
sns.boxplot(data=df, x=x_column, y=y_column)
elif plot_type == 'line':
sns.lineplot(data=df, x=x_column, y=y_column)
elif plot_type == 'heatmap':
corr = df.select_dtypes(include='number').corr()
sns.heatmap(corr, annot=True, cmap='coolwarm', fmt='.2f')
plt.title(title or "Correlation Heatmap")
plt.tight_layout()
plt.savefig(output_path, dpi=150)
plt.close()
return f"Heatmap saved to {output_path}"
else:
return f"Error: Unsupported plot type '{plot_type}'. Choose from: histogram, scatter, bar, box, line, heatmap"
if title:
plt.title(title)
plt.tight_layout()
plt.savefig(output_path, dpi=150)
plt.close()
return f"Chart saved to {output_path}"
except Exception as e:
plt.close()
return f"Error creating visualization: {type(e).__name__}: {str(e)}"Creating the Agent
Now we define the agent itself. The agent needs clear instructions that establish its role as a data analyst, describe the available tools, and specify the analytical methodology it should follow. Here's the main agent definition:
# agent.py
from openai_agents import Agent, Runner
from tools.data_tools import (
load_csv, get_dataframe_info, run_pandas_query, create_visualization
)
DATA_ANALYST_INSTRUCTIONS = """You are a senior data analyst agent with expertise in statistics,
exploratory data analysis, and business intelligence.
Your workflow when receiving a data analysis request:
1. UNDERSTAND: Clarify the user's question and what metrics or insights they need.
2. LOAD: If the data source is provided as a file path, use load_csv to bring it into memory.
3. INSPECT: Use get_dataframe_info to understand distributions, missing values, and data types.
4. EXPLORE: Formulate hypotheses and use run_pandas_query to test them one at a time.
5. SUMMARIZE: After sufficient exploration, synthesize findings into a clear narrative.
6. VISUALIZE: Use create_visualization to generate charts that support your conclusions.
Critical rules:
- Always inspect data BEFORE running queries. Know the columns and types.
- Run one query at a time. Do not chain multiple analyses in a single code block.
- If a query fails, read the error, adjust the code, and retry.
- When creating visualizations, choose plot types appropriate for the data.
- Present findings in plain language with concrete numbers and actionable insights.
- If you cannot answer a question due to data limitations, say so honestly.
- Never fabricate data or results. Only report what the data actually shows."""
data_analyst_agent = Agent(
name="DataAnalyst",
instructions=DATA_ANALYST_INSTRUCTIONS,
tools=[load_csv, get_dataframe_info, run_pandas_query, create_visualization],
model="gpt-4o",
)Running the Agent
To execute the agent, use the Runner class which manages the conversation loop — sending prompts, receiving tool calls, executing them, and feeding results back to the model. Here's a complete run script:
# run_analysis.py
import asyncio
from openai_agents import Runner
from agent import data_analyst_agent
async def main():
# Sample dataset: sales data
user_query = """
I have a CSV file at './data/sales_2024.csv' containing e-commerce sales data.
Please analyze it and tell me:
1. What are the top-selling product categories?
2. Is there a relationship between discount percentage and units sold?
3. Which regions generate the highest revenue?
Generate visualizations for your findings.
"""
print("Starting analysis...")
result = await Runner.run(
starting_agent=data_analyst_agent,
input=user_query,
)
print("\n=== ANALYSIS COMPLETE ===\n")
print(result.final_output)
# Print tool call trace for debugging
print("\n--- Tool Calls Executed ---")
for i, step in enumerate(result.steps, 1):
if step.tool_calls:
for call in step.tool_calls:
print(f"Step {i}: {call.tool_name}")
if __name__ == "__main__":
asyncio.run(main())Connecting to Databases with a SQL Tool
Real-world analysis often requires querying live databases. Let's add a SQL execution tool that connects via SQLAlchemy, allowing the agent to work directly with PostgreSQL, MySQL, or SQLite:
# tools/sql_tools.py
from sqlalchemy import create_engine, text, inspect
from openai_agents import function_tool
from typing import Optional
# Store active connections
_active_connections: dict[str, any] = {}
@function_tool
def connect_database(connection_string: str, alias: str = "default") -> str:
"""Connect to a SQL database and cache the connection.
Args:
connection_string: SQLAlchemy connection string
(e.g., 'postgresql://user:pass@localhost/dbname' or 'sqlite:///data.db').
alias: A nickname to reference this connection in subsequent calls.
Returns:
Status message and list of available tables.
"""
try:
engine = create_engine(connection_string, echo=False)
# Test connection
with engine.connect() as conn:
conn.execute(text("SELECT 1"))
_active_connections[alias] = engine
# Discover tables
inspector = inspect(engine)
tables = inspector.get_table_names()
table_list = "\n".join([f" - {t}" for t in tables])
return f"Connected to '{alias}'. Available tables:\n{table_list}"
except Exception as e:
return f"Connection error: {type(e).__name__}: {str(e)}"
@function_tool
def describe_table(connection_alias: str, table_name: str) -> str:
"""Get the schema and row count for a specific table.
Args:
connection_alias: The alias used when connecting.
table_name: Name of the table to describe.
Returns:
Column definitions and approximate row count.
"""
if connection_alias not in _active_connections:
return f"Error: No connection found for alias '{connection_alias}'. Connect first."
engine = _active_connections[connection_alias]
inspector = inspect(engine)
try:
columns = inspector.get_columns(table_name)
col_info = "\n".join([
f" {col['name']} ({col['type']}){' PRIMARY KEY' if col.get('primary_key') else ''}"
for col in columns
])
with engine.connect() as conn:
count = conn.execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar()
return f"Table: {table_name}\nRow count: {count}\nColumns:\n{col_info}"
except Exception as e:
return f"Error describing table: {type(e).__name__}: {str(e)}"
@function_tool
def run_sql_query(connection_alias: str, query: str, limit: int = 100) -> str:
"""Execute a read-only SQL query and return results.
Args:
connection_alias: The connection alias.
query: A SELECT query to execute. Write-only queries are blocked.
limit: Maximum rows to return (default 100).
Returns:
Query results as a formatted string.
"""
# Safety: only allow SELECT statements
query_stripped = query.strip().upper()
if not query_stripped.startswith("SELECT"):
return "Error: Only SELECT queries are allowed for safety."
if connection_alias not in _active_connections:
return f"Error: No connection for '{connection_alias}'."
engine = _active_connections[connection_alias]
try:
# Wrap in limit if not already present and not an aggregate
if "LIMIT" not in query_stripped and "LIMIT" not in query.lower():
query = f"{query.strip().rstrip(';')} LIMIT {limit}"
with engine.connect() as conn:
result = conn.execute(text(query))
rows = result.fetchall()
columns = result.keys()
if not rows:
return "Query returned 0 rows."
# Format as table
col_widths = [max(len(str(c)), max(len(str(r[i])) for r in rows)) for i, c in enumerate(columns)]
header = " | ".join(str(c).ljust(col_widths[i]) for i, c in enumerate(columns))
separator = "-+-".join("-" * w for w in col_widths)
data_rows = "\n".join(
" | ".join(str(val).ljust(col_widths[i]) for i, val in enumerate(row))
for row in rows[:limit]
)
return f"Rows returned: {len(rows)}\n\n{header}\n{separator}\n{data_rows}"
except Exception as e:
return f"SQL Error: {type(e).__name__}: {str(e)}"Multi-Agent Architecture for Complex Analysis
As analysis requirements grow more sophisticated, a single monolithic agent can become overwhelmed — its instructions grow bloated, and it struggles to context-switch between SQL querying, statistical modeling, and visualization. The OpenAI Agents SDK supports agent handoffs, allowing you to decompose the problem into specialized sub-agents that each excel at one domain:
Specialized Sub-Agents
# agents/specialized_agents.py
from openai_agents import Agent
from tools.sql_tools import connect_database, describe_table, run_sql_query
from tools.data_tools import load_csv, get_dataframe_info, run_pandas_query, create_visualization
# SQL Specialist Agent
sql_agent = Agent(
name="SQLSpecialist",
instructions="""You are an expert SQL query writer. Your job:
- Connect to databases using connect_database
- Inspect schemas with describe_table before writing queries
- Write efficient, correct SQL queries using run_sql_query
- Return formatted results with clear explanations
- If you need statistical analysis on results, hand off to the StatsSpecialist.
- Never execute destructive queries (INSERT, UPDATE, DELETE, DROP).""",
tools=[connect_database, describe_table, run_sql_query],
model="gpt-4o",
)
# Statistical Analysis Specialist
stats_agent = Agent(
name="StatsSpecialist",
instructions="""You are a statistical analysis expert. Your job:
- Load data with load_csv if needed
- Inspect data thoroughly with get_dataframe_info
- Perform statistical tests, regressions, and correlations using run_pandas_query
- Generate visualizations with create_visualization
- Explain statistical concepts in plain language
- Provide confidence intervals and p-values when relevant.""",
tools=[load_csv, get_dataframe_info, run_pandas_query, create_visualization],
model="gpt-4o",
)
# Orchestrator Agent — routes tasks to specialists
orchestrator_agent = Agent(
name="AnalysisOrchestrator",
instructions="""You are the lead data analysis orchestrator. Your role:
1. Understand the user's full analysis request.
2. Decompose it into subtasks.
3. Route SQL-heavy tasks to the SQLSpecialist agent.
4. Route statistical analysis and visualization tasks to the StatsSpecialist agent.
5. Synthesize results from both specialists into a coherent final report.
Handoff guidelines:
- If the user asks about database content, send to SQLSpecialist.
- If the user wants statistical insights, charts, or CSV analysis, send to StatsSpecialist.
- After each specialist responds, integrate their findings before making another handoff.
- When all subtasks are complete, deliver a comprehensive final summary.""",
handoffs=[sql_agent, stats_agent],
tools=[],
model="gpt-4o",
)Running the Multi-Agent System
# run_orchestrated_analysis.py
import asyncio
from openai_agents import Runner
from agents.specialized_agents import orchestrator_agent
async def main():
query = """
We have a PostgreSQL database 'ecommerce' with sales data and a CSV export
at './data/customer_churn.csv'. Please:
1. From the database, find the top 10 products by revenue and the monthly sales trend.
2. From the CSV, analyze customer churn patterns and identify the top 3 predictors of churn.
3. Create visualizations for both findings.
4. Provide a unified summary connecting database insights with churn analysis.
Database connection: postgresql://analyst:password@localhost:5432/ecommerce
"""
result = await Runner.run(
starting_agent=orchestrator_agent,
input=query,
)
print(result.final_output)
asyncio.run(main())The orchestrator intelligently routes each subtask to the appropriate specialist. The SQLSpecialist connects to the database, inspects tables, writes queries, and returns structured results. The StatsSpecialist loads the CSV, runs pandas analyses, and generates charts. The orchestrator then synthesizes everything into a unified report — all within a single conversation flow tracked by the SDK's built-in tracing.
Adding Guardrails and Safety
Data analysis agents operate on potentially sensitive data. The SDK supports guardrails — validation checks that run on inputs and outputs to enforce safety policies. Here's how to add input validation that prevents dangerous operations:
# guardrails.py
from openai_agents import input_guardrail, output_guardrail, GuardrailResult
@input_guardrail
def block_destructive_sql(input_text: str) -> GuardrailResult:
"""Block any input containing destructive SQL keywords."""
dangerous_keywords = [
"DROP", "DELETE", "UPDATE", "INSERT", "ALTER",
"TRUNCATE", "CREATE", "GRANT", "REVOKE"
]
upper_input = input_text.upper()
for keyword in dangerous_keywords:
if keyword in upper_input:
return GuardrailResult(
allowed=False,
reason=f"Input contains potentially destructive SQL keyword: {keyword}"
)
return GuardrailResult(allowed=True)
@output_guardrail
def mask_sensitive_data(output_text: str) -> GuardrailResult:
"""Check output for potential PII patterns and mask them."""
import re
# Simple email pattern check
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
if re.search(email_pattern, output_text):
return GuardrailResult(
allowed=False,
reason="Output may contain email addresses. Mask before returning."
)
return GuardrailResult(allowed=True)
# Attach guardrails to your agent
from agent import data_analyst_agent
data_analyst_agent.input_guardrails = [block_destructive_sql]
data_analyst_agent.output_guardrails = [mask_sensitive_data]Best Practices for Data Analysis Agents
1. Design Focused Tools
Each tool should do one thing well. Resist the temptation to create an "analyze_everything" mega-tool. Smaller, composable tools give the agent more flexibility to chain operations creatively. A good rule of thumb: if a tool function exceeds 50 lines, consider splitting it.
2. Write Detailed Tool Descriptions
The agent relies on your function docstrings to understand what each tool does and when to use it. Include clear parameter descriptions, return value expectations, and usage examples in the docstring. The model reads these descriptions to decide which tool to invoke — ambiguous descriptions lead to incorrect tool selection.
3. Implement Progressive Data Loading
For large datasets, don't load everything into memory at once. Design tools that support sampling, pagination, and incremental loading. For SQL tools, always enforce a default LIMIT clause. For file-based tools, consider adding parameters for row ranges or chunked reading.
4. Build Error Recovery Loops
Analysis inevitably hits errors — malformed queries, missing columns, type mismatches. Your agent instructions should explicitly tell the model to read error messages carefully and retry with corrected code. The SDK's loop naturally supports this: a failed tool call returns an error string, the model sees it in context, and can attempt a fix in the next turn.
5. Separate Exploration from Reporting
Instruct your agent to explore freely but withhold final conclusions until it has gathered sufficient evidence. A common failure mode is the agent making a declarative statement after running only one query. Encourage it to run multiple queries, cross-reference results, and only then synthesize a conclusion.
6. Use Handoffs for Domain Boundaries
When your analysis spans fundamentally different data sources or methodologies (SQL vs. statistical modeling vs. machine learning), use separate agents connected via handoffs rather than cramming all tools into one agent. This keeps each agent's context clean and focused, improving reliability.
7. Trace Everything for Debugging
The OpenAI Agents SDK includes built-in tracing. Enable it during development to see every tool call, handoff, and model response. This is invaluable for understanding why an agent made a particular decision or where an analysis went wrong.
# Enable tracing in your runner
from openai_agents import Runner, Trace
async def run_with_tracing():
with Trace(project_name="data_analysis") as trace:
result = await Runner.run(
starting_agent=data_analyst_agent,
input="Analyze the sales trends in ./data/sales_2024.csv",
)
print(f"Trace ID: {trace.id}") # Use this to inspect in the dashboard
return result8. Validate and Sanitize Inputs
Always use guardrails to prevent SQL injection, block destructive operations, and mask sensitive outputs. Data analysis agents operate on real data — a single unguarded query could expose PII or corrupt a database. Layer input validation (blocking dangerous keywords), output validation (detecting PII patterns), and tool-level restrictions (SELECT-only queries) for defense in depth.
9. Test with Diverse Datasets
Before deploying, test your agent against a variety of datasets: clean CSVs, messy CSVs with missing values, large databases, empty tables, and malformed files. The agent should handle each case gracefully, reporting errors clearly rather than crashing silently.
10. Iterate on Instructions
Agent instructions are the most high-leverage thing you can improve. Start with a basic instruction set, run 10–15 diverse analysis scenarios, note where the agent stumbles, and refine the instructions to address those gaps. Repeat until the agent consistently produces reliable, thorough analyses.
Conclusion
Building a data analysis agent with the OpenAI Agents SDK transforms how you interact with data — from manually writing queries and scripts to having a capable AI assistant that autonomously explores, analyzes, and visualizes information based on natural language questions. The SDK provides the essential infrastructure: tool definitions, conversation management, handoffs between specialized agents, and built-in tracing for observability. By starting with focused tools, clear instructions, and a progressive exploration workflow, you can create an agent that handles everything from simple CSV summaries to complex multi-source analyses spanning databases and statistical models. As you scale, the multi-agent architecture lets you decompose complexity into manageable, specialized units that collaborate seamlessly. The result is a powerful, auditable, and accessible data analysis system that amplifies human analytical capacity rather than replacing it.