Skip to content

Encrypted Sessions

EncryptedSession provides transparent encryption for any session implementation, securing conversation data with automatic expiration of old items.

Features

  • Transparent encryption: Wraps any session with Fernet encryption
  • Per-session keys: Uses HKDF key derivation for unique encryption per session
  • Automatic expiration: Old items are silently skipped when TTL expires
  • Drop-in replacement: Works with any existing session implementation

Installation

Encrypted sessions require the encrypt extra:

pip install openai-agents[encrypt]

Quick start

import asyncio
from agents import Agent, Runner
from agents.extensions.memory import EncryptedSession, SQLAlchemySession

async def main():
    agent = Agent("Assistant")

    # Create underlying session
    underlying_session = SQLAlchemySession.from_url(
        "user-123",
        url="sqlite+aiosqlite:///:memory:",
        create_tables=True
    )

    # Wrap with encryption
    session = EncryptedSession(
        session_id="user-123",
        underlying_session=underlying_session,
        encryption_key="your-secret-key-here",
        ttl=600  # 10 minutes
    )

    result = await Runner.run(agent, "Hello", session=session)
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

Configuration

Encryption key

The encryption key can be either a Fernet key or any string:

from agents.extensions.memory import EncryptedSession

# Using a Fernet key (base64-encoded)
session = EncryptedSession(
    session_id="user-123",
    underlying_session=underlying_session,
    encryption_key="your-fernet-key-here",
    ttl=600
)

# Using a raw string (will be derived to a key)
session = EncryptedSession(
    session_id="user-123", 
    underlying_session=underlying_session,
    encryption_key="my-secret-password",
    ttl=600
)

TTL (Time To Live)

Set how long encrypted items remain valid:

# Items expire after 1 hour
session = EncryptedSession(
    session_id="user-123",
    underlying_session=underlying_session,
    encryption_key="secret",
    ttl=3600  # 1 hour in seconds
)

# Items expire after 1 day
session = EncryptedSession(
    session_id="user-123",
    underlying_session=underlying_session,
    encryption_key="secret", 
    ttl=86400  # 24 hours in seconds
)

Usage with different session types

With SQLite sessions

from agents import SQLiteSession
from agents.extensions.memory import EncryptedSession

# Create encrypted SQLite session
underlying = SQLiteSession("user-123", "conversations.db")

session = EncryptedSession(
    session_id="user-123",
    underlying_session=underlying,
    encryption_key="secret-key"
)

With SQLAlchemy sessions

from agents.extensions.memory import EncryptedSession, SQLAlchemySession

# Create encrypted SQLAlchemy session
underlying = SQLAlchemySession.from_url(
    "user-123",
    url="postgresql+asyncpg://user:pass@localhost/db",
    create_tables=True
)

session = EncryptedSession(
    session_id="user-123",
    underlying_session=underlying,
    encryption_key="secret-key"
)

Advanced Session Features

When using EncryptedSession with advanced session implementations like AdvancedSQLiteSession, note that:

  • Methods like find_turns_by_content() won't work effectively since message content is encrypted
  • Content-based searches operate on encrypted data, limiting their effectiveness

Key derivation

EncryptedSession uses HKDF (HMAC-based Key Derivation Function) to derive unique encryption keys per session:

  • Master key: Your provided encryption key
  • Session salt: The session ID
  • Info string: "agents.session-store.hkdf.v1"
  • Output: 32-byte Fernet key

This ensures that: - Each session has a unique encryption key - Keys cannot be derived without the master key - Session data cannot be decrypted across different sessions

Automatic expiration

When items exceed the TTL, they are automatically skipped during retrieval:

# Items older than TTL are silently ignored
items = await session.get_items()  # Only returns non-expired items

# Expired items don't affect session behavior
result = await Runner.run(agent, "Continue conversation", session=session)

API Reference