OpenAI Agents SDK Integration

HatiData integrates with the OpenAI Agents SDK by exposing warehouse operations as tools that agents can call during their reasoning loop. Agents get schema discovery, SQL execution, memory search, and vector-aware querying -- all attributed and metered per-agent.

Installation

pip install hatidata-agent
pip install openai-agents

Configure

from hatidata_agent import HatiDataAgent

hati = HatiDataAgent(
    host="your-org.proxy.hatidata.com",
    port=5439,
    agent_id="openai-agent",
    framework="openai-agents",
    password="hd_live_your_api_key",
)

Parameter	Default	Description
host	"localhost"	HatiData proxy hostname
port	5439	Proxy port
agent_id	"agent"	Unique identifier for billing and audit
framework	"custom"	Set to "openai-agents" for proper tracking
database	"hatidata"	Target database
password	""	API key (hd_live_* or hd_test_*)

Basic Usage

Define HatiData operations as tool functions and register them with an OpenAI agent:

from hatidata_agent import HatiDataAgent
from agents import Agent, Runner, function_tool

hati = HatiDataAgent(
    host="your-org.proxy.hatidata.com",
    port=5439,
    agent_id="openai-analyst",
    framework="openai-agents",
    password="hd_live_your_api_key",
)

@function_tool
def query_warehouse(sql: str) -> str:
    """Execute a SQL query against the data warehouse and return results."""
    try:
        rows = hati.query(sql)
        return str(rows) if rows else "No results."
    except Exception as e:
        return f"Error: {e}"

@function_tool
def list_tables() -> str:
    """List all available tables in the data warehouse."""
    rows = hati.query(
        "SELECT table_name FROM information_schema.tables WHERE table_schema = 'main'"
    )
    return ", ".join(r["table_name"] for r in rows)

@function_tool
def describe_table(table_name: str) -> str:
    """Get column names and data types for a specific table."""
    rows = hati.query(
        f"SELECT column_name, data_type FROM information_schema.columns "
        f"WHERE table_name = '{table_name}'"
    )
    return "\n".join(f"{r['column_name']} {r['data_type']}" for r in rows)

@function_tool
def search_memory(query: str) -> str:
    """Search agent memory for relevant past findings."""
    results = hati.search_memory(query=query, top_k=5)
    if not results:
        return "No relevant memories found."
    return "\n".join(f"[{r['importance']:.1f}] {r['content']}" for r in results)

# Create the agent
agent = Agent(
    name="data_analyst",
    instructions="""You are a data analyst with access to a SQL data warehouse.
    Always start by listing tables and describing relevant schemas before querying.
    Store important findings in memory for future reference.""",
    tools=[query_warehouse, list_tables, describe_table, search_memory],
)

# Run the agent
result = Runner.run_sync(
    agent,
    "What is the average order value by customer segment this quarter?",
)
print(result.final_output)

Multi-Agent Handoffs

The OpenAI Agents SDK supports agent handoffs. Combine a schema explorer, SQL analyst, and report writer with shared HatiData access:

from agents import Agent, Runner, function_tool

# Schema explorer agent
explorer = Agent(
    name="schema_explorer",
    instructions="You discover and describe database schemas. Use list_tables and describe_table.",
    tools=[list_tables, describe_table],
)

# SQL analyst agent
analyst = Agent(
    name="sql_analyst",
    instructions="""You write and execute SQL queries based on schema information.
    Use query_warehouse to execute queries. Always validate results.""",
    tools=[query_warehouse, search_memory],
)

# Orchestrator agent that delegates
orchestrator = Agent(
    name="orchestrator",
    instructions="""You coordinate data analysis tasks.
    Hand off to schema_explorer first to understand the data,
    then to sql_analyst to run queries and produce results.""",
    handoffs=[explorer, analyst],
)

result = Runner.run_sync(
    orchestrator,
    "Analyze customer retention rates and identify the top factors driving churn.",
)
print(result.final_output)

Storing Findings

Agents can persist insights for future sessions:

@function_tool
def store_finding(content: str, importance: float = 0.7) -> str:
    """Store an analytical finding in long-term agent memory."""
    result = hati.store_memory(
        content=content,
        memory_type="fact",
        importance=importance,
    )
    return f"Stored: {result['memory_id']}"

agent = Agent(
    name="analyst_with_memory",
    instructions="Analyze data and store important findings using store_finding.",
    tools=[query_warehouse, list_tables, describe_table, search_memory, store_finding],
)

Guardrails with HatiData Policies

HatiData's ABAC policies enforce guardrails server-side, complementing the Agents SDK's built-in guardrail system:

• Row-level security -- Agents only see data they are authorized to access
• Column masking -- Sensitive columns are redacted based on agent role
• Query cost limits -- Per-agent credit quotas prevent runaway queries
• DDL restrictions -- Agents can be blocked from schema modifications

These policies are enforced at the proxy level, so they apply regardless of what the agent attempts. See Authorization for policy configuration.

Next Steps

• Agent Integrations -- All supported frameworks
• Agent Memory -- How persistent memory works
• MCP Tools Reference -- All 24 MCP tools
• Python SDK -- Full SDK reference