Sessions API Reference

Session management and context provider system for conversation state and context engineering.

AgentSession

A lightweight conversation session with an agent.

from agent_framework import AgentSession

Constructor

session = AgentSession(
    session_id="session-123",
    service_session_id="thread-abc"
)

str | None

Unique identifier for this session. If None, a UUID will be generated.

str | None

Service-managed session ID (e.g., thread ID from Assistants API). If using service-side storage.

Properties

str

The unique identifier for this session.

str | None

Service-managed session ID. Can be updated after creation.

dict[str, Any]

Mutable state dictionary shared with all providers. Used to store conversation history, context, and provider-specific data.

Methods

to_dict()

Serialize session to a plain dict for storage/transfer.

data = session.to_dict()

dict[str, Any]

Serialized session dictionary. Values in state that implement to_dict()/from_dict() are automatically serialized.

from_dict() (classmethod)

Restore session from a previously serialized dict.

session = AgentSession.from_dict(data)

dict[str, Any]

required

Dictionary from a previous to_dict() call.

AgentSession

Restored AgentSession instance.

Example: Basic Session Usage

from agent_framework import Agent, AgentSession

# Create a session
session = AgentSession()

# Use with agent runs
agent = Agent(client=client, name="assistant")
response1 = await agent.run("Hello", session=session)
response2 = await agent.run("How are you?", session=session)

# Serialize for storage
data = session.to_dict()
# Store in database...

# Later, restore from storage
restored_session = AgentSession.from_dict(data)
response3 = await agent.run("Continue conversation", session=restored_session)

SessionContext

Per-invocation state passed through the context provider pipeline.

from agent_framework import SessionContext

Attributes

str | None

The ID of the current session.

str | None

Service-managed session ID.

list[Message]

The new messages being sent to the agent (set by caller).

dict[str, list[Message]]

Dictionary mapping source_id -> messages added by that provider. Maintains insertion order (provider execution order).

list[str]

Additional instructions added by providers.

list[Any]

Additional tools added by providers.

AgentResponse | None

After invocation, contains the full AgentResponse. Read-only for providers.

dict[str, Any]

Options passed to agent.run() - read-only, for reflection only.

dict[str, Any]

Shared metadata dictionary for cross-provider communication.

Methods

extend_messages()

Add context messages from a specific source.

context.extend_messages(
    source=provider,  # or source_id string
    messages=[Message(role="system", text="Context info")]
)

str | object

required

Either a plain source_id string, or an object with a source_id attribute (e.g., a context provider).

Sequence[Message]

required

The messages to add. Messages are copied before attribution is added.

extend_instructions()

Add instructions to be prepended to the conversation.

context.extend_instructions(
    source_id="rag_provider",
    instructions="Use the following context: ..."
)

str

required

The provider source_id adding these instructions.

str | Sequence[str]

required

A single instruction string or sequence of strings.

extend_tools()

Add tools to be available for this invocation.

context.extend_tools(
    source_id="tool_provider",
    tools=[search_tool, calculator_tool]
)

str

required

The provider source_id adding these tools.

Sequence[Any]

required

The tools to add. Tools are attributed with source metadata.

get_messages()

Get context messages, optionally filtered and including input/response.

all_messages = context.get_messages(
    sources={"history", "rag"},
    exclude_sources={"temp"},
    include_input=True,
    include_response=True
)

set[str] | None

If provided, only include context messages from these sources.

set[str] | None

If provided, exclude context messages from these sources.

bool

default:"False"

If True, append input_messages after context.

bool

default:"False"

If True, append response.messages at the end.

list[Message]

Flattened list of messages in conversation order.

BaseContextProvider

Base class for context providers that participate in the context engineering pipeline.

from agent_framework import BaseContextProvider

Constructor

class MyContextProvider(BaseContextProvider):
    def __init__(self):
        super().__init__(source_id="my_provider")
    
    async def before_run(self, *, agent, session, context, state):
        # Add context before model invocation
        context.extend_messages(self, [...messages...])
    
    async def after_run(self, *, agent, session, context, state):
        # Process response after model invocation
        pass

provider = MyContextProvider()

str

required

Unique identifier for this provider instance. Used for message/tool attribution.

Methods

before_run()

Called before model invocation.

async def before_run(
    self,
    *,
    agent: SupportsAgentRun,
    session: AgentSession,
    context: SessionContext,
    state: dict[str, Any]
) -> None:
    """Add context before model invocation."""
    context.extend_messages(self, messages)
    context.extend_instructions(self.source_id, "Custom instructions")
    context.extend_tools(self.source_id, [tool])

SupportsAgentRun

required

The agent running this invocation.

AgentSession

required

The current session.

SessionContext

required

The invocation context - add messages/instructions/tools here.

dict[str, Any]

required

The provider-scoped mutable state dict. This is a reference to session.state[provider.source_id].

after_run()

Called after model invocation.

async def after_run(
    self,
    *,
    agent: SupportsAgentRun,
    session: AgentSession,
    context: SessionContext,
    state: dict[str, Any]
) -> None:
    """Process response after model invocation."""
    # Access the response
    if context.response:
        # Store messages, extract info, etc.
        pass

SupportsAgentRun

required

The agent that ran this invocation.

AgentSession

required

The current session.

SessionContext

required

The invocation context with response populated.

dict[str, Any]

required

The provider-scoped mutable state dict.

Example: RAG Context Provider

from agent_framework import BaseContextProvider, Agent, Message

class RAGProvider(BaseContextProvider):
    def __init__(self, retriever):
        super().__init__(source_id="rag")
        self.retriever = retriever
    
    async def before_run(self, *, agent, session, context, state):
        # Get user query from input messages
        query = context.input_messages[-1].text if context.input_messages else ""
        
        # Retrieve relevant documents
        docs = await self.retriever.retrieve(query)
        
        # Add as context messages
        context_msgs = [
            Message(role="system", text=f"Context: {doc}")
            for doc in docs
        ]
        context.extend_messages(self, context_msgs)

# Use with agent
agent = Agent(
    client=client,
    name="rag-assistant",
    context_providers=[RAGProvider(my_retriever)]
)

BaseHistoryProvider

Base class for conversation history storage providers.

from agent_framework import BaseHistoryProvider

Constructor

class MyHistoryProvider(BaseHistoryProvider):
    async def get_messages(self, session_id, **kwargs):
        # Load messages from storage
        return messages
    
    async def save_messages(self, session_id, messages, **kwargs):
        # Save messages to storage
        pass

provider = MyHistoryProvider(
    source_id="my_history",
    load_messages=True,
    store_inputs=True,
    store_outputs=True
)

str

required

Unique identifier for this provider instance.

bool

default:"True"

Whether to load messages before invocation. When False, the agent skips calling before_run entirely.

bool

default:"True"

Whether to store input messages.

bool

default:"False"

Whether to store context from other providers.

set[str] | None

If set, only store context from these source_ids.

bool

default:"True"

Whether to store response messages.

Abstract Methods

get_messages()

Retrieve stored messages for this session.

async def get_messages(
    self,
    session_id: str | None,
    **kwargs: Any
) -> list[Message]:
    """Load messages from storage."""
    # Implementation
    return messages

str | None

required

The session ID to retrieve messages for.

Any

Additional arguments (e.g., state for in-memory providers).

list[Message]

List of stored messages.

save_messages()

Persist messages for this session.

async def save_messages(
    self,
    session_id: str | None,
    messages: Sequence[Message],
    **kwargs: Any
) -> None:
    """Save messages to storage."""
    # Implementation

str | None

required

The session ID to store messages for.

Sequence[Message]

required

The messages to persist.

Any

Additional arguments.

Example: Database History Provider

from agent_framework import BaseHistoryProvider, Message

class DatabaseHistoryProvider(BaseHistoryProvider):
    def __init__(self, db_connection):
        super().__init__(source_id="db_history")
        self.db = db_connection
    
    async def get_messages(self, session_id, **kwargs):
        if not session_id:
            return []
        rows = await self.db.query(
            "SELECT * FROM messages WHERE session_id = ?",
            session_id
        )
        return [Message.from_dict(row) for row in rows]
    
    async def save_messages(self, session_id, messages, **kwargs):
        if not session_id:
            return
        for msg in messages:
            await self.db.execute(
                "INSERT INTO messages (session_id, data) VALUES (?, ?)",
                session_id, msg.to_dict()
            )

# Use with agent
agent = Agent(
    client=client,
    name="persistent-agent",
    context_providers=[DatabaseHistoryProvider(db)]
)

InMemoryHistoryProvider

Built-in history provider that stores messages in session.state.

from agent_framework import InMemoryHistoryProvider

Constructor

provider = InMemoryHistoryProvider(
    source_id="in_memory",  # Optional, defaults to "in_memory"
    load_messages=True,
    store_inputs=True,
    store_outputs=True
)

str | None

default:"in_memory"

Unique identifier for this provider instance.

bool

default:"True"

Whether to load messages before invocation.

bool

default:"True"

Whether to store input messages.

bool

default:"False"

Whether to store context from other providers.

set[str] | None

If set, only store context from these source_ids.

bool

default:"True"

Whether to store response messages.

Notes

Messages are stored in state["messages"] as a list of Message objects
Serialization to/from dicts is handled by AgentSession.to_dict()/from_dict()
This provider holds no instance state - all data lives in the session’s state dict
Auto-added by the agent for local sessions when no providers are configured and service-side storage is not requested

Example

from agent_framework import Agent, AgentSession, InMemoryHistoryProvider

# Explicitly add the provider
agent = Agent(
    client=client,
    name="assistant",
    context_providers=[InMemoryHistoryProvider()]
)

# Or let the agent auto-add it
session = AgentSession()
agent = Agent(client=client, name="assistant")
response = await agent.run("Hello", session=session)
# InMemoryHistoryProvider is automatically added

# Serialize session with history
data = session.to_dict()
print(data["state"]["messages"])  # Messages are here

register_state_type()

from agent_framework import register_state_type

Usage

from pydantic import BaseModel
from agent_framework import register_state_type, AgentSession

class MyModel(BaseModel):
    value: str

# Register for serialization round-trips
register_state_type(MyModel)

# Use in session state
session = AgentSession()
session.state["my_data"] = MyModel(value="test")

# Serialize and restore
data = session.to_dict()
restored = AgentSession.from_dict(data)
assert isinstance(restored.state["my_data"], MyModel)

type

required

The type to register. Types with to_dict/from_dict methods or Pydantic BaseModel subclasses are handled automatically.

Notes

Pydantic models are auto-registered on first serialization
Pre-registering ensures deserialization works even if the model hasn’t been serialized in this process yet
The type identifier defaults to cls.__name__.lower() but can be overridden by defining a _get_type_identifier classmethod

Getting Started

Guides

API Reference

Sessions

Sessions API Reference

AgentSession

Constructor

Properties

Methods

to_dict()

from_dict() (classmethod)

Example: Basic Session Usage

SessionContext

Attributes

Methods

extend_messages()

extend_instructions()

extend_tools()

get_messages()

BaseContextProvider

Constructor

Methods

before_run()

after_run()

Example: RAG Context Provider

BaseHistoryProvider

Constructor

Abstract Methods

get_messages()

save_messages()

Example: Database History Provider

InMemoryHistoryProvider

Constructor

Notes

Example

register_state_type()

Usage

Notes

​Sessions API Reference

​AgentSession

​Constructor

​Properties

​Methods

​to_dict()

​from_dict() (classmethod)

​Example: Basic Session Usage

​SessionContext

​Attributes

​Methods

​extend_messages()

​extend_instructions()

​extend_tools()

​get_messages()

​BaseContextProvider

​Constructor

​Methods

​before_run()

​after_run()

​Example: RAG Context Provider

​BaseHistoryProvider

​Constructor

​Abstract Methods

​get_messages()

​save_messages()

​Example: Database History Provider

​InMemoryHistoryProvider

​Constructor

​Notes

​Example

​register_state_type()

​Usage

​Notes

Sessions API Reference

AgentSession

Constructor

Properties

Methods

to_dict()

from_dict() (classmethod)

Example: Basic Session Usage

SessionContext

Attributes

Methods

extend_messages()

extend_instructions()

extend_tools()

get_messages()

BaseContextProvider

Constructor

Methods

before_run()

after_run()

Example: RAG Context Provider

BaseHistoryProvider

Constructor

Abstract Methods

get_messages()

save_messages()

Example: Database History Provider

InMemoryHistoryProvider

Constructor

Notes

Example

register_state_type()

Usage

Notes