When building conversational AI agents, one of the key requirements is maintaining context across multiple interactions. While agent functions are essentially stateless by design, the BeeAI platform provides built-in mechanisms to access and manage conversation history. The platform automatically stores all conversation messages in context memory, allowing your agent to retrieve previous interactions when needed.
The stateless design of agent functions ensures reliability and scalability, while the context memory system provides the conversation continuity that users expect from AI assistants.

​
Quickstart

1

Access conversation history

Use the RunContext parameter to access the conversation store and load previous messages.
2

Filter and process history

Retrieve and filter the conversation history to get the messages relevant to your agent’s logic.
3

Configure persistent storage

Set up persistent context storage to maintain conversation history across agent restarts.

​
Basic History Access

Here’s how to access conversation history in your agent: 
import os

from a2a.types import Message
from a2a.utils.message import get_message_text
from beeai_sdk.server import Server
from beeai_sdk.server.context import RunContext
from beeai_sdk.a2a.types import AgentMessage

server = Server()

@server.agent()
async def example_agent(input: Message, context: RunContext):
    """Agent that demonstrates conversation history access"""
    
    # Get the current user message
    current_message = get_message_text(input)
    print(f"Current message: {current_message}")
    
    # Load all messages from conversation history (including current message)
    history = [
        message async for message in context.store.load_history() 
        if isinstance(message, Message) and message.parts
    ]
    
    # Process the conversation history
    print(f"Found {len(history)} messages in conversation (including current)")
    
    # Your agent logic here - you can now reference all messages in the conversation
    yield AgentMessage(text=f"Hello! I can see we have {len(history)} messages in our conversation.")

def run():
    server.run(host=os.getenv("HOST", "127.0.0.1"), port=int(os.getenv("PORT", 8000)))

if __name__ == "__main__":
    run()

​
Understanding Conversation History

The context.store.load_history() method returns an async iterator containing all items in the conversation, including the current message. This can include:
  • A2A Messages: Both user and assistant messages from the conversation (including the current message)
  • Artifacts: Any files, documents, or other artifacts shared during the conversation
For multi-turn conversations, you’ll primarily work with A2A messages, which include:
  • User messages: Messages sent by the user
  • Assistant messages: Previous responses from your agent
The history includes the current message, so if you want only previous messages, you may need to filter out the last message or use the current message separately.
The history iterator returns all message types. Always filter messages using isinstance(message, Message) to ensure you’re working with the correct message format.

​
Persistent Storage

By default, conversation history is stored in memory, which means it’s lost when the agent process restarts. For production applications, you’ll want to use persistent storage.

​
Using Platform Context Store

To maintain conversation history across agent restarts, configure your server to use the platform’s persistent context store: 
from beeai_sdk.server.context import PlatformContextStore

def run():
    server.run(
        host=os.getenv("HOST", "127.0.0.1"), 
        port=int(os.getenv("PORT", 8000)),
        context_store=PlatformContextStore()
    )
The PlatformContextStore automatically handles conversation persistence, ensuring that users can continue their conversations even after agent restarts or deployments.

​
Advanced History Usage with BeeAI Framework

Here’s a sophisticated example using the BeeAI Framework to build a multi-turn chat agent that leverages conversation history and LLM capabilities: 
import os
from typing import Annotated

from a2a.types import Message, Role
from a2a.utils.message import get_message_text
from beeai_framework.backend import AssistantMessage, UserMessage
from beeai_sdk.server import Server
from beeai_sdk.server.context import RunContext
from beeai_sdk.server.store.platform_context_store import PlatformContextStore
from beeai_sdk.a2a.extensions import (
    LLMServiceExtensionServer,
    LLMServiceExtensionSpec,
)
from beeai_framework.adapters.openai import OpenAIChatModel
from beeai_framework.agents.experimental import RequirementAgent
from beeai_framework.agents.experimental.requirements.conditional import ConditionalRequirement
from beeai_framework.tools.think import ThinkTool
from beeai_framework.backend.types import ChatModelParameters

server = Server()

FrameworkMessage = UserMessage | AssistantMessage


def to_framework_message(message: Message) -> FrameworkMessage:
    """Convert A2A Message to BeeAI Framework Message format"""
    message_text = "".join(part.root.text for part in message.parts if part.root.kind == "text")

    if message.role == Role.agent:
        return AssistantMessage(message_text)
    elif message.role == Role.user:
        return UserMessage(message_text)
    else:
        raise ValueError(f"Invalid message role: {message.role}")


@server.agent()
async def multi_turn_chat_agent(
    input: Message,
    context: RunContext,
    llm: Annotated[LLMServiceExtensionServer, LLMServiceExtensionSpec.single_demand()],
):
    """Multi-turn chat agent with conversation memory and LLM integration"""
    
    # Load conversation history
    history = [
        message async for message in context.store.load_history() 
        if isinstance(message, Message) and message.parts
    ]

    # Get LLM configuration from the platform
    llm_config = llm.data.llm_fulfillments.get("default")

    # Initialize BeeAI Framework LLM client
    llm_client = OpenAIChatModel(
        model_id=llm_config.api_model,
        base_url=llm_config.api_base,
        api_key=llm_config.api_key,
        parameters=ChatModelParameters(temperature=0.0),
        tool_choice_support=set(),
    )

    # Create a RequirementAgent with conversation memory
    agent = RequirementAgent(
        name="Agent",
        llm=llm_client,
        role="helpful assistant",
        instructions="You are a helpful assistant that is supposed to remember users name. Ask them for their name and remember it.",
        tools=[ThinkTool()],
        requirements=[ConditionalRequirement(ThinkTool, force_at_step=1)],
        save_intermediate_steps=False,
        middlewares=[],
    )
    
    # Load conversation history into agent memory
    await agent.memory.add_many(to_framework_message(item) for item in history)

    # Process the current message and generate response
    async for event, meta in agent.run(get_message_text(input)):
        if meta.name == "success" and event.state.steps:
            step = event.state.steps[-1]
            if not step.tool:
                continue

            tool_name = step.tool.name

            if tool_name == "final_answer":
                yield step.input["response"]


def run():
    server.run(
        host=os.getenv("HOST", "127.0.0.1"), 
        port=int(os.getenv("PORT", "8000")), 
        context_store=PlatformContextStore()  # Enable persistent storage
    )


if __name__ == "__main__":
    run()
This advanced example demonstrates several key concepts:
  • LLM Integration: Uses the platform’s LLM service extension to get model access
  • Framework Integration: Leverages the BeeAI Framework for sophisticated agent capabilities
  • Memory Management: Converts conversation history to framework format and loads it into agent memory
  • Tool Usage: Includes thinking tools and conditional requirements for better reasoning
  • Persistent Storage: Uses PlatformContextStore for conversation persistence