Skip to main content

OpenAI Agents SDK Integration

HatiData integrates with the OpenAI Agents SDK to give agents persistent memory that survives process restarts, stateful tool usage backed by SQL, and automatic context injection from prior interactions. Every memory read and write is attributed, audited, and metered through HatiData's ABAC policy pipeline.

Installation​

pip install hatidata openai-agents

Requirements: Python 3.10+, an OpenAI API key, and a running HatiData proxy (local or cloud).

Set your credentials:

export OPENAI_API_KEY="sk-..."
export HATIDATA_API_KEY="hd_live_your_api_key"
export HATIDATA_HOST="localhost"  # or your-org.proxy.hatidata.com

Memory-Augmented Agent​

The core pattern is to inject relevant memories into the agent's system prompt before each turn and persist new facts after each response.

import os
from agents import Agent, Runner
from hatidata import HatiDataClient
from hatidata.memory import MemoryClient

# Initialize HatiData
hd = HatiDataClient(
    host=os.environ["HATIDATA_HOST"],
    port=5439,
    api_key=os.environ["HATIDATA_API_KEY"],
)
memory = MemoryClient(hd)

AGENT_ID = "openai-support-agent"

def build_context(user_query: str, session_id: str) -> str:
    """Retrieve relevant memories and format as context block."""
    results = memory.search(
        agent_id=AGENT_ID,
        query=user_query,
        filters={"session_id": session_id},
        top_k=5,
        min_score=0.7,
    )
    if not results:
        return ""
    lines = [f"- {r.content}" for r in results]
    return "Relevant context from prior interactions:\n" + "\n".join(lines)

async def run_agent(user_query: str, session_id: str = "default"):
    context = build_context(user_query, session_id)

    agent = Agent(
        name="Support Agent",
        instructions=(
            "You are a helpful support agent. Use the provided context "
            "to personalize your responses.\n\n" + context
        ),
        model="gpt-4o",
    )

    result = await Runner.run(agent, user_query)

    # Persist the interaction as a new memory
    memory.store(
        agent_id=AGENT_ID,
        content=f"User asked: {user_query}\nAgent replied: {result.final_output[:200]}",
        metadata={
            "session_id": session_id,
            "type": "interaction",
        },
    )

    return result.final_output

Stateful Tool Usage​

Define tools that read and write HatiData tables. The OpenAI Agents SDK will invoke these tools automatically based on the agent's reasoning.

from agents import Agent, Runner, function_tool

@function_tool
def lookup_customer(customer_id: str) -> str:
    """Look up a customer by ID and return their profile."""
    rows = hd.query(
        "SELECT name, email, tier, created_at FROM customers WHERE customer_id = %s",
        [customer_id],
    )
    if not rows:
        return f"No customer found with ID {customer_id}"
    row = rows[0]
    return f"Name: {row['name']}, Email: {row['email']}, Tier: {row['tier']}"

@function_tool
def search_knowledge_base(query: str) -> str:
    """Search the knowledge base for articles relevant to the query."""
    rows = hd.query(f"""
        SELECT title, content,
               semantic_rank(content, '{query}') AS relevance
        FROM knowledge_base
        WHERE semantic_match(content, '{query}', 0.7)
        ORDER BY relevance DESC
        LIMIT 3
    """)
    if not rows:
        return "No relevant articles found."
    return "\n\n".join(
        f"**{r['title']}** (relevance: {r['relevance']:.2f})\n{r['content']}"
        for r in rows
    )

@function_tool
def store_customer_preference(customer_id: str, preference: str) -> str:
    """Store a customer preference for future reference."""
    memory.store(
        agent_id=AGENT_ID,
        content=preference,
        metadata={
            "customer_id": customer_id,
            "type": "preference",
        },
    )
    return f"Stored preference for customer {customer_id}"

agent = Agent(
    name="Support Agent",
    instructions="You are a support agent with access to customer data and a knowledge base.",
    tools=[lookup_customer, search_knowledge_base, store_customer_preference],
    model="gpt-4o",
)

Context Injection Patterns​

Pattern 1: Pre-Turn Injection​

Inject memories into the system prompt before each agent turn. Best for short-lived agents that process a single request.

context = build_context(user_query, session_id)
agent = Agent(
    name="Agent",
    instructions=f"You are a helpful agent.\n\nContext:\n{context}",
    model="gpt-4o",
)

Pattern 2: Tool-Based Retrieval​

Let the agent decide when to search memory by providing a recall_memory tool. Best for multi-turn agents that need selective retrieval.

@function_tool
def recall_memory(query: str) -> str:
    """Search agent memory for relevant past interactions."""
    results = memory.search(
        agent_id=AGENT_ID,
        query=query,
        top_k=5,
    )
    if not results:
        return "No relevant memories found."
    return "\n".join(f"[{r.score:.2f}] {r.content}" for r in results)

Pattern 3: Handoff with Shared Memory​

Use the OpenAI Agents SDK handoff mechanism with a shared HatiData memory namespace so that context transfers between specialized agents.

from agents import Agent, Runner

triage_agent = Agent(
    name="Triage",
    instructions="Route the user to the right specialist.",
    handoffs=["billing_agent", "technical_agent"],
    model="gpt-4o",
)

billing_agent = Agent(
    name="Billing Agent",
    instructions="Handle billing inquiries. Check memory for prior billing interactions.",
    tools=[recall_memory, lookup_customer],
    model="gpt-4o",
)

technical_agent = Agent(
    name="Technical Agent",
    instructions="Handle technical issues. Check memory for prior technical context.",
    tools=[recall_memory, search_knowledge_base],
    model="gpt-4o",
)

Both specialist agents read from the same HatiData memory store, so context gathered by the triage agent is available to whichever specialist handles the request.

Chain-of-Thought Logging​

Log every agent reasoning step to HatiData's immutable CoT ledger for compliance and debugging.

from hatidata.cot import CotClient, StepType

cot = CotClient(hd)

async def run_with_audit(user_query: str, session_id: str):
    cot.log_step(
        agent_id=AGENT_ID,
        session_id=session_id,
        step_type=StepType.OBSERVATION,
        content=f"Received query: {user_query}",
    )

    result = await Runner.run(agent, user_query)

    cot.log_step(
        agent_id=AGENT_ID,
        session_id=session_id,
        step_type=StepType.CONCLUSION,
        content=f"Final output: {result.final_output[:200]}",
        metadata={"tool_calls": len(result.raw_responses)},
    )

    return result.final_output

Stay in the loop

Product updates, engineering deep-dives, and agent-native insights. No spam.