OpenAI Agents SDK + HatiData: Stateful Agent Patterns
The Stateless Problem in OpenAI Agents
The OpenAI Agents SDK provides a clean framework for building agents with tool use, handoffs, and guardrails. But agents built with the SDK are fundamentally stateless — each invocation starts fresh. The agent has no memory of previous conversations, no accumulated knowledge, and no persistent state.
This is fine for single-turn tasks: "Summarize this document," "Convert this data," "Answer this question." But production agents often need continuity: "Remember that this customer prefers email communication," "Last time we discussed migrating to the cloud tier," "The user's timezone is UTC+8."
HatiData adds persistent state to OpenAI Agents SDK agents through three patterns: memory-augmented context injection, stateful tool definitions, and checkpoint-based state management.
Pattern 1: Memory-Augmented Context Injection
The simplest integration pattern retrieves relevant memories before each agent invocation and injects them into the system prompt. This requires no changes to the agent's tool definitions — the memories appear as additional context.
import httpx
from openai import agents
HATIDATA_URL = "http://localhost:5439"
HATIDATA_KEY = "hd_live_your_key"
async def get_relevant_memories(user_input: str, namespace: str, limit: int = 5):
"""Retrieve relevant memories from HatiData."""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{HATIDATA_URL}/v1/memory/search",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
json={
"query": user_input,
"namespace": namespace,
"limit": limit,
"min_similarity": 0.65,
},
)
return response.json()["memories"]
async def store_memory(content: str, namespace: str, metadata: dict = None):
"""Store a memory in HatiData."""
async with httpx.AsyncClient() as client:
await client.post(
f"{HATIDATA_URL}/v1/memory/store",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
json={
"content": content,
"namespace": namespace,
"metadata": metadata or {},
},
)
async def run_agent_with_memory(user_input: str, user_id: str):
# Retrieve relevant memories
memories = await get_relevant_memories(user_input, f"user/{user_id}")
# Format memories as context
memory_context = "\n".join(
[f"- {m['content']} (similarity: {m['similarity']:.2f})"
for m in memories]
)
# Create agent with memory-augmented prompt
agent = agents.Agent(
name="Support Agent",
instructions=f"""You are a helpful support agent.
Previous relevant context about this user:
{memory_context}
Use this context to provide personalized, consistent responses.""",
model="gpt-4o",
tools=[...], # Your existing tools
)
# Run the agent
result = await agents.Runner.run(agent, user_input)
# Store the interaction as a new memory
await store_memory(
content=f"User asked: {user_input}. Agent responded: {result.final_output}",
namespace=f"user/{user_id}",
metadata={"type": "interaction", "user_id": user_id},
)
return resultThis pattern works with any OpenAI Agents SDK agent without modifying its tool definitions. The memories are injected as part of the system instructions, so the agent naturally incorporates them into its responses.
Pattern 2: Stateful Tool Definitions
For agents that need more control over memory operations, you can define HatiData operations as explicit tools that the agent can invoke:
from openai import agents
from openai.agents import tool
@tool
async def remember(content: str, importance: str = "normal") -> str:
"""Store an important fact or observation for future reference.
Args:
content: The information to remember
importance: How important this memory is (low, normal, high)
"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{HATIDATA_URL}/v1/memory/store",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
json={
"content": content,
"namespace": "agent/observations",
"metadata": {"importance": importance},
},
)
data = response.json()
return f"Stored memory: {data['memory_id']}"
@tool
async def recall(query: str, limit: int = 5) -> str:
"""Search memories for information relevant to a query.
Args:
query: What to search for in memories
limit: Maximum number of memories to return
"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{HATIDATA_URL}/v1/memory/search",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
json={
"query": query,
"namespace": "agent/observations",
"limit": limit,
},
)
memories = response.json()["memories"]
if not memories:
return "No relevant memories found."
return "\n".join(
[f"- {m['content']} (relevance: {m['similarity']:.0%})"
for m in memories]
)
@tool
async def query_data(sql: str) -> str:
"""Execute a SQL query against the data warehouse.
Args:
sql: The SQL query to execute
"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{HATIDATA_URL}/v1/query",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
json={"sql": sql},
)
return str(response.json()["results"])
agent = agents.Agent(
name="Research Agent",
instructions="You are a research agent with persistent memory. Use the remember tool to store important findings and the recall tool to retrieve past knowledge.",
model="gpt-4o",
tools=[remember, recall, query_data],
)With this pattern, the agent decides when to store and retrieve memories. It might recall past research before starting a new task, store key findings as it works, and retrieve stored conclusions when writing a summary.
Pattern 3: Checkpoint State Management
For long-running agent workflows, HatiData's agent state provides checkpoint functionality — the ability to save and resume workflow progress across sessions:
@tool
async def save_checkpoint(key: str, state: dict) -> str:
"""Save workflow progress so it can be resumed later.
Args:
key: A unique identifier for this checkpoint
state: The workflow state to save
"""
async with httpx.AsyncClient() as client:
await client.post(
f"{HATIDATA_URL}/v1/state/set",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
json={"key": key, "value": state},
)
return f"Checkpoint saved: {key}"
@tool
async def load_checkpoint(key: str) -> str:
"""Load a previously saved workflow checkpoint.
Args:
key: The checkpoint identifier
"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{HATIDATA_URL}/v1/state/get/{key}",
headers={"Authorization": f"Bearer {HATIDATA_KEY}"},
)
state = response.json()["value"]
if state is None:
return "No checkpoint found."
return str(state)This pattern is essential for agents that process large datasets incrementally, agents that run multi-step workflows spanning multiple sessions, or agents that need to resume after failures without repeating completed work.
Agent Handoffs with Memory
The OpenAI Agents SDK supports handoffs between agents. Combined with HatiData memory, handoffs can include persistent context:
triage_agent = agents.Agent(
name="Triage Agent",
instructions="Classify incoming requests and hand off to the appropriate specialist. Store the classification in memory for future reference.",
tools=[remember, recall],
handoffs=[technical_agent, billing_agent],
)
technical_agent = agents.Agent(
name="Technical Support",
instructions="Handle technical issues. Check memory for previous interactions with this user.",
tools=[remember, recall, query_data],
)When the triage agent hands off to the technical agent, the technical agent can recall memories stored by the triage agent (or by previous sessions), providing continuity across the handoff.
Production Deployment
For production OpenAI Agents SDK deployments with HatiData:
- API key per agent role — The triage agent, technical agent, and billing agent each get their own HatiData API key with appropriate namespace access
- Error handling — Wrap HatiData calls in try/except blocks with fallback behavior (agent continues without memory if HatiData is temporarily unavailable)
- Rate limiting — HatiData's quota system prevents any single agent from overwhelming the memory store
- Monitoring — Track memory store/retrieve latency and hit rates to understand agent memory usage patterns
Next Steps
The OpenAI Agents SDK integration works through standard HTTP APIs, making it compatible with any deployment model. For more sophisticated memory patterns, see the LangChain deep dive. For multi-agent shared memory, see the CrewAI integration guide. For the full MCP tool suite, see the Claude MCP guide.