> ## 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

[Agent Stack](https://beeai.dev/) 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](https://beeai.dev/)** 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:

<CodeGroup>
  ```py Python [expandable] theme={null}
  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())

  ```

  ```ts TypeScript [expandable] theme={null}
  import "dotenv/config.js";
  import { AgentStackAgent } from "beeai-framework/adapters/agentstack/agents/agent";
  import { createConsoleReader } from "examples/helpers/io.js";
  import { FrameworkError } from "beeai-framework/errors";
  import { TokenMemory } from "beeai-framework/memory/tokenMemory";

  ////////////////////////////////////////////////////////
  ///   Supports only BeeAI platform version v0.2.xx   ///
  ////////////////////////////////////////////////////////

  const agentName = "chat";

  const instance = new AgentStackAgent({
    url: "http://127.0.0.1:8333/api/v1/acp",
    agentName,
    memory: new TokenMemory(),
  });

  const reader = createConsoleReader();

  try {
    for await (const { prompt } of reader) {
      const result = await instance.run({ input: prompt }).observe((emitter) => {
        emitter.on("update", (data) => {
          reader.write(`Agent (received progress) 🤖 : `, JSON.stringify(data.value, null, 2));
        });
        emitter.on("error", (data) => {
          reader.write(`Agent (error) 🤖 : `, data.message);
        });
      });

      reader.write(`Agent (${agentName}) 🤖 : `, result.result.text);
    }
  } catch (error) {
    reader.write("Agent (error)  🤖", FrameworkError.ensure(error).dump());
  }

  ```
</CodeGroup>

**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.

```py Python [expandable] theme={null}
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.

```py Python theme={null}
agents = await AgentStackAgent.from_agent_stack(url="https://my-agentstack-server.com/", auth_token="ey***")
```

or create a custom client:

```py Python theme={null}
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

<CodeGroup>
  ```py Python [expandable] theme={null}
  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()

  ```

  ```ts TypeScript [expandable] theme={null}
  // COMING SOON
  ```
</CodeGroup>

<Note>
  Agent Stack supports only one entry per server. To register more, you need to spawn more servers.
</Note>

<Tip>
  You can use the platform to do LLM inference by using the `AgentStackChatModel(...)` class.
</Tip>

### 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.

```py Python [expandable] theme={null}
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.

```py Python [expandable] theme={null}
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](https://docs.beeai.dev/concepts/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

<Note>The current entries are listed in the `BaseAgentStackExtensions` class.</Note>

**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](https://docs.beeai.dev/build-agents/citations), which is managed by the  `PlatformCitationMiddleware` class.

<Note>Learn more about the [Middleware](/modules/middleware) concept.</Note>

<CodeGroup>
  ```py Python [expandable] theme={null}
  # 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())

  ```

  ```ts TypeScript [expandable] theme={null}
  // COMING SOON
  ```
</CodeGroup>

<Tip>The `AgentStackContext.get()` can be called from anywhere if the code is running inside the given context.</Tip>

### Platform RAG Agent

You can use the [Vector Store Search Tool](/modules/tools#vector-store-search) to query the platform-native [vector store service](https://agentstack.beeai.dev/stable/agent-integration/rag#store) and leverage the platform-provided [embedding service](https://agentstack.beeai.dev/stable/agent-integration/rag#embedding).

<CodeGroup>
  ```py Python [expandable] theme={null}
  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()

  ```

  ```ts TypeScript [expandable] theme={null}
  // COMING SOON
  ```
</CodeGroup>