Skip to main content

Documentation Index

Fetch the complete documentation index at: https://framework.beeai.dev/llms.txt

Use this file to discover all available pages before exploring further.

Agent Stack is an open platform to help you discover, run, and compose AI agents from any framework. This tutorial demonstrates how to consume agents from the Agent Stack and expose agents built in BeeAI Framework to the Agent Stack. Prerequisites
  • Agent Stack installed and running locally
  • BeeAI Framework installed with pip install beeai-framework[agentstack]

​
Consuming from the platform (client)

The AgentStackAgent class allows you to connect to any agent hosted on the Agent Stack. This means that you can interact with agents built from any framework! Use AgentStackAgent when:
  • You’re connecting specifically to the Agent Stack services.
  • You want forward compatibility for the Agent Stack, no matter which protocol it is based on.
Here’s a simple example that uses the built-in chat agent: 
import asyncio
import sys
import traceback

from beeai_framework.adapters.agentstack.agents import AgentStackAgent
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import UnconstrainedMemory
from examples.helpers.io import ConsoleReader


async def main() -> None:
    reader = ConsoleReader()

    agents = await AgentStackAgent.from_agent_stack(url="http://127.0.0.1:8333", memory=UnconstrainedMemory())
    if len(agents) > 1:
        reader.write("Prompt: ", "Select one of the available agents:\n")
        while True:
            for index, agent in enumerate(agents):
                reader.write("AgentStack: ", f"{index}) {agent.name} - {agent.meta.description}")

            agents_index = reader.ask_single_question("Write agent's number: ")
            try:
                agent = agents[int(agents_index)]
                if agent:
                    break

            except (ValueError, IndexError):
                reader.write(
                    "AgentStack (error) : ",
                    f"Invalid selection: `{agents_index}`. Please enter a valid agent number.\n",
                )
    elif len(agents) == 1:
        agent = agents[0]
    else:
        reader.write("AgentStack (error) : ", "No agent registered within the agent stack.\n")
        exit(0)

    reader.write("AgentStack: ", f"Selected {agent.name}:\n")
    for prompt in reader:
        # Run the agent and observe events
        response = await agent.run(prompt).on(
            "update",
            lambda data, event: (reader.write(f"{agent.name} 🤖 (debug) : ", data)),
        )

        reader.write(f"{agent.name} Agent 🤖 : ", response.last_message.text)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
Usage in Workflow You can compose multiple Agent Stack agents into advanced workflows using the BeeAI framework’s workflow capabilities. This example demonstrates a research and content creation pipeline: In this example, the GPT Researcher agent researches a topic, and the Podcast creator takes the research report and produces a podcast transcript. You can adjust or expand this pattern to orchestrate more complex multi-agent workflows.
Python
import asyncio
import sys
import traceback

from pydantic import BaseModel

# pyrefly: ignore [missing-module-attribute]
from beeai_framework.adapters.agentstack import AgentStackAgent
from beeai_framework.errors import FrameworkError
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory
from beeai_framework.workflows import Workflow
from examples.helpers.io import ConsoleReader


async def main() -> None:
    reader = ConsoleReader()

    class State(BaseModel):
        topic: str
        research: str | None = None
        output: str | None = None

    agents = await AgentStackAgent.from_agent_stack(url="http://127.0.0.1:8333", memory=UnconstrainedMemory())

    async def research(state: State) -> None:
        # Run the agent and observe events
        try:
            research_agent = next(agent for agent in agents if agent.name == "GPT Researcher")
        except StopIteration:
            raise ValueError("Agent 'GPT Researcher' not found") from None
        response = await research_agent.run(state.topic).on(
            "update",
            lambda data, _: (reader.write("Agent 🤖 (debug) : ", data)),
        )
        state.research = response.last_message.text

    async def podcast(state: State) -> None:
        # Run the agent and observe events
        try:
            podcast_agent = next(agent for agent in agents if agent.name == "Podcast creator")
        except StopIteration:
            raise ValueError("Agent 'Podcast creator' not found") from None
        response = await podcast_agent.run(state.research or "").on(
            "update",
            lambda data, _: (reader.write("Agent 🤖 (debug) : ", data)),
        )
        state.output = response.last_message.text

    # Define the structure of the workflow graph
    workflow = Workflow(State)
    workflow.add_step("research", research)
    workflow.add_step("podcast", podcast)

    # Execute the workflow
    result = await workflow.run(State(topic="Connemara"))

    print("\n*********************")
    print("Topic: ", result.state.topic)
    print("Research: ", result.state.research)
    print("Output: ", result.state.output)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
Use agent from the remote Agent Stack If you want to integrate an agent from a remote server with authorization, you must first obtain a JWT auth token, and then create a remote agent and set its required parameters.
Python
agents = await AgentStackAgent.from_agent_stack(url="https://my-agentstack-server.com/", auth_token="ey***")
or create a custom client:
Python
from agentstack_sdk.platform import PlatformClient
from beeai_framework.adapters.agentstack.agents import AgentStackAgent

 async with PlatformClient(auth_token="ey***", base_url="https://my-agentstack-server.com/") as client:
      agents = await AgentStackAgent.from_agent_stack(url=client)

​
Exposing to the platform (server)

The AgentStackServer class exposes an agent or any other runnable (tool/chat model, …) to the Agent Stack. It gets automatically registered to the platform and allows you to access and use the agents directly in the platform. Key Features:
  • easy to expose (deploy) the current application to the production-ready environment
  • built-in trajectory, forms integration, LLM inference support, …
  • easy to extend and debug
from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
from beeai_framework.adapters.agentstack.serve.server import AgentStackMemoryManager, AgentStackServer
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModelParameters
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.weather import OpenMeteoTool

try:
    from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail
except ModuleNotFoundError as e:
    raise ModuleNotFoundError(
        "Optional module [agentstack] not found.\nRun 'pip install \"beeai-framework[agentstack]\"' to install."
    ) from e


def main() -> None:
    llm = AgentStackChatModel(
        preferred_models=["openai:gpt-4o", "ollama:llama3.1:8b"],
        parameters=ChatModelParameters(stream=True),
    )
    agent = RequirementAgent(
        llm=llm,
        tools=[DuckDuckGoSearchTool(), OpenMeteoTool()],
        memory=UnconstrainedMemory(),
        middlewares=[GlobalTrajectoryMiddleware()],
    )

    # Runs HTTP server that registers to Agent Stack
    server = AgentStackServer(memory_manager=AgentStackMemoryManager())
    server.register(
        agent,
        name="Framework chat agent",  # (optional)
        description="Simple chat agent",  # (optional)
        detail=AgentDetail(interaction_mode="multi-turn"),  # default is multi-turn (optional)
    )
    server.serve()


if __name__ == "__main__":
    main()
Agent Stack supports only one entry per server. To register more, you need to spawn more servers.
You can use the platform to do LLM inference by using the AgentStackChatModel(...) class.

​
Configuration

Server The server’s behavior can be influenced via attributes listed in AgentStackServerConfig class (host, port, self registration, …). Internally the server preserves every conversation, the custom strategy can be used by implementing the base MemoryManager class.
Python
from beeai_framework.adapters.agentstack.serve.server import AgentStackServer, AgentStackServerConfig
from beeai_framework.serve.utils import LRUMemoryManager

memory_manager = LRUMemoryManager(maxsize=64) # keeps max 64 conversations
config = AgentStackServerConfig(host="127.0.0.1", port=9999, run_limit=3600, limit_concurrency=10)
server = AgentStackServer(config=config, memory_manager=memory_manager)
Agent The agent’s meta information is inferred from its metadata (the agent.meta property). However, this information can be overridden during agent registration. See the following example.
Python
from beeai_framework.adapters.agentstack.serve.server import AgentStackServer
from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail

server = AgentStackServer()
agent = RequirementAgent(llm=AgentStackChatModel())
server.register(agent,
	name="SmartAgent",
	description="Knows everything!",
	url="https://example.com",
	version="1.0.0",
	default_input_modes=["text", "text/plain"],
	default_output_modes=["text"],
	detail=AgentDetail(interaction_mode="multi-turn", user_greeting="What can I do for you?")
)

​
Customization

The Agent Stack has a concept of extensions that enable access to external services and UI components via dependency injection. The framework internally uses the following extensions:
  • Form Extension: for displaying prompts and other checks (for instance when using AskPermissionRequirement)..
  • Trajectory Extension: for showing agent’s intermediate steps throughout the execution
The current entries are listed in the BaseAgentStackExtensions class.
Custom Extensions The following implementation demonstrates an agent that conducts an internet search and provides an answer to the given question, with inline citations. It does so by leveraging the Citation Extension, which is managed by the PlatformCitationMiddleware class.
Learn more about the Middleware concept.
# Copyright 2025 © BeeAI a Series of LF Projects, LLC
# SPDX-License-Identifier: Apache-2.0
import re
import sys
import traceback
from typing import Annotated

from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
from beeai_framework.adapters.agentstack.context import AgentStackContext
from beeai_framework.adapters.agentstack.serve.server import AgentStackMemoryManager, AgentStackServer
from beeai_framework.adapters.agentstack.serve.types import BaseAgentStackExtensions
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.agents.requirement.events import RequirementAgentSuccessEvent
from beeai_framework.agents.requirement.requirements.conditional import ConditionalRequirement
from beeai_framework.backend import AssistantMessage
from beeai_framework.context import RunContext, RunMiddlewareProtocol
from beeai_framework.emitter import EmitterOptions, EventMeta
from beeai_framework.errors import FrameworkError
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.search.wikipedia import WikipediaTool
from beeai_framework.tools.think import ThinkTool

try:
    from agentstack_sdk.a2a.extensions import Citation, CitationExtensionServer, CitationExtensionSpec
    from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail
    from agentstack_sdk.a2a.types import AgentMessage
except ModuleNotFoundError as e:
    raise ModuleNotFoundError(
        "Optional module [agentstack] not found.\nRun 'pip install \"beeai-framework[agentstack]\"' to install."
    ) from e


class CitationMiddleware(RunMiddlewareProtocol):
    def __init__(self) -> None:
        self._context: AgentStackContext | None = None

    def bind(self, ctx: RunContext) -> None:
        self._context = AgentStackContext.get()
        # add emitter with the highest priority to ensure citations are sent before any other event handling
        ctx.emitter.on("success", self._handle_success, options=EmitterOptions(priority=10, is_blocking=True))

    async def _handle_success(self, data: RequirementAgentSuccessEvent, meta: EventMeta) -> None:
        assert self._context is not None
        citation_ext = self._context.extensions.get("citation")

        # check it is the final step
        if data.state.answer is not None:
            citations, clean_text = extract_citations(data.state.answer.text)

            if citations:
                await self._context.context.yield_async(
                    AgentMessage(metadata=citation_ext.citation_metadata(citations=citations))  # type: ignore[attr-defined]
                )
                # replace an assistant message with an updated text without citation links
                data.state.answer = AssistantMessage(content=clean_text)


# define custom extensions
class CustomExtensions(BaseAgentStackExtensions):
    citation: Annotated[CitationExtensionServer, CitationExtensionSpec()]


def main() -> None:
    agent = RequirementAgent(
        llm=AgentStackChatModel(preferred_models=["openai/gpt-4o"]),
        tools=[WikipediaTool(), ThinkTool(), DuckDuckGoSearchTool()],
        instructions=(
            "You are an AI assistant focused on retrieving information from online sources. "
            "Mandatory Search: Always search for the topic on Wikipedia and always search for related current news. "
            "Mandatory Output Structure: Return the result in two separate sections with headings: "
            " 1. Basic Information (primarily utilizing data from Wikipedia, if relevant). "
            " 2. News (primarily utilizing current news results). "
            "Mandatory Citation: Always include a source link for all given information, especially news."
        ),
        requirements=[
            ConditionalRequirement(ThinkTool, force_at_step=1, consecutive_allowed=False),
            ConditionalRequirement(WikipediaTool, min_invocations=1),
            ConditionalRequirement(DuckDuckGoSearchTool, min_invocations=1),
        ],
        description="Search for information based on a given phrase.",
        middlewares=[
            GlobalTrajectoryMiddleware(),
            CitationMiddleware(),
        ],  # add platform middleware to get citations from the platform
    )

    # Runs HTTP server that registers to Agent Stack
    server = AgentStackServer(memory_manager=AgentStackMemoryManager())  # use platform memory
    server.register(
        agent,
        name="Information retrieval",
        detail=AgentDetail(interaction_mode="single-turn", user_greeting="What can I search for you?"),
        extensions=CustomExtensions,
    )
    server.serve()


# function to extract citations from text and return clean text without citation links
def extract_citations(text: str) -> tuple[list[Citation], str]:
    citations, offset = [], 0
    pattern = r"\[([^\]]+)\]\(([^)]+)\)"

    for match in re.finditer(pattern, text):
        content, url = match.groups()
        start = match.start() - offset

        citations.append(
            Citation(
                url=url,
                title=url.split("/")[-1].replace("-", " ").title() or content[:50],
                description=content[:100] + ("..." if len(content) > 100 else ""),
                start_index=start,
                end_index=start + len(content),
            )
        )
        offset += len(match.group(0)) - len(content)

    return citations, re.sub(pattern, r"\1", text)


if __name__ == "__main__":
    try:
        main()
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
The AgentStackContext.get() can be called from anywhere if the code is running inside the given context.

​
Platform RAG Agent

You can use the Vector Store Search Tool to query the platform-native vector store service and leverage the platform-provided embedding service. 
from typing import Annotated

from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
from beeai_framework.adapters.agentstack.backend.embedding import AgentstackEmbeddingModel
from beeai_framework.adapters.agentstack.backend.vector_store import NativeVectorStore
from beeai_framework.adapters.agentstack.serve.server import AgentStackMemoryManager, AgentStackServer
from beeai_framework.adapters.agentstack.serve.types import BaseAgentStackExtensions
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModelParameters
from beeai_framework.backend.types import Document
from beeai_framework.context import RunContext, RunContextStartEvent, RunMiddlewareProtocol
from beeai_framework.emitter import EventMeta
from beeai_framework.emitter.utils import create_internal_event_matcher
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.search.retrieval import VectorStoreSearchTool

try:
    from agentstack_sdk.a2a.extensions import EmbeddingServiceExtensionServer, EmbeddingServiceExtensionSpec
except ModuleNotFoundError as e:
    raise ModuleNotFoundError(
        "Optional module [agentstack] not found.\nRun 'pip install \"beeai-framework[agentstack]\"' to install."
    ) from e


# The middleware is necessary since the embedding service's initialization occurs after the client's call to the agent.
class RAGMiddleware(RunMiddlewareProtocol):
    def __init__(self, vector_store: NativeVectorStore) -> None:
        self._vector_store = vector_store

    def bind(self, ctx: RunContext) -> None:  # pyrefly: ignore [bad-override]
        # Only insert the documents during initialization.
        ctx.emitter.on(create_internal_event_matcher("start"), self._on_start)

    async def _on_start(self, _: RunContextStartEvent, meta: EventMeta) -> None:
        if not self._vector_store.is_initialized:
            print("debug: initializing vector store")
            await self._vector_store.add_documents(
                [
                    Document(content="My name is John.", metadata={}),
                    Document(content="I am a python programmer.", metadata={}),
                    Document(content="I am 30 years old.", metadata={}),
                ]
            )


def main() -> None:
    llm = AgentStackChatModel(
        preferred_models=["openai:gpt-4o", "ollama:llama3.1:8b"],
        parameters=ChatModelParameters(stream=True),
    )

    # Initialize the embedding model from the Agent Stack.
    embedding_model = AgentstackEmbeddingModel(preferred_models=["ollama:nomic-embed-text:latest"])

    vector_store = NativeVectorStore(embedding_model)
    # vector_store = VectorStore.from_name("AgentStack:NativeVectorStore", embedding_model=embedding_model)

    agent = RequirementAgent(
        llm=llm,
        tools=[VectorStoreSearchTool(vector_store)],
        memory=UnconstrainedMemory(),
        middlewares=[RAGMiddleware(vector_store)],  # add middleware to initialize vector store
    )

    # define custom extensions
    class CustomExtensions(BaseAgentStackExtensions):
        # "The property name must be 'embedding'.
        embedding: Annotated[
            EmbeddingServiceExtensionServer,
            EmbeddingServiceExtensionSpec.single_demand(suggested=tuple(embedding_model.preferred_models)),
        ]

    # Runs HTTP server that registers to Agent Stack
    server = AgentStackServer(memory_manager=AgentStackMemoryManager())
    server.register(agent, name="Framework RAG agent", extensions=CustomExtensions)
    server.serve()


if __name__ == "__main__":
    main()