Skip to main content

​
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"
)
​
session_id
str | None
Unique identifier for this session. If None, a UUID will be generated.
​
service_session_id
str | None
Service-managed session ID (e.g., thread ID from Assistants API). If using service-side storage.

​
Properties

​
session_id
str
The unique identifier for this session.
​
service_session_id
str | None
Service-managed session ID. Can be updated after creation.
​
state
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()
​
return
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)
​
data
dict[str, Any]
required
Dictionary from a previous to_dict() call.
​
return
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

​
session_id
str | None
The ID of the current session.
​
service_session_id
str | None
Service-managed session ID.
​
input_messages
list[Message]
The new messages being sent to the agent (set by caller).
​
context_messages
dict[str, list[Message]]
Dictionary mapping source_id -> messages added by that provider. Maintains insertion order (provider execution order).
​
instructions
list[str]
Additional instructions added by providers.
​
tools
list[Any]
Additional tools added by providers.
​
response
AgentResponse | None
After invocation, contains the full AgentResponse. Read-only for providers.
​
options
dict[str, Any]
Options passed to agent.run() - read-only, for reflection only.
​
metadata
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")]
)
​
source
str | object
required
Either a plain source_id string, or an object with a source_id attribute (e.g., a context provider).
​
messages
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: ..."
)
​
source_id
str
required
The provider source_id adding these instructions.
​
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]
)
​
source_id
str
required
The provider source_id adding these tools.
​
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
)
​
sources
set[str] | None
If provided, only include context messages from these sources.
​
exclude_sources
set[str] | None
If provided, exclude context messages from these sources.
​
include_input
bool
default:"False"
If True, append input_messages after context.
​
include_response
bool
default:"False"
If True, append response.messages at the end.
​
return
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()
​
source_id
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])
​
agent
SupportsAgentRun
required
The agent running this invocation.
​
session
AgentSession
required
The current session.
​
context
SessionContext
required
The invocation context - add messages/instructions/tools here.
​
state
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
​
agent
SupportsAgentRun
required
The agent that ran this invocation.
​
session
AgentSession
required
The current session.
​
context
SessionContext
required
The invocation context with response populated.
​
state
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
)
​
source_id
str
required
Unique identifier for this provider instance.
​
load_messages
bool
default:"True"
Whether to load messages before invocation. When False, the agent skips calling before_run entirely.
​
store_inputs
bool
default:"True"
Whether to store input messages.
​
store_context_messages
bool
default:"False"
Whether to store context from other providers.
​
store_context_from
set[str] | None
If set, only store context from these source_ids.
​
store_outputs
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
​
session_id
str | None
required
The session ID to retrieve messages for.
​
**kwargs
Any
Additional arguments (e.g., state for in-memory providers).
​
return
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
​
session_id
str | None
required
The session ID to store messages for.
​
messages
Sequence[Message]
required
The messages to persist.
​
**kwargs
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
)
​
source_id
str | None
default:"in_memory"
Unique identifier for this provider instance.
​
load_messages
bool
default:"True"
Whether to load messages before invocation.
​
store_inputs
bool
default:"True"
Whether to store input messages.
​
store_context_messages
bool
default:"False"
Whether to store context from other providers.
​
store_context_from
set[str] | None
If set, only store context from these source_ids.
​
store_outputs
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()

Register a type for automatic deserialization in session state. 
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)
​
cls
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

Build docs developers (and LLMs) love

Get started for freeTalk to us