Skip to content

Session Management

Sessions allow you to persist state across conversation turns, enabling multi-step workflows where tools can share data.


Overview

A Session is a key-value store that:

  • Persists across turns - Data survives between user messages
  • Travels with the protocol - Automatically serialized in responses
  • Supports typed models - Store Pydantic models and dataclasses with auto-serialization
  • Provides simple API - Dict-like access with type hints
  • Available everywhere - In tools, interceptors, custom agent functions, and agent.run()

Basic Usage

Creating and Using Sessions

from dcaf.core import Session

session = Session()

# Set values
session.set("user_id", "12345")
session.set("preferences", {"theme": "dark", "language": "en"})

# Get values
user_id = session.get("user_id")  # "12345"
prefs = session.get("preferences")  # {"theme": "dark", ...}

# Get with default
count = session.get("count", 0)  # Returns 0 if not set

# Check existence
if session.has("user_id"):
    print("User is logged in")

# Delete values
session.delete("user_id")

Dict-Like Access

Session supports familiar dictionary-style access:

# Set/get with brackets
session["key"] = "value"
value = session["key"]

# Iteration
for key in session.keys():
    print(f"{key}: {session[key]}")

# Get all items
for key, value in session.items():
    print(f"{key}: {value}")

Bulk Operations

# Update multiple values at once
session.update({
    "cart": [],
    "user_id": "12345",
    "step": 1,
})

# Clear all data
session.clear()

# Convert to dict
data = session.to_dict()

Typed Storage

Session supports automatic serialization and deserialization of Pydantic models and dataclasses. This gives you type safety, IDE autocomplete, and cleaner code.

Pydantic Models

from pydantic import BaseModel, Field
from dcaf.core import Session

class UserPreferences(BaseModel):
    theme: str = "light"
    language: str = "en"
    notifications: bool = True

session = Session()

# Store a Pydantic model (auto-serializes to dict)
prefs = UserPreferences(theme="dark", notifications=False)
session.set("prefs", prefs)

# What's actually stored:
# {"theme": "dark", "language": "en", "notifications": False}

# Retrieve as typed model (auto-deserializes)
loaded_prefs = session.get("prefs", as_type=UserPreferences)
print(loaded_prefs.theme)  # "dark"
print(type(loaded_prefs))  # <class 'UserPreferences'>

# Retrieve without type (returns raw dict)
raw = session.get("prefs")
print(raw)  # {"theme": "dark", "language": "en", "notifications": False}

Dataclasses

from dataclasses import dataclass
from dcaf.core import Session

@dataclass
class Config:
    debug: bool = False
    max_retries: int = 3
    timeout: float = 30.0

session = Session()

# Store a dataclass (auto-serializes via asdict())
session.set("config", Config(debug=True, max_retries=5))

# Retrieve as typed
config = session.get("config", as_type=Config)
print(config.debug)  # True
print(config.max_retries)  # 5

Nested Models

Complex nested structures work seamlessly:

from pydantic import BaseModel, Field
from typing import Optional

class CartItem(BaseModel):
    name: str
    quantity: int
    price: float

class ShoppingCart(BaseModel):
    items: list[CartItem] = Field(default_factory=list)
    discount_code: Optional[str] = None

    @property
    def total(self) -> float:
        return sum(item.price * item.quantity for item in self.items)

# Store complex model
cart = ShoppingCart()
cart.items.append(CartItem(name="Widget", quantity=2, price=9.99))
cart.items.append(CartItem(name="Gadget", quantity=1, price=24.99))
session.set("cart", cart)

# Retrieve with full type hierarchy intact
loaded_cart = session.get("cart", as_type=ShoppingCart)
print(loaded_cart.total)  # 44.97
print(loaded_cart.items[0].name)  # "Widget"

Default Values with Types

# Returns None if key doesn't exist
cart = session.get("cart", as_type=ShoppingCart)  # None if not found

