Building Encrypted Sessions for Secure Agent Memory

Why Encrypt Agent Sessions?

Agent conversations often contain sensitive data: personal health information, financial details, customer support interactions with account numbers, or proprietary business information. Storing this data in plain text — even in a database with access controls — may not meet regulatory requirements or your organization's security posture.

The OpenAI Agents SDK includes an EncryptedSession wrapper that adds transparent encryption and decryption around any session backend. Your agents work exactly the same way, but the data at rest is encrypted with AES-GCM.

How EncryptedSession Works

EncryptedSession is a decorator pattern. It wraps any existing session implementation (SQLiteSession, RedisSession, SQLAlchemySession) and encrypts items before writing and decrypts them after reading.

flowchart TD
    START["Building Encrypted Sessions for Secure Agent Memo…"] --> A
    A["Why Encrypt Agent Sessions?"]
    A --> B
    B["How EncryptedSession Works"]
    B --> C
    C["Transparent Encryption and Decryption"]
    C --> D
    D["The Encryption Flow"]
    D --> E
    E["Key Management Strategies"]
    E --> F
    F["Key Rotation"]
    F --> G
    G["TTL for Session Expiry"]
    G --> H
    H["HIPAA and SOC2 Compliance Patterns"]
    H --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

from agents.extensions.sessions import SQLiteSession, EncryptedSession

# Base session — stores data
base_session = SQLiteSession(db_path="./sessions.db")

# Encrypted wrapper — encrypts/decrypts transparently
encryption_key = "your-32-byte-encryption-key-here"  # Must be 32 bytes for AES-256
session = EncryptedSession(
    session=base_session,
    encryption_key=encryption_key,
)

From the agent's perspective, nothing changes. You pass the encrypted session to Runner.run() just like any other session.

Transparent Encryption and Decryption

The encryption is completely transparent to your application code. The agent, runner, and tools never see encrypted data — they work with plain text. Only the storage layer sees ciphertext.

import asyncio
from agents import Agent, Runner
from agents.extensions.sessions import (
    SQLiteSession,
    EncryptedSession,
)

ENCRYPTION_KEY = b"0123456789abcdef0123456789abcdef"  # 32 bytes

base_session = SQLiteSession(db_path="./encrypted_sessions.db")
session = EncryptedSession(session=base_session, encryption_key=ENCRYPTION_KEY)

agent = Agent(
    name="HealthAgent",
    instructions="You are a medical assistant. Handle patient information with care.",
)

async def main():
    sid = "patient-consultation-101"

    # Store sensitive information
    result = await Runner.run(
        agent,
        "Patient John Doe, DOB 1985-03-15, diagnosed with Type 2 diabetes.",
        session=session,
        session_id=sid,
    )
    print(result.final_output)

    # Retrieve it — decrypted transparently
    result = await Runner.run(
        agent,
        "What is the patient's diagnosis?",
        session=session,
        session_id=sid,
    )
    print(result.final_output)  # References Type 2 diabetes

asyncio.run(main())

If you open the SQLite database directly, the stored data is encrypted ciphertext — unreadable without the key.

The Encryption Flow

Here is what happens on each operation:

flowchart TD
    ROOT["Building Encrypted Sessions for Secure Agent…"] 
    ROOT --> P0["Key Management Strategies"]
    P0 --> P0C0["Strategy 1: Environment Variables"]
    P0 --> P0C1["Strategy 2: AWS KMS with Data Key Encry…"]
    P0 --> P0C2["Strategy 3: HashiCorp Vault"]
    ROOT --> P1["HIPAA and SOC2 Compliance Patterns"]
    P1 --> P1C0["HIPAA Requirements for Agent Sessions"]
    P1 --> P1C1["SOC2 Requirements"]
    style ROOT fill:#4f46e5,stroke:#4338ca,color:#fff
    style P0 fill:#e0e7ff,stroke:#6366f1,color:#1e293b
    style P1 fill:#e0e7ff,stroke:#6366f1,color:#1e293b

Writing (add_items):

1. Agent generates response items (plain text)
2. EncryptedSession serializes each item to JSON
3. Each JSON blob is encrypted with AES-256-GCM
4. A unique nonce is generated per item
5. The ciphertext + nonce are stored in the base session

Reading (get_items):

1. EncryptedSession retrieves ciphertext from base session
2. Each item is decrypted using the key and stored nonce
3. JSON is deserialized back to item objects
4. Plain text items are returned to the runner

Key Management Strategies

The encryption key is the most critical piece. How you manage it determines your actual security level.

Strategy 1: Environment Variables

Simplest approach — suitable for single-service deployments:

import os
from agents.extensions.sessions import SQLiteSession, EncryptedSession

key = os.environ["AGENT_SESSION_ENCRYPTION_KEY"].encode()
assert len(key) == 32, "Key must be 32 bytes for AES-256"

session = EncryptedSession(
    session=SQLiteSession(db_path="./sessions.db"),
    encryption_key=key,
)

Strategy 2: AWS KMS with Data Key Encryption

For production systems, use an envelope encryption pattern with a cloud KMS:

import boto3
import os
from agents.extensions.sessions import RedisSession, EncryptedSession

def get_data_key() -> bytes:
    """Retrieve or generate a data encryption key from AWS KMS."""
    kms = boto3.client("kms")

    # Check for cached data key
    cached_key = os.environ.get("CACHED_DATA_KEY")
    if cached_key:
        return bytes.fromhex(cached_key)

    # Generate new data key
    response = kms.generate_data_key(
        KeyId="alias/agent-sessions",
        KeySpec="AES_256",
    )

    # Store encrypted version for recovery
    # In production, persist response["CiphertextBlob"] somewhere safe

    return response["Plaintext"]  # 32-byte raw key

session = EncryptedSession(
    session=RedisSession.from_url("redis://redis:6379/0"),
    encryption_key=get_data_key(),
)

Strategy 3: HashiCorp Vault

For organizations using Vault for secrets management:

import hvac
from agents.extensions.sessions import SQLAlchemySession, EncryptedSession

def get_key_from_vault() -> bytes:
    client = hvac.Client(url="https://vault.internal:8200")
    client.auth.kubernetes.login(role="agent-service")
    secret = client.secrets.kv.v2.read_secret_version(
        path="agent-sessions/encryption-key"
    )
    return bytes.fromhex(secret["data"]["data"]["key"])

async def create_encrypted_session():
    base = await SQLAlchemySession.from_url(
        "postgresql+asyncpg://user:pass@db:5432/agents",
        create_tables=False,
    )
    return EncryptedSession(
        session=base,
        encryption_key=get_key_from_vault(),
    )

Key Rotation

Key rotation is essential for long-lived systems. The approach depends on your backend, but the general pattern involves re-encrypting existing sessions with a new key.

flowchart TD
    CENTER(("Core Concepts"))
    CENTER --> N0["Agent generates response items plain te…"]
    CENTER --> N1["EncryptedSession serializes each item t…"]
    CENTER --> N2["Each JSON blob is encrypted with AES-25…"]
    CENTER --> N3["A unique nonce is generated per item"]
    CENTER --> N4["The ciphertext + nonce are stored in th…"]
    CENTER --> N5["EncryptedSession retrieves ciphertext f…"]
    style CENTER fill:#4f46e5,stroke:#4338ca,color:#fff

async def rotate_encryption_key(
    base_session,
    old_key: bytes,
    new_key: bytes,
    session_ids: list[str],
):
    """Re-encrypt all sessions with a new key."""
    old_encrypted = EncryptedSession(session=base_session, encryption_key=old_key)
    new_encrypted = EncryptedSession(session=base_session, encryption_key=new_key)

    for sid in session_ids:
        # Read with old key
        items = await old_encrypted.get_items(sid)

        # Clear old data
        await old_encrypted.clear_session(sid)

        # Write with new key
        await new_encrypted.add_items(sid, items)

    print(f"Rotated {len(session_ids)} sessions to new key")

TTL for Session Expiry

Combine encryption with TTL to ensure sensitive conversations are automatically purged:

from agents.extensions.sessions import RedisSession, EncryptedSession

redis_session = RedisSession.from_url("redis://redis:6379/0")

encrypted_session = EncryptedSession(
    session=redis_session,
    encryption_key=ENCRYPTION_KEY,
)

# After each interaction, refresh the TTL
async def handle_with_ttl(session_id: str, message: str):
    result = await Runner.run(
        agent, message, session=encrypted_session, session_id=session_id
    )

    # Set 24-hour TTL — session auto-deletes if inactive
    await redis_session.client.expire(
        f"session:{session_id}",
        60 * 60 * 24  # 24 hours
    )

    return result.final_output

For SQLAlchemy-backed sessions, implement TTL with a scheduled cleanup job:

from sqlalchemy import text
from datetime import datetime, timedelta

async def cleanup_expired_sessions(engine, max_age_days: int = 30):
    """Delete sessions older than max_age_days."""
    cutoff = datetime.utcnow() - timedelta(days=max_age_days)
    async with engine.begin() as conn:
        result = await conn.execute(
            text("DELETE FROM session_items WHERE created_at < :cutoff"),
            {"cutoff": cutoff},
        )
        print(f"Purged {result.rowcount} expired session items")

HIPAA and SOC2 Compliance Patterns

HIPAA Requirements for Agent Sessions

If your agent handles Protected Health Information (PHI), HIPAA requires:

1. Encryption at rest: EncryptedSession satisfies this.
2. Encryption in transit: Use TLS for database connections.
3. Access controls: Restrict who can access the encryption keys.
4. Audit logging: Log access to session data.
5. Data retention policies: Implement TTL-based purging.

import logging
from agents.extensions.sessions import EncryptedSession

logger = logging.getLogger("hipaa_audit")

class AuditedEncryptedSession(EncryptedSession):
    """EncryptedSession with HIPAA audit logging."""

    async def get_items(self, session_id: str, **kwargs):
        logger.info(f"SESSION_READ session_id={session_id} timestamp={datetime.utcnow().isoformat()}")
        return await super().get_items(session_id, **kwargs)

    async def add_items(self, session_id: str, items, **kwargs):
        logger.info(f"SESSION_WRITE session_id={session_id} items={len(items)} timestamp={datetime.utcnow().isoformat()}")
        return await super().add_items(session_id, items, **kwargs)

    async def clear_session(self, session_id: str, **kwargs):
        logger.info(f"SESSION_DELETE session_id={session_id} timestamp={datetime.utcnow().isoformat()}")
        return await super().clear_session(session_id, **kwargs)

SOC2 Requirements

SOC2 Type II compliance focuses on availability, security, processing integrity, confidentiality, and privacy. For agent sessions, the key requirements are:

• Encryption at rest and in transit — EncryptedSession plus TLS connections
• Key management — Use a KMS, not hardcoded keys
• Retention policies — Implement automatic data purging
• Access logging — Log all reads and writes to session data
• Incident response — Ability to purge a specific user's sessions immediately

async def purge_user_sessions(user_id: str, session: EncryptedSession):
    """Emergency purge all sessions for a user — for incident response."""
    session_ids = await get_user_session_ids(user_id)  # From your user-session mapping
    for sid in session_ids:
        await session.clear_session(sid)
    logger.critical(f"EMERGENCY_PURGE user_id={user_id} sessions_purged={len(session_ids)}")

Encryption is one layer. True compliance requires encryption, access controls, audit logging, retention policies, and incident response procedures working together.

Sources:

• https://openai.github.io/openai-agents-python/sessions/
• https://www.hhs.gov/hipaa/for-professionals/security/guidance/index.html