Session Management#

This guide covers how to use Redis-backed session storage with the OpenAI Agents SDK.

Overview#

Redis OpenAI Agents provides two session types:

AgentSession: Hash-based storage built on RedisVL’s MessageHistory
JSONSession: JSON document storage for complex nested data

Both provide persistent, distributed conversation storage that replaces the default SQLite sessions.

Prerequisites#

Make sure you have Redis 8 running:

docker run -d --name redis -p 6379:6379 redis:8

import os
os.environ.setdefault("OPENAI_API_KEY", "your-api-key-here")

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")

AgentSession#

AgentSession is the primary session class, providing SDK-compatible conversation storage.

Creating a Session#

from redis_openai_agents import AgentSession

# Create a new session
session = AgentSession.create(
    user_id="user_123",
    redis_url=REDIS_URL,
    ttl=3600  # Optional: 1 hour TTL
)

print(f"Session created: {session.conversation_id}")
print(f"User ID: {session.user_id}")

Storing Agent Results#

from agents import Agent, Runner

# Define an agent
agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant. Remember information the user tells you."
)

# Run the agent
result = await Runner.run(agent, input="My name is Alice and I live in Seattle.")
print(f"Agent response: {result.final_output}")

# Store the conversation
session.store_agent_result(result)
print(f"Conversation stored with {len(session.get_messages())} messages")

Loading a Session#

# Save the conversation ID for later
conversation_id = session.conversation_id

# Later: Load the session
loaded_session = AgentSession.load(
    conversation_id=conversation_id,
    user_id="user_123",
    redis_url=REDIS_URL
)

print(f"Loaded session: {loaded_session.conversation_id}")
print(f"Messages: {len(loaded_session.get_messages())}")

Continuing a Conversation#

# Get conversation history in SDK format
history = loaded_session.to_agent_inputs()

# Continue the conversation
result = await Runner.run(
    agent, 
    input=history + [{"role": "user", "content": "What's my name and where do I live?"}]
)

print(f"Agent response: {result.final_output}")

# Store the new exchange
loaded_session.store_agent_result(result)

Listing User Sessions#

# List all sessions for a user
sessions = AgentSession.list_sessions(
    user_id="user_123",
    redis_url=REDIS_URL
)

print(f"Found {len(sessions)} sessions for user_123")
for s in sessions:
    print(f"  - {s}")

JSONSession#

JSONSession provides async-compatible storage for complex nested session data.

from redis_openai_agents import JSONSession

# Create a JSON session
json_session = JSONSession(
    session_id="complex-session-1",
    redis_url=REDIS_URL,
    ttl=3600
)

# Store complex nested data
await json_session.set("user_preferences", {
    "theme": "dark",
    "language": "en",
    "notifications": {
        "email": True,
        "push": False
    }
})

# Store conversation with metadata
await json_session.set("conversation", {
    "messages": [
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi there!"}
    ],
    "metadata": {
        "agent": "support",
        "started_at": "2025-01-29T10:00:00Z"
    }
})

# Retrieve data
preferences = await json_session.get("user_preferences")
print(f"Theme: {preferences['theme']}")
print(f"Email notifications: {preferences['notifications']['email']}")

# Get all session data
all_data = await json_session.get_all()
print(f"\nAll session keys: {list(all_data.keys())}")

Best Practices#

1. Use TTL for Automatic Cleanup#

session = AgentSession.create(
    user_id="user_123",
    redis_url=REDIS_URL,
    ttl=86400  # 24 hours
)

2. Store Conversation IDs Externally#

Keep track of conversation IDs in your application database to resume sessions later.

3. Use Prefixes for Multi-Tenancy#

# Each tenant gets isolated sessions
session = AgentSession.create(
    user_id=f"tenant_{tenant_id}:user_{user_id}",
    redis_url=REDIS_URL
)

4. Clear Sensitive Data#

# Clear session when no longer needed
session.clear()

Cleanup#

# Clean up test sessions
session.clear()
loaded_session.clear()
await json_session.clear()

print("Sessions cleaned up!")

Version