# Provide a default instance
cart = session.get("cart", ShoppingCart(), as_type=ShoppingCart)  # Never None

Using Session in Tools

Tools can declare a Session parameter that DCAF automatically injects. The parameter name must be exactly session with type Session.

Basic Example

from dcaf.core import Session
from dcaf.tools import tool

@tool(description="Remember a value")
def remember(key: str, value: str, session: Session) -> str:
    """Store a value in session."""
    session.set(key, value)
    return f"Remembered {key}={value}"

@tool(description="Recall a value")
def recall(key: str, session: Session) -> str:
    """Retrieve a value from session."""
    value = session.get(key)
    if value is None:
        return f"I don't remember '{key}'"
    return f"{key} is '{value}'"

Shopping Cart Example

from dcaf.core import Session
from dcaf.tools import tool

@tool(description="Add item to shopping cart")
def add_to_cart(item: str, quantity: int, session: Session) -> str:
    """Add an item to the user's cart."""
    cart = session.get("cart", [])
    cart.append({"item": item, "quantity": quantity})
    session.set("cart", cart)

    total_items = sum(i["quantity"] for i in cart)
    return f"Added {quantity}x {item}. Cart now has {total_items} items."

@tool(description="View shopping cart contents")
def view_cart(session: Session) -> str:
    """Show what's in the cart."""
    cart = session.get("cart", [])

    if not cart:
        return "Your cart is empty."

    lines = ["Your cart contains:"]
    for item in cart:
        lines.append(f"  - {item['quantity']}x {item['item']}")

    total = sum(i["quantity"] for i in cart)
    lines.append(f"Total: {total} items")
    return "\n".join(lines)

@tool(description="Clear the shopping cart")
def clear_cart(session: Session) -> str:
    """Remove all items from the cart."""
    session.delete("cart")
    return "Cart cleared."

@tool(description="Checkout and complete purchase")
def checkout(session: Session) -> str:
    """Complete the purchase."""
    cart = session.get("cart", [])

    if not cart:
        return "Cannot checkout - cart is empty."

    total_items = sum(i["quantity"] for i in cart)
    session.delete("cart")  # Clear after checkout

    return f"✅ Order placed! {total_items} items will be shipped."

Multi-Step Workflow Example

from dcaf.core import Session
from dcaf.tools import tool
from typing import Literal

@tool(description="Start a deployment workflow")
def start_deployment(
    environment: Literal["staging", "production"],
    session: Session
) -> str:
    """Initialize a deployment workflow."""
    session.set("deployment", {
        "environment": environment,
        "step": "started",
        "services": [],
    })
    return f"Deployment workflow started for {environment}. Add services to deploy."

@tool(description="Add a service to the deployment")
def add_service(service_name: str, version: str, session: Session) -> str:
    """Add a service to the pending deployment."""
    deployment = session.get("deployment")

    if not deployment:
        return "No deployment in progress. Run start_deployment first."

    deployment["services"].append({
        "name": service_name,
        "version": version,
    })
    deployment["step"] = "services_added"
    session.set("deployment", deployment)

    return f"Added {service_name}:{version} to deployment. {len(deployment['services'])} services total."

@tool(requires_approval=True, description="Execute the deployment")
def execute_deployment(session: Session) -> str:
    """Execute the pending deployment (requires approval)."""
    deployment = session.get("deployment")

    if not deployment:
        return "No deployment in progress."

    if not deployment.get("services"):
        return "No services to deploy. Add services first."

    env = deployment["environment"]
    services = deployment["services"]

    # Simulate deployment
    results = []
    for svc in services:
        results.append(f"✅ Deployed {svc['name']}:{svc['version']} to {env}")

    # Clear deployment state
    session.delete("deployment")

    return "\n".join(results)

Using Session with Agent.run()

You can pass session data directly to agent.run() and agent.chat():

from dcaf.core import Agent, Session

agent = Agent(tools=[...])

