Sessions

Sessions enable multi-turn conversations by maintaining conversation history and context across multiple agent invocations. They provide a lightweight mechanism for stateful interactions without requiring external storage.

What is a Session?

A session is a container for conversation state that includes:

Session IDs for identifying and tracking conversations
Conversation history (stored messages)
Mutable state for context providers
Provider-managed data (memory, context, metadata)

Python
.NET

In Python, AgentSession is a lightweight state container:

class AgentSession:
    session_id: str              # Unique session identifier
    service_session_id: str | None  # Service-managed ID (if using service storage)
    state: dict[str, Any]        # Mutable state shared with providers

Sessions are created per agent and can be serialized for persistence.

In .NET, AgentSession is an abstract class managed by agents:

public abstract class AgentSession
{
    // Session-specific implementation
}

Each agent type provides its own session implementation.

Creating and Using Sessions

Basic Session Usage

Python
.NET

from agent_framework.azure import AzureOpenAIResponsesClient
import os

client = AzureOpenAIResponsesClient(
    project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
    deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
)

agent = client.as_agent(
    name="ConversationAgent",
    instructions="You are a friendly assistant. Keep your answers brief.",
)

# Create a session
session = agent.create_session()

# Multi-turn conversation
response1 = await agent.run("My name is Alice and I love hiking.", session=session)
print(f"Agent: {response1.text}")

# Agent remembers previous context
response2 = await agent.run("What do you remember about me?", session=session)
print(f"Agent: {response2.text}")
# Output: "You said your name is Alice and you love hiking."

Key points:

Call agent.create_session() to create a new session
Pass the same session object to all related run() calls
The agent automatically manages conversation history

var agent = client.GetChatClient(deploymentName).AsAIAgent(
    name: "ConversationAgent",
    instructions: "You are a friendly assistant. Keep your answers brief.");

// Create a session
var session = await agent.CreateSessionAsync();

// Multi-turn conversation
var response1 = await agent.RunAsync(
    "My name is Alice and I love hiking.",
    session: session);
Console.WriteLine($"Agent: {response1.Messages.LastOrDefault()?.Text}");

// Agent remembers previous context
var response2 = await agent.RunAsync(
    "What do you remember about me?",
    session: session);
Console.WriteLine($"Agent: {response2.Messages.LastOrDefault()?.Text}");

Session with Custom ID

Python
.NET

# Create session with specific ID (useful for resuming)
session = agent.create_session(session_id="user-123-conversation-456")

# Or get/create session for a service-managed ID
session = agent.get_session(
    service_session_id="azure-thread-abc123",
    session_id="local-session-789",
)

// Create session (ID is managed by the agent implementation)
var session = await agent.CreateSessionAsync();

// Access session ID
Console.WriteLine($"Session ID: {session.Id}");

Conversation History Management

Automatic History Storage

Python
.NET

When you provide a session without configuring storage, the framework automatically adds an in-memory history provider:

agent = client.as_agent(name="Assistant")
session = agent.create_session()

# First turn
await agent.run("Hello, I'm learning Python", session=session)

# Second turn - history is automatically maintained
await agent.run("Can you recommend some resources?", session=session)

# The agent sees both messages in the conversation

The InMemoryHistoryProvider is added automatically when:

A session is provided
No context providers are configured
Service-side storage is not requested (no service_session_id)

var agent = client.GetChatClient(deploymentName).AsAIAgent(name: "Assistant");
var session = await agent.CreateSessionAsync();

// First turn
await agent.RunAsync("Hello, I'm learning C#", session: session);

// Second turn - history is automatically maintained
await agent.RunAsync("Can you recommend some resources?", session: session);

Custom History Providers

Python
.NET

Implement custom history storage by extending BaseHistoryProvider:

from agent_framework import BaseHistoryProvider, Message

class DatabaseHistoryProvider(BaseHistoryProvider):
    def __init__(self, db_connection):
        super().__init__(source_id="database")
        self.db = db_connection

    async def get_messages(self, session_id: str | None, **kwargs) -> list[Message]:
        """Load messages from database."""
        if not session_id:
            return []
        rows = await self.db.query(
            "SELECT role, content FROM messages WHERE session_id = ?",
            session_id
        )
        return [Message(role=row['role'], text=row['content']) for row in rows]

    async def save_messages(
        self,
        session_id: str | None,
        messages: Sequence[Message],
        **kwargs
    ) -> None:
        """Save messages to database."""
        if not session_id:
            return
        for msg in messages:
            await self.db.execute(
                "INSERT INTO messages (session_id, role, content) VALUES (?, ?, ?)",
                session_id, msg.role, msg.text
            )

# Use custom provider
agent = Agent(
    client=client,
    name="Assistant",
    context_providers=[DatabaseHistoryProvider(db)],
)

Configure what gets stored:

history_provider = DatabaseHistoryProvider(
    db,
    load_messages=True,         # Load history before runs
    store_inputs=True,          # Store user messages
    store_outputs=True,         # Store agent responses
    store_context_messages=False,  # Don't store provider context
)

Custom history is typically managed through AIContextProvider:

// Context providers manage history and state
// Implementation varies by agent type

Session State

Python
.NET

Sessions maintain a mutable state dictionary for provider data:

session = agent.create_session()

# Providers store data in session.state
session.state["user_preferences"] = {"theme": "dark", "language": "en"}
session.state["conversation_count"] = 0

# Each provider gets its own state namespace
# session.state = {
#     "in_memory": {"messages": [...]},  # InMemoryHistoryProvider
#     "rag": {"documents": [...]},        # RAG provider
#     "user_preferences": {...},           # Your custom data
# }

Serialization

Sessions can be serialized for persistence:

# Serialize session
session_dict = session.to_dict()
# Store in database, file, etc.
json.dump(session_dict, open("session.json", "w"))

# Deserialize session
session_dict = json.load(open("session.json"))
restored_session = AgentSession.from_dict(session_dict)

# Continue conversation
await agent.run("What were we discussing?", session=restored_session)

Custom types in state are automatically handled if they implement to_dict()/from_dict() or are Pydantic models.

Session state is managed through the session implementation:

// State management varies by agent/session type
var session = await agent.CreateSessionAsync();

// Serialize session (if supported)
var serialized = await agent.SerializeSessionAsync(session);

// Deserialize session
var restored = await agent.DeserializeSessionAsync(serialized);

Context Providers

Context providers add context to agent invocations through the session:

Python
.NET

from agent_framework import BaseContextProvider, SessionContext

class UserContextProvider(BaseContextProvider):
    """Add user profile to conversations."""

    def __init__(self, user_service):
        super().__init__(source_id="user_context")
        self.user_service = user_service

    async def before_run(
        self,
        *,
        agent,
        session,
        context: SessionContext,
        state: dict,
    ) -> None:
        """Add user context before agent runs."""
        user_id = context.options.get("user_id")
        if user_id:
            user_profile = await self.user_service.get_user(user_id)
            # Add context as instructions
            context.extend_instructions(
                self.source_id,
                f"User profile: {user_profile.name}, preferences: {user_profile.preferences}"
            )

# Use provider
agent = Agent(
    client=client,
    name="PersonalizedAgent",
    context_providers=[UserContextProvider(user_service)],
)

# Context is automatically added
result = await agent.run(
    "What should I watch tonight?",
    session=session,
    user_id="user-123",
)

Provider Lifecycle

Context providers have two hooks:

class MyProvider(BaseContextProvider):
    async def before_run(self, *, agent, session, context, state):
        """Called before agent invocation.
        
        Add context by calling:
        - context.extend_messages(self, messages)
        - context.extend_instructions(self.source_id, instructions)
        - context.extend_tools(self.source_id, tools)
        """
        pass

    async def after_run(self, *, agent, session, context, state):
        """Called after agent invocation.
        
        Access response via context.response
        Store data in state dict
        """
        pass

// Context providers in .NET implement AIContextProvider
// Example: ChatHistoryContextProvider for storing chat history

Service-Managed Sessions

Python
.NET

Some services (like Azure AI Foundry) manage sessions server-side:

# Service creates and manages the session
session = agent.get_session(
    service_session_id="azure-thread-abc123"
)

# Or pass service session ID in options
result = await agent.run(
    "Hello",
    options={"conversation_id": "azure-thread-abc123"},
)

# Service maintains history; no local storage needed

The framework automatically:

Skips local history providers when service_session_id is set
Updates session.service_session_id from service responses
Propagates conversation IDs between calls

// Service-managed sessions are handled automatically
var session = await agent.CreateSessionAsync();

// The session ID is managed by the underlying service
var response = await agent.RunAsync("Hello", session: session);

Session Streaming

Python
.NET

Sessions work with both streaming and non-streaming modes:

session = agent.create_session()

# Streaming mode
async for chunk in agent.run("Tell me a story", session=session, stream=True):
    print(chunk.text, end="", flush=True)

# History is automatically updated after stream completes

# Next turn remembers the story
result = await agent.run("What happened next?", session=session)

var session = await agent.CreateSessionAsync();

// Streaming mode
await foreach (var chunk in agent.RunStreamingAsync("Tell me a story", session: session))
{
    Console.Write(chunk.Text);
}

// Next turn remembers the story
var response = await agent.RunAsync("What happened next?", session: session);

Complete Example: Multi-Agent Session

Python
.NET

from agent_framework import Agent, tool
from agent_framework.azure import AzureOpenAIResponsesClient
import os

@tool(approval_mode="never_require")
def search_products(query: str) -> str:
    """Search product catalog."""
    return f"Found 3 products matching '{query}'"

@tool(approval_mode="never_require")
def check_inventory(product_id: str) -> str:
    """Check product inventory."""
    return f"Product {product_id} has 42 units in stock"

async def main():
    client = AzureOpenAIResponsesClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
    )

    # Create agent with tools
    agent = client.as_agent(
        name="ShoppingAgent",
        instructions="You help users shop for products. Be helpful and concise.",
        tools=[search_products, check_inventory],
    )

    # Start conversation
    session = agent.create_session()

    print("Shopping Assistant (type 'exit' to quit)\n")

    while True:
        user_input = input("You: ")
        if user_input.lower() in ["exit", "quit"]:
            break

        print("Agent: ", end="", flush=True)
        async for chunk in agent.run(user_input, session=session, stream=True):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print("\n")

    # Save session for later
    session_data = session.to_dict()
    # Store session_data in database...

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

[Description("Search product catalog.")]
static string SearchProducts([Description("Search query")] string query)
{
    return $"Found 3 products matching '{query}'";
}

[Description("Check product inventory.")]
static string CheckInventory([Description("Product ID")] string productId)
{
    return $"Product {productId} has 42 units in stock";
}

var agent = client.GetChatClient(deploymentName).AsAIAgent(
    name: "ShoppingAgent",
    instructions: "You help users shop for products. Be helpful and concise.",
    tools: [
        AIFunctionFactory.Create(SearchProducts),
        AIFunctionFactory.Create(CheckInventory)
    ]);

var session = await agent.CreateSessionAsync();

Console.WriteLine("Shopping Assistant (type 'exit' to quit)\n");

while (true)
{
    Console.Write("You: ");
    var userInput = Console.ReadLine();
    if (string.IsNullOrWhiteSpace(userInput) || 
        userInput.Equals("exit", StringComparison.OrdinalIgnoreCase))
        break;

    Console.Write("Agent: ");
    await foreach (var chunk in agent.RunStreamingAsync(userInput, session: session))
    {
        Console.Write(chunk.Text);
    }
    Console.WriteLine("\n");
}

// Serialize session
var sessionData = await agent.SerializeSessionAsync(session);
// Store sessionData...

Best Practices

Session Management Tips

Session Scope: Create one session per conversation/user
Cleanup: Clear sessions when conversations end
Serialization: Persist sessions for long-running conversations
State Size: Keep session state small; offload large data to external storage
Provider Order: Providers run in order; plan dependencies carefully
Testing: Test session behavior with multi-turn scenarios

Production Considerations

Expiration: Implement session timeouts to free memory
Storage: Use durable storage (database/cache) for production sessions
Scaling: Sessions should be stateless externally for horizontal scaling
Security: Validate session ownership before resuming conversations
Limits: Set maximum session size to prevent memory exhaustion
Monitoring: Track active sessions and memory usage

Next Steps

Agents

Learn how agents interact with sessions

Middleware

Modify session behavior with middleware

Tools

Tools can access session state

Observability

Monitor session lifecycle with telemetry

Get Started

Core Concepts

Workflows

Providers

Hosting & Deployment

Migration Guides

What is a Session?

Creating and Using Sessions

Basic Session Usage

Session with Custom ID

Conversation History Management

Automatic History Storage

Custom History Providers

Session State

Serialization

Context Providers

Provider Lifecycle

Service-Managed Sessions

Session Streaming

Complete Example: Multi-Agent Session

Best Practices

Next Steps

Agents

Middleware

Tools

Observability

​What is a Session?

​Creating and Using Sessions

​Basic Session Usage

​Session with Custom ID

​Conversation History Management

​Automatic History Storage

​Custom History Providers

​Session State

​Serialization

​Context Providers

​Provider Lifecycle

​Service-Managed Sessions

​Session Streaming

​Complete Example: Multi-Agent Session

​Best Practices

​Next Steps

Agents

Middleware

Tools

Observability

What is a Session?

Creating and Using Sessions

Basic Session Usage

Session with Custom ID

Conversation History Management

Automatic History Storage

Custom History Providers

Session State

Serialization

Context Providers

Provider Lifecycle

Service-Managed Sessions

Session Streaming

Complete Example: Multi-Agent Session

Best Practices

Next Steps