# Pass session as a dict
response = agent.run(
    messages=[{"role": "user", "content": "Continue the wizard"}],
    session={"wizard_step": 2, "user_name": "Alice"},
)

# Access updated session from response
print(response.session)  # {"wizard_step": 3, "user_name": "Alice", ...}

# Pass session to next request
next_response = agent.run(
    messages=[{"role": "user", "content": "Next step"}],
    session=response.session,
)

You can also pass a Session instance:

from dcaf.core import Agent, Session

session = Session()
session.set("user_id", "12345")

response = agent.run(messages=[...], session=session)

# Session changes are in response
print(response.session)

Using Session in Custom Agent Functions

Custom agent functions receive session as an optional third parameter:

from dcaf.core import serve, Session
from dcaf.core.primitives import AgentResult

def my_agent(messages: list, context: dict, session: Session) -> AgentResult:
    """Custom agent with session access."""
    # Read from session
    call_count = session.get("call_count", 0)

    # Modify session
    session.set("call_count", call_count + 1)
    session.set("last_message", messages[-1]["content"])

    # Return result with session data
    return AgentResult(
        text=f"This is call #{call_count + 1}",
        session=session.to_dict(),  # Include updated session in response
    )

serve(my_agent)

Backward Compatibility: Functions without a session parameter still work:

# Old style (still supported)
def my_agent(messages: list, context: dict) -> AgentResult:
    return AgentResult(text="Hello!")

# New style with session
def my_agent(messages: list, context: dict, session: Session) -> AgentResult:
    return AgentResult(text="Hello!", session=session.to_dict())

Using Session in Interceptors

Request and response interceptors have access to session:

Request Interceptor

from dcaf.core import Agent, LLMRequest

def add_user_context(request: LLMRequest) -> LLMRequest:
    """Add user-specific context from session."""
    # Access session data
    user_prefs = request.session.get("user_preferences", {})
    user_name = request.session.get("user_name", "User")

    # Add context to system prompt
    request.add_system_context(f"User: {user_name}")
    if user_prefs.get("verbose"):
        request.add_system_context("User prefers detailed explanations.")

    # Track request count
    count = request.session.get("request_count", 0)
    request.session.set("request_count", count + 1)

    return request

agent = Agent(
    tools=[...],
    request_interceptors=add_user_context,
)

Response Interceptor

from dcaf.core import Agent, LLMResponse

def track_response_metrics(response: LLMResponse) -> LLMResponse:
    """Track response metrics in session."""
    # Update session with response info
    response.session.set("last_response_length", len(response.text))
    response.session.set("had_tool_calls", response.has_tool_calls())

    # Accumulate total tokens if available
    if response.usage:
        total = response.session.get("total_tokens", 0)
        total += response.usage.get("output_tokens", 0)
        response.session.set("total_tokens", total)

    return response

agent = Agent(
    tools=[...],
    response_interceptors=track_response_metrics,
)

Protocol Integration

Session data is automatically included in the HelpDesk protocol's data.session field.

Response Format

When your agent responds, session data is included:

{
  "role": "assistant",
  "content": "Added 2x Widget to cart.",
  "data": {
    "session": {
      "cart": [
        {"item": "Widget", "quantity": 2}
      ],
      "user_preference": "dark_mode"
    },
    "tool_calls": [],
    "executed_tool_calls": [...]
  }
}

Request Format

The HelpDesk sends session data back on subsequent requests:

{
  "messages": [
    {
      "role": "user",
      "content": "Add another widget",
      "data": {
        "session": {
          "cart": [
            {"item": "Widget", "quantity": 2}
          ],
          "user_preference": "dark_mode"
        }
      }
    }
  ]
}

Session Lifecycle

Turn 1:
  User: "Add 2 widgets to cart"
  Tool: add_to_cart("Widget", 2, session)
  Session After: {"cart": [{"item": "Widget", "quantity": 2}]}
  Response: "Added 2x Widget. Cart has 2 items."

Turn 2:
  Session Before: {"cart": [{"item": "Widget", "quantity": 2}]}
  User: "Add 3 gadgets"
  Tool: add_to_cart("Gadget", 3, session)
  Session After: {"cart": [..., {"item": "Gadget", "quantity": 3}]}
  Response: "Added 3x Gadget. Cart has 5 items."

Turn 3:
  Session Before: {"cart": [{...}, {...}]}
  User: "Checkout"
  Tool: checkout(session)
  Session After: {}  ← Cleared
  Response: "Order placed! 5 items shipped."

Turn 4:
  Session Before: {}
  User: "What's in my cart?"
  Tool: view_cart(session)
  Response: "Your cart is empty."

Session API Reference

Session Class

from dcaf.core import Session

Methods

Method Description
get(key, default=None, *, as_type=None) Get a value, optionally deserializing to as_type
set(key, value) Set a value (auto-serializes Pydantic/dataclass)
delete(key) Delete a value
has(key) Check if key exists
keys() Get all keys
items() Get all key-value pairs (raw data)
update(dict) Update multiple values (auto-serializes)
clear() Remove all values
to_dict() Convert to plain dict

The as_type Parameter

The get() method's as_type parameter enables typed retrieval:

from pydantic import BaseModel

class UserPrefs(BaseModel):
    theme: str = "light"

# Store a model
session.set("prefs", UserPrefs(theme="dark"))

# Retrieve as typed model
prefs = session.get("prefs", as_type=UserPrefs)  # UserPrefs instance
print(prefs.theme)  # "dark"

# Retrieve without type (raw dict)
raw = session.get("prefs")  # {"theme": "dark"}

# With default value
prefs = session.get("prefs", UserPrefs(), as_type=UserPrefs)  # Never None

Supported Types: - Pydantic models (v2) - via model_validate() - Dataclasses - via constructor - Primitives - returned as-is

Class Methods

Method Description
Session.from_dict(data) Create session from dict

Properties

Property Description
is_modified True if session was changed since creation
is_empty True if session has no data

Best Practices

1. Use Meaningful Keys

# Good - clear, namespaced
session.set("deployment_state", {...})
session.set("user_preferences", {...})

# Bad - ambiguous
session.set("state", {...})
session.set("data", {...})

2. Clean Up When Done

@tool(description="Complete the workflow")
def finish_workflow(session: Session) -> str:
    # Process the workflow...
    result = process(session.get("workflow_data"))

    # Clean up session state
    session.delete("workflow_data")
    session.delete("workflow_step")

    return result

3. Use Defaults for Safety

# Always provide defaults for optional data
cart = session.get("cart", [])  # Empty list if not set
count = session.get("attempt_count", 0)  # Zero if not set

# With typed models, provide a default instance
prefs = session.get("prefs", UserPrefs(), as_type=UserPrefs)

4. Use Typed Models for Complex Data

# Good - typed model with validation
class DeploymentState(BaseModel):
    environment: str
    services: list[str] = []
    step: int = 0

state = session.get("deployment", as_type=DeploymentState)
if state:
    # IDE autocomplete, type checking, validation
    print(state.environment)
    print(state.step)

# Less ideal - raw dicts require manual key access
state = session.get("deployment")
if state:
    # No autocomplete, prone to typos
    print(state.get("enviroment"))  # Typo goes unnoticed!

4. Keep Session Data Serializable

Session data must be JSON-serializable:

# Good - JSON-serializable types
session.set("items", ["a", "b", "c"])
session.set("config", {"key": "value"})
session.set("count", 42)

# Bad - non-serializable types
session.set("connection", db_connection)  # Will fail
session.set("callback", lambda x: x)  # Will fail

5. Don't Store Sensitive Data

Session data is included in protocol messages. Avoid storing:

  • Passwords or tokens
  • API keys
  • Personal identifiable information (PII)
# Bad - sensitive data in session
session.set("api_key", "sk-secret-key")

# Good - store only references
session.set("user_id", "12345")  # Look up details server-side