BeeAI Framework

https://framework.beeai.dev
Admin
BeeAI Framework is an open-source framework for building production-grade multi-agent systems with...
Tokens:91,963
Snippets:624
Trust Score:7.5
Update:1 month ago
Show doc for...
Tokens:
Raw
### Configure LLM backend

Source: https://framework.beeai.dev/introduction/quickstart

Commands and configuration examples for setting up local or hosted LLM backends.

```bash
ollama pull granite3.3
```

```bash
# Examples (pick one that matches your provider/account):
LLM_CHAT_MODEL_NAME="openai:gpt-5-mini"
```

```bash
# Examples (pick one that matches your provider/account):
OPENAI_API_KEY="YOUR API KEY"
```

--------------------------------

### Install BeeAI Framework with ACP-Zed support

Source: https://framework.beeai.dev/integrations/acp-zed

Install the necessary dependencies to enable the ACP-Zed adapter.

```bash
pip install "beeai-framework[acp-zed]"
```

--------------------------------

### Initialize Instrumentation

Source: https://framework.beeai.dev/modules/observability

Execute the setup function before initializing any BeeAI Framework components to ensure proper tracing.

```python
setup_observability()
```

--------------------------------

### Install BeeAI Framework and Orchestrate Extension

Source: https://framework.beeai.dev/integrations/watsonx-orchestrate

Commands to install the core framework and the specific extension required for watsonx Orchestrate integration.

```sh
pip install beeai-framework
```

```sh
pip install 'beeai-framework[watsonx-orchestrate]'
```

--------------------------------

### Install dependencies

Source: https://framework.beeai.dev/introduction/quickstart

Install the necessary project dependencies for Python or TypeScript environments.

```bash
uv sync
```

```bash
nvm install && nvm use
npm ci
```

--------------------------------

### Install RAG components

Source: https://framework.beeai.dev/modules/rag

Install the framework with the necessary RAG extras to enable support for document loaders, splitters, and vector stores.

```bash
pip install "beeai-framework[rag]"
```

--------------------------------

### Usage Example for FileCache

Source: https://framework.beeai.dev/modules/cache

Demonstrates initializing a cache, setting values, handling capacity eviction, and reloading from disk.

```python
async def main() -> None:
    with tempfile.TemporaryDirectory() as tmpdir:
        path = Path(tmpdir) / "bee_cache.json"
        cache: JsonFileCache[dict[str, str]] = JsonFileCache(path, size=2, ttl=1.5)

        await cache.set("profile", {"name": "Bee", "role": "assistant"})
        await cache.set("settings", {"theme": "dark"})
        print(f"Cache persisted to {cache.source}")

        await cache.set("session", {"token": "abc123"})
        print(await cache.has("profile"))  # False -> evicted when capacity exceeded

        reloaded: JsonFileCache[dict[str, str]] = JsonFileCache(path, size=2, ttl=1.5)
        print(await reloaded.get("settings"))  # {'theme': 'dark'}

        await asyncio.sleep(1.6)
        await reloaded.reload()
        print(await reloaded.get("session"))  # None -> TTL expired


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
```

--------------------------------

### Run the agent

Source: https://framework.beeai.dev/introduction/quickstart

Execute the agent script to start the activity planner.

```bash
uv run python beeai_framework_starter/agent.py
```

```bash
npm run start src/agent.ts
```

--------------------------------

### Install BeeAI OpenInference Instrumentor

Source: https://framework.beeai.dev/modules/observability

Install the required package to enable OpenInference instrumentation for the BeeAI Framework.

```bash
pip install openinference-instrumentation-beeai
```

--------------------------------

### Initialize and start an OpenAIServer

Source: https://framework.beeai.dev/integrations/openai-api

Configures and runs an OpenAIServer instance to expose an agent and an LLM on a specified port.

```python
from beeai_framework.adapters.openai.serve.server import OpenAIAPIType, OpenAIServer, OpenAIServerConfig
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.weather import OpenMeteoTool


def main() -> None:
    llm = ChatModel.from_name("ollama:granite4:micro")
    agent = RequirementAgent(
        llm=llm,
        tools=[OpenMeteoTool()],
        memory=UnconstrainedMemory(),
    )

    server = OpenAIServer(
        config=OpenAIServerConfig(
            port=9998,
            api=OpenAIAPIType.RESPONSES,
        )
    )
    server.register(agent, name="agent")
    server.register(llm)
    server.serve()


if __name__ == "__main__":
    main()
```

```typescript
 // coming soon
```

--------------------------------

### Setup and Load Knowledge Base in Python

Source: https://framework.beeai.dev/introduction/tour

Initializes a vector store with an Ollama embedding model and processes Markdown files into chunks for retrieval.

```python
import asyncio
from beeai_framework.backend.document_loader import DocumentLoader
from beeai_framework.backend.embedding import EmbeddingModel
from beeai_framework.backend.text_splitter import TextSplitter
from beeai_framework.backend.vector_store import VectorStore

async def setup_knowledge_base():
    # Create embedding model using Ollama
    embedding_model = EmbeddingModel.from_name("ollama:nomic-embed-text")

    # Create vector store
    vector_store = VectorStore.from_name(
        "beeai:TemporalVectorStore",
        embedding_model=embedding_model
    )

    # Setup text splitter for chunking documents
    text_splitter = TextSplitter.from_name(
        "langchain:RecursiveCharacterTextSplitter",
        chunk_size=1000,
        chunk_overlap=200
    )

    return vector_store, text_splitter

async def load_documents(vector_store, text_splitter, file_paths):
    """Load documents into the vector store"""
    all_chunks = []

    for file_path in file_paths:
        try:
            # Load the document
            loader = DocumentLoader.from_name(
                "langchain:UnstructuredMarkdownLoader",
                file_path=file_path
            )
            documents = await loader.load()

            # Split into chunks
            chunks = await text_splitter.split_documents(documents)
            all_chunks.extend(chunks)
            print(f"Loaded {len(chunks)} chunks from {file_path}")
        except Exception as e:
            print(f"Failed to load {file_path}: {e}")

    # Add all chunks to vector store
    if all_chunks:
        await vector_store.add_documents(all_chunks)
        print(f"Total chunks added: {len(all_chunks)}")

    return vector_store if all_chunks else None

async def main():
    # Setup the knowledge base
    vector_store, text_splitter = await setup_knowledge_base()

    # Replace with your actual markdown files
    file_paths = [
        "your_document1.md", #replace these documents with the path to your local document
        "your_document2.md", #replace these documents with the path to your local document
    ]

    # Load documents
    loaded_vector_store = await load_documents(vector_store, text_splitter, file_paths)

    if loaded_vector_store:
        print("Knowledge base ready!")
        return loaded_vector_store
    else:
        print("No documents loaded")
        return None

if __name__ == "__main__":
    # Run this first to setup your knowledge base
    asyncio.run(main())
```

--------------------------------

### Connect to MCP Server via Stdio

Source: https://framework.beeai.dev/modules/tools

Examples of initializing an MCP client and exposing tools to an agent using stdio transport.

```python
import asyncio
import os

from mcp import StdioServerParameters, stdio_client

from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools import Tool
from beeai_framework.tools.mcp import MCPTool

server_params = StdioServerParameters(
    command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", os.getcwd()]
)


async def main() -> None:
    """Using a local MCP file server to explore the file system."""

    client = stdio_client(server_params)
    mcp_tools = await MCPTool.from_client(client)

    agent = RequirementAgent(
        llm=ChatModel.from_name("ollama:granite4:micro"),
        tools=mcp_tools,
        middlewares=[GlobalTrajectoryMiddleware(included=[Tool, ChatModel])],
    )

    prompt = "What's the current working directory?"
    print(f"User: {prompt}")
    response = await agent.run(prompt)
    print(f"Agent: {response.last_message.text}")


if __name__ == "__main__":
    asyncio.run(main())
```

```typescript
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { MCPTool } from "beeai-framework/tools/mcp";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { ReActAgent } from "beeai-framework/agents/react/agent";
import { UnconstrainedMemory } from "beeai-framework/memory/unconstrainedMemory";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";

// Create MCP Client
const client = new Client(
  {
    name: "test-client",
    version: "1.0.0",
  },
  {
    capabilities: {},
  },
);

// Connect the client to any MCP server with tools capablity
await client.connect(
  new StdioClientTransport({
    command: "npx",
    args: ["-y", "@modelcontextprotocol/server-everything"],
  }),
);
```

--------------------------------

### Main Execution Loop

Source: https://framework.beeai.dev/modules/backend

The main entry point that orchestrates various backend examples and handles errors.

```python
async def main() -> None:
    print("*" * 10, "openai_from_name")
    await openai_from_name()
    print("*" * 10, "openai_granite_from_name")
    await openai_granite_from_name()
    print("*" * 10, "openai_sync")
    await openai_sync()
    print("*" * 10, "openai_stream")
    await openai_stream()
    print("*" * 10, "openai_stream_abort")
    await openai_stream_abort()
    print("*" * 10, "openai_structure")
    await openai_structure()
    print("*" * 10, "openai_stream_parser")
    await openai_stream_parser()
    print("*" * 10, "openai_tool_calling")
    await openai_tool_calling()
    print("*" * 10, "openai_embedding")
    await openai_embedding()
    print("*" * 10, "openai_cloning")
    await openai_cloning()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
```

--------------------------------

### Expose an agent as an IBM watsonx Orchestrate server

Source: https://framework.beeai.dev/introduction/tour

Configures and starts a server instance to serve an agent via the /chat/completions endpoint.

```python
from beeai_framework.adapters.watsonx_orchestrate import WatsonxOrchestrateServer, WatsonxOrchestrateServerConfig
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.serve.utils import LRUMemoryManager
from beeai_framework.tools.weather import OpenMeteoTool

def main() -> None:
    llm = ChatModel.from_name("ollama:granite3.3")
    agent = RequirementAgent(
        llm=llm,
        tools=[OpenMeteoTool()],
        memory=UnconstrainedMemory(),
        instructions="You are a weather agent that provides accurate weather information."
    )

    config = WatsonxOrchestrateServerConfig(port=8080, host="0.0.0.0", api_key=None)  # optional
    # use LRU memory manager to keep limited amount of sessions in the memory
    server = WatsonxOrchestrateServer(config=config, memory_manager=LRUMemoryManager(maxsize=100))
    server.register(agent)

    # start an API with /chat/completions endpoint which is compatible with Watsonx Orchestrate
    server.serve()


if __name__ == "__main__":
    main()
```

```typescript
COMING SOON
```

--------------------------------

### Initialize and serve an ACP Zed agent

Source: https://framework.beeai.dev/integrations/acp-zed

Configures and starts an ACPZedServer instance. Omit agent_name and agent_description to use the agent's internal metadata.

```python
from beeai_framework.adapters.acp_zed import ACPZedServer, ACPZedServerConfig

ACPZedServer(
    config=ACPZedServerConfig(
        agent_name="beeai-coder",
        agent_description="A coding agent backed by BeeAI.",
    )
).register(agent).serve()
```

--------------------------------

### Implement ToolCallingAgent in Python and TypeScript

Source: https://framework.beeai.dev/modules/agents

Examples demonstrating how to initialize a ToolCallingAgent with memory and tools, and how to process user input via a console reader.

```python
import asyncio
import logging
import sys
import traceback
from typing import Any

from dotenv import load_dotenv

from beeai_framework.agents.tool_calling import ToolCallingAgent
from beeai_framework.backend import ChatModel
from beeai_framework.emitter import EventMeta
from beeai_framework.errors import FrameworkError
from beeai_framework.logger import Logger
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.weather import OpenMeteoTool
from examples.helpers.io import ConsoleReader

# Load environment variables
load_dotenv()

# Configure logging - using DEBUG instead of trace
logger = Logger("app", level=logging.DEBUG)

reader = ConsoleReader()


def process_agent_events(data: Any, event: EventMeta) -> None:
    """Process agent events and log appropriately"""

    if event.name == "start":
        reader.write("Agent (debug) 🤖 : ", "starting new iteration")
    elif event.name == "success":
        reader.write("Agent (debug) 🤖 : ", data.state.memory.messages[-1])


async def main() -> None:
    """Main application loop"""

    # Create agent
    agent = ToolCallingAgent(
        llm=ChatModel.from_name("ollama:llama3.1"), memory=UnconstrainedMemory(), tools=[OpenMeteoTool()]
    )

    # Main interaction loop with user input
    for prompt in reader:
        response = await agent.run(prompt).on("*", process_agent_events)
        reader.write("Agent 🤖 : ", response.last_message.text)

    print("======DONE (showing the full message history)=======")

    # pyrefly: ignore [unbound-name]
    messages = response.state.memory.messages
    for msg in messages:
        print(msg)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
```

```typescript
import "dotenv/config.js";
import { createConsoleReader } from "../../helpers/io.js";
import { FrameworkError } from "beeai-framework/errors";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { Logger } from "beeai-framework/logger/logger";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";

Logger.root.level = "silent"; // disable internal logs
const logger = new Logger({ name: "app", level: "trace" });

// Other models to try:
// "llama3.1:70b"
// "granite3.3"
// "deepseek-r1:32b"
// ensure the model is pulled before running
const llm = new OllamaChatModel("granite4:micro");

const agent = new ToolCallingAgent({
  llm,
  memory: new TokenMemory(),
  templates: {
    system: (template) =>
      template.fork((config) => {
        config.defaults.instructions =
          "You are a helpful assistant that uses tools to answer questions.";
      }),
  },
  tools: [
    new OpenMeteoTool(), // weather tool
  ],
});

const reader = createConsoleReader();

try {
  for await (const { prompt } of reader) {
    let messagesCount = agent.memory.messages.length + 1;

    const response = await agent.run({ prompt }).observe((emitter) => {
      emitter.on("success", async ({ state }) => {
        const newMessages = state.memory.messages.slice(messagesCount);
        messagesCount += newMessages.length;

        reader.write(
          `Agent (${newMessages.length} new messages) 🤖 :\n`,
          newMessages.map((msg) => `-> ${JSON.stringify(msg.toPlain())}`).join("\n"),
        );
      });

      // To observe all events (uncomment following block)
      // emitter.match("*.*", async (data: unknown, event) => {
      //   logger.trace(event, `Received event "${event.path}"`);
      // }, {
      //   matchNested: true
      // });

      // To get raw LLM input (uncomment following block)
      // emitter.match(
      //   (event) => event.creator === llm && event.name === "start",
      //   async (data: InferCallbackValue<GenerateEvents["start"]>, event) => {
      //     logger.trace(
      //       event,
      //       [
      //         `Received LLM event "${event.path}"`,
      //         JSON.stringify(data.input), // array of messages
      //       ].join("\n"),
      //     );
      //   },
      // );
    });

    reader.write(`Agent 🤖 : `, response.result.text);
  }
} catch (error) {
  logger.error(FrameworkError.ensure(error).dump());
} finally {
  reader.close();
}
```

--------------------------------

### Define Tool Description

Source: https://framework.beeai.dev/modules/tools

Provide a succinct natural language description to guide the agent on when to utilize the tool.

```python
description = "Takes X action when given Y input resulting in Z output"
```

--------------------------------

### Implement a basic custom tool

Source: https://framework.beeai.dev/modules/tools

Examples of creating a custom tool using Python class inheritance or TypeScript's CustomTool.fromSourceCode helper.

```python
import asyncio
import random
import sys
from typing import Any

from pydantic import BaseModel, Field

from beeai_framework.context import RunContext
from beeai_framework.emitter import Emitter
from beeai_framework.errors import FrameworkError
from beeai_framework.tools import StringToolOutput, Tool, ToolRunOptions


class RiddleToolInput(BaseModel):
    riddle_number: int = Field(description="Index of riddle to retrieve.")


class RiddleTool(Tool[RiddleToolInput, ToolRunOptions, StringToolOutput]):
    name = "Riddle"
    description = "It selects a riddle to test your knowledge."
    input_schema = RiddleToolInput

    data = (
        "What has hands but can't clap?",
        "What has a face and two hands but no arms or legs?",
        "What gets wetter the more it dries?",
        "What has to be broken before you can use it?",
        "What has a head, a tail, but no body?",
        "The more you take, the more you leave behind. What am I?",
        "What goes up but never comes down?",
    )

    def __init__(self, options: dict[str, Any] | None = None) -> None:
        super().__init__(options)

    def _create_emitter(self) -> Emitter:
        return Emitter.root().child(
            namespace=["tool", "example", "riddle"],
            creator=self,
        )

    async def _run(
        self, input: RiddleToolInput, options: ToolRunOptions | None, context: RunContext
    ) -> StringToolOutput:
        index = input.riddle_number % (len(self.data))
        riddle = self.data[index]
        return StringToolOutput(result=riddle)


async def main() -> None:
    tool = RiddleTool()
    tool_input = RiddleToolInput(riddle_number=random.randint(0, len(RiddleTool.data)))
    result = await tool.run(tool_input)
    print(result)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        sys.exit(e.explain())
```

```typescript
import "dotenv/config";
import { CustomTool } from "beeai-framework/tools/custom";

const customTool = await CustomTool.fromSourceCode(
  {
    // Ensure the env exists
    url: process.env.CODE_INTERPRETER_URL!,
    env: { API_URL: "https://riddles-api.vercel.app/random" },
  },
  `import requests
import os
from typing import Optional, Union, Dict

def get_riddle() -> Optional[Dict[str, str]]:
  """
  Fetches a random riddle from the Riddles API.

  This function retrieves a random riddle and its answer. It does not accept any input parameters.

  Returns:
      Optional[Dict[str, str]]: A dictionary containing:
          - 'riddle' (str): The riddle question.
          - 'answer' (str): The answer to the riddle.
      Returns None if the request fails.
  """
  url = os.environ.get('API_URL')
  
  try:
      response = requests.get(url)
      response.raise_for_status() 
      return response.json() 
  except Exception as e:
      return None`,
);
```

--------------------------------

### Setup Vector Store Search in Python

Source: https://framework.beeai.dev/modules/tools

Configures a vector store for semantic search, including document loading, chunking, and embedding generation. Requires a pre-populated store or offline population logic.

```python
import asyncio
import os

from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.backend.document_loader import DocumentLoader
from beeai_framework.backend.embedding import EmbeddingModel
from beeai_framework.backend.text_splitter import TextSplitter
from beeai_framework.backend.vector_store import VectorStore
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools import Tool
from beeai_framework.tools.search.retrieval import VectorStoreSearchTool

POPULATE_VECTOR_DB = True
VECTOR_DB_PATH_4_DUMP = ""  # Set this path for persistency


async def setup_vector_store() -> VectorStore | None:
    """
    Setup vector store with BeeAI framework documentation.
    """
    embedding_model = EmbeddingModel.from_name("watsonx:ibm/slate-125m-english-rtrvr-v2", truncate_input_tokens=500)

    # Load existing vector store if available
    # pyrefly: ignore [redundant-condition]
    if VECTOR_DB_PATH_4_DUMP and os.path.exists(VECTOR_DB_PATH_4_DUMP):
        print(f"Loading vector store from: {VECTOR_DB_PATH_4_DUMP}")
        from beeai_framework.adapters.beeai.backend.vector_store import TemporalVectorStore

        preloaded_vector_store: VectorStore = TemporalVectorStore.load(
            path=VECTOR_DB_PATH_4_DUMP, embedding_model=embedding_model
        )
        return preloaded_vector_store

    # Create new vector store if population is enabled
    # NOTE: Vector store population is typically done offline in production applications
    if POPULATE_VECTOR_DB:
        # Load documentation about BeeAI agents - this serves as our knowledge base
        # for answering questions about the different types of agents available
        loader = DocumentLoader.from_name(
            name="langchain:UnstructuredMarkdownLoader", file_path="docs/modules/agents.mdx"
        )
        try:
            documents = await loader.load()
        except Exception as e:
            print(f"Failed to load documents: {e}")
            return None

        # Split documents into chunks
        text_splitter = TextSplitter.from_name(
            name="langchain:RecursiveCharacterTextSplitter", chunk_size=1000, chunk_overlap=200
        )
        documents = await text_splitter.split_documents(documents)
        print(f"Loaded {len(documents)} document chunks")

        # Create vector store and add documents
        vector_store = VectorStore.from_name(name="beeai:TemporalVectorStore", embedding_model=embedding_model)
        await vector_store.add_documents(documents=documents)
        print("Vector store populated with documents")

        return vector_store

    return None


async def main() -> None:
    """
    Example demonstrating RequirementAgent using VectorStoreSearchTool.

```

--------------------------------

### Initialize and run a LiteAgent

Source: https://framework.beeai.dev/modules/agents

Demonstrates setting up a minimal LiteAgent without a default system prompt and handling streaming output via event emitters.

```python
import asyncio

from beeai_framework.agents.lite import LiteAgent
from beeai_framework.backend import ChatModel, ChatModelOutput, ChatModelParameters, SystemMessage
from beeai_framework.emitter import EventMeta
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.think import ThinkTool
from beeai_framework.tools.weather import OpenMeteoTool


async def main() -> None:
    agent = LiteAgent(
        llm=ChatModel.from_name("ollama:granite4:micro", ChatModelParameters(stream=True)),
        tools=[ThinkTool(), OpenMeteoTool(), DuckDuckGoSearchTool()],
        middlewares=[GlobalTrajectoryMiddleware()],
    )

    # Optionally set a custom system prompt
    await agent.memory.add(SystemMessage("You are a helpful assistant."))

    @agent.emitter.on("final_answer")
    def stream_final_answer(data: ChatModelOutput, meta: EventMeta) -> None:
        print(data.get_text_content())  # emits chunks

    await agent.run("Hello")


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Initialize Vector Store Search Tool

Source: https://framework.beeai.dev/modules/tools

Demonstrates setting up a vector store and integrating it into a RequirementAgent for documentation-based queries.

```python
# Setup vector store with BeeAI documentation
    vector_store = await setup_vector_store()
    if vector_store is None:
        raise FileNotFoundError(
            "Failed to instantiate Vector Store. "
            "Either set POPULATE_VECTOR_DB=True to create a new one, or ensure the database file exists."
        )

    # Create the vector store search tool
    search_tool = VectorStoreSearchTool(vector_store=vector_store)

    # Alternative: Create search tool using dynamic loading
    # embedding_model = EmbeddingModel.from_name("watsonx:ibm/slate-125m-english-rtrvr-v2", truncate_input_tokens=500)
    # search_tool = VectorStoreSearchTool.from_name(
    #     name="beeai:TemporalVectorStore",
    #     embedding_model=embedding_model
    # )

    # Create RequirementAgent with the vector store search tool
    llm = ChatModel.from_name("ollama:llama3.1:8b")
    agent = RequirementAgent(
        llm=llm,
        memory=UnconstrainedMemory(),
        instructions=(
            "You are a helpful assistant that answers questions about the BeeAI framework. "
            "Use the vector store search tool to find relevant information from the documentation "
            "before providing your answer. Always search for information first, then provide a "
            "comprehensive response based on what you found."
        ),
        tools=[search_tool],
        # Log all tool calls to the console for easier debugging
        middlewares=[GlobalTrajectoryMiddleware(included=[Tool])],
    )

    query = "What types of agents are available in BeeAI?"
    response = await agent.run(query)
    print(f"query: {query}\nResponse: {response.last_message.text}")


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Initialize project environment

Source: https://framework.beeai.dev/guides/mcp-slackbot

Commands to create the project directory and set up the Python virtual environment.

```bash
mkdir beeai-slack-agent && cd beeai-slack-agent
```

```bash
python -m venv venv && source venv/bin/activate
```

--------------------------------

### Clone the starter repository

Source: https://framework.beeai.dev/introduction/quickstart

Use these commands to clone the appropriate starter template for your preferred language.

```bash
git clone https://github.com/i-am-bee/beeai-framework-py-starter.git
cd beeai-framework-py-starter
```

```bash
git clone https://github.com/i-am-bee/beeai-framework-ts-starter.git
cd beeai-framework-ts-starter
```

--------------------------------

### Connect to Agent Stack Agents

Source: https://framework.beeai.dev/integrations/agent-stack

Use the AgentStackAgent class to interact with agents hosted on the Agent Stack. The Python example supports dynamic agent selection, while the TypeScript example targets specific agent instances.

```python
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())
```

```typescript
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());
}
```

--------------------------------

### Initialize and run models with Python

Source: https://framework.beeai.dev/modules/backend

Demonstrates various interaction patterns including basic chat, streaming, abort signals, structured output, and custom parsing.

```python
import asyncio
import datetime
import sys
import traceback

from pydantic import BaseModel, Field

from beeai_framework.adapters.openai import OpenAIChatModel, OpenAIEmbeddingModel
from beeai_framework.backend import (
    ChatModel,
    ChatModelNewTokenEvent,
    ChatModelParameters,
    MessageToolResultContent,
    ToolMessage,
    UserMessage,
)
from beeai_framework.emitter import EventMeta
from beeai_framework.errors import AbortError, FrameworkError
from beeai_framework.parsers.field import ParserField
from beeai_framework.parsers.line_prefix import LinePrefixParser, LinePrefixParserNode
from beeai_framework.tools.weather import OpenMeteoTool, OpenMeteoToolInput
from beeai_framework.utils import AbortSignal


async def openai_from_name() -> None:
    llm = ChatModel.from_name("openai:gpt-4.1-mini")
    user_message = UserMessage("what states are part of New England?")
    response = await llm.run([user_message])
    print(response.get_text_content())


async def openai_granite_from_name() -> None:
    llm = ChatModel.from_name("openai:gpt-4.1-mini")
    user_message = UserMessage("what states are part of New England?")
    response = await llm.run([user_message])
    print(response.get_text_content())


async def openai_sync() -> None:
    llm = OpenAIChatModel("gpt-4.1-mini")
    user_message = UserMessage("what is the capital of Massachusetts?")
    response = await llm.run([user_message])
    print(response.get_text_content())


async def openai_stream() -> None:
    llm = OpenAIChatModel("gpt-4.1-mini")
    user_message = UserMessage("How many islands make up the country of Cape Verde?")
    response = await llm.run([user_message], stream=True)
    print(response.get_text_content())


async def openai_stream_abort() -> None:
    llm = OpenAIChatModel("gpt-4.1-mini")
    user_message = UserMessage("What is the smallest of the Cape Verde islands?")

    try:
        response = await llm.run([user_message], stream=True, signal=AbortSignal.timeout(0.5))

        if response is not None:
            print(response.get_text_content())
        else:
            print("No response returned.")
    except AbortError as err:
        print(f"Aborted: {err}")


async def openai_structure() -> None:
    class TestSchema(BaseModel):
        answer: str = Field(description="your final answer")

    llm = OpenAIChatModel("gpt-4.1-mini")
    user_message = UserMessage("How many islands make up the country of Cape Verde?")
    response = await llm.run([user_message], response_format=TestSchema, stream=True)
    print(response.output_structured)


async def openai_stream_parser() -> None:
    llm = OpenAIChatModel("gpt-4.1-mini")

    parser = LinePrefixParser(
        nodes={
            "test": LinePrefixParserNode(
                prefix="Prefix: ", field=ParserField.from_type(str), is_start=True, is_end=True
            )
        }
    )

    async def on_new_token(data: ChatModelNewTokenEvent, event: EventMeta) -> None:
        await parser.add(data.value.get_text_content())

    user_message = UserMessage("Produce 3 lines each starting with 'Prefix: ' followed by a sentence and a new line.")
    await llm.run([user_message], stream=True).observe(lambda emitter: emitter.on("new_token", on_new_token))
    result = await parser.end()
    print(result)
```

--------------------------------

### TypeScript FileCache Usage

Source: https://framework.beeai.dev/modules/cache

Example of initializing and using the FileCache in a TypeScript environment.

```typescript
import { FileCache } from "beeai-framework/cache/fileCache";
import * as os from "node:os";

const cache = new FileCache({
  fullPath: `${os.tmpdir()}/bee_file_cache_${Date.now()}.json`,
});
console.log(`Saving cache to "${cache.source}"`);
await cache.set("abc", { firstName: "John", lastName: "Doe" });
```

--------------------------------

### Serialized File Content Schema

Source: https://framework.beeai.dev/modules/backend

Example of the flattened JSON schema for file content parts.

```json
{
  "type": "file",
  "file_id": "https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf",
  "format": "application/pdf"
}
```

--------------------------------

### Configure environment and agent files

Source: https://framework.beeai.dev/guides/mcp-slackbot

Commands to initialize the .env file and create the agent module file.

```bash
echo -e "SLACK_BOT_TOKEN=\nSLACK_TEAM_ID=" >> .env
```

```bash
mkdir my_agents && touch my_agents/slack_agent.py
```

--------------------------------

### Main Execution Loop for LangChain Examples

Source: https://framework.beeai.dev/modules/backend

Orchestrates the execution of various LangChain Ollama test functions.

```python
async def main() -> None:
    print("*" * 10, "langchain_ollama_from_name")
    await langchain_ollama_from_name()
    print("*" * 10, "langchain_ollama_granite_from_name")
    await langchain_ollama_granite_from_name()
    print("*" * 10, "langchain_ollama_sync")
    await langchain_ollama_sync()
    print("*" * 10, "langchain_ollama_stream")
    await langchain_ollama_stream()
    print("*" * 10, "langchain_ollama_stream_abort")
    await langchain_ollama_stream_abort()
    print("*" * 10, "langchain_ollama_structure")
    await langchain_ollama_structure()
    print("*" * 10, "langchain_ollama_stream_parser")
    await langchain_ollama_stream_parser()
    print("*" * 10, "langchain_ollama_tool_calling")
    await langchain_ollama_tool_calling()
    print("*" * 10, "langchain_ollama_cloning")
    await langchain_ollama_cloning()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
```

--------------------------------

### Create and Run an MCP Server

Source: https://framework.beeai.dev/integrations/mcp

Demonstrates how to define tools and register them with an MCPServer instance using custom transport configurations.

```python
from beeai_framework.adapters.mcp.serve.server import MCPServer, MCPServerConfig, MCPSettings
from beeai_framework.tools import tool
from beeai_framework.tools.types import StringToolOutput
from beeai_framework.tools.weather.openmeteo import OpenMeteoTool


@tool
def reverse_tool(word: str) -> StringToolOutput:
    """A tool that reverses a word"""
    return StringToolOutput(result=word[::-1])


def main() -> None:
    """Create an MCP server with custom config, register ReverseTool and OpenMeteoTool to the MCP server and run it."""

    config = MCPServerConfig(transport="streamable-http", settings=MCPSettings(port=8001))  # optional
    server = MCPServer(config=config)
    server.register_many([reverse_tool, OpenMeteoTool()])
    server.serve()


if __name__ == "__main__":
    main()
```

```typescript
import { StringToolOutput, Tool, ToolEmitter, ToolInput } from "beeai-framework/tools/base";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { MCPServer, MCPServerConfig } from "beeai-framework/adapters/mcp/serve/server";
import { Emitter } from "beeai-framework/emitter/emitter";
import { z } from "zod";

export class ReverseTool extends Tool<StringToolOutput> {
  name = "ReverseTool";
  description = "A tool that reverses a word";

  public readonly emitter: ToolEmitter<ToolInput<this>, StringToolOutput> = Emitter.root.child({
    namespace: ["tool", "reverseTool"],
    creator: this,
  });

  inputSchema() {
    return z.object({
      word: z.string(),
    });
  }

  protected async _run(input: ToolInput<this>): Promise<StringToolOutput> {
    return new StringToolOutput(input.word.split("").reverse().join(""));
  }
}

// create an MCP server with custom config, register ReverseTool and OpenMeteoTool and run it
const config = new MCPServerConfig({ transport: "streamable-http", port: 8001 });
const server = new MCPServer(config);
server.registerMany([new ReverseTool(), new OpenMeteoTool()]);
await server.serve();
```

--------------------------------

### Initialize OpenTelemetry Observability

Source: https://framework.beeai.dev/introduction/tour

Set up the tracer provider and instrument the BeeAI framework before initializing any agents.

```python
from openinference.instrumentation.beeai import BeeAIInstrumentor
from opentelemetry import trace as trace_api
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk import trace as trace_sdk
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import SimpleSpanProcessor  # Use BatchSpanProcessor in prod

def setup_observability() -> None:
    tracer_provider = trace_sdk.TracerProvider(resource=Resource.create({}))
    tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter()))
    trace_api.set_tracer_provider(tracer_provider)
    BeeAIInstrumentor().instrument()  # auto-instruments BeeAI Framework

# Call this BEFORE creating/running any BeeAI agents
setup_observability()
```

```typescript
// Note: OpenTelemetry instrumentation for TypeScript is coming soon
// For now, use standard OpenTelemetry setup with manual instrumentation
```

--------------------------------

### ReAct Agent Output Format

Source: https://framework.beeai.dev/modules/agents

Example of the structured output emitted by the ReAct agent during its reasoning process.

```log
thought: I have the current weather in Las Vegas in Celsius.
final_answer: The current weather in Las Vegas is 20.5°C with an apparent temperature of 18.3°C.
```

--------------------------------

### Execute a tool with basic input

Source: https://framework.beeai.dev/modules/tools

Instantiate a tool and execute it using the run method with required input parameters.

```python
import asyncio
import datetime
import sys

from beeai_framework.errors import FrameworkError
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools.weather import OpenMeteoTool, OpenMeteoToolInput


async def main() -> None:
    tool = OpenMeteoTool()
    result = await tool.run(
        input=OpenMeteoToolInput(
            location_name="New York", start_date=datetime.date.today(), end_date=datetime.date.today()
        )
    ).middleware(GlobalTrajectoryMiddleware())
    print(result.get_text_content())


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        sys.exit(e.explain())
```

```typescript
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";

const tool = new OpenMeteoTool();

const today = new Date().toISOString().split("T")[0];
const result = await tool.run({
  location: { name: "New York" },
  start_date: today,
  end_date: today,
});
console.log(result.getTextContent());
```

--------------------------------

### ReAct Agent iteration log

Source: https://framework.beeai.dev/modules/agents

Example log output showing the reasoning and tool invocation structure of a ReAct agent.

```log
thought: I need to retrieve the current weather in Las Vegas. I can use the OpenMeteo function to get the current weather forecast for a location.
tool_name: OpenMeteo
tool_input: {"location": {"name": "Las Vegas"}, "start_date": "2024-10-17", "end_date": "2024-10-17", "temperature_unit": "celsius"}
```

--------------------------------

### Initialize OpenAPITool for Agents

Source: https://framework.beeai.dev/modules/tools

Demonstrates loading an OpenAPI schema to create tools for an agent to interact with external APIs.

```python
import asyncio
import json
import os
import sys
import traceback

from aiofiles import open

from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.errors import FrameworkError
from beeai_framework.tools.openapi import OpenAPITool


async def main() -> None:
    # Retrieve the schema
    current_dir = os.path.dirname(__file__)
    # pyrefly: ignore [bad-context-manager]
    async with open(f"{current_dir}/assets/github_openapi.json") as file:
        content = await file.read()
        open_api_schema = json.loads(content)

        # Create a tool for each operation in the schema
        tools = OpenAPITool.from_schema(open_api_schema)
        print(f"Retrieved {len(tools)} tools")
        print("\n".join([t.name for t in tools]))

        # Create an agent
        agent = RequirementAgent(llm="ollama:granite4:micro", tools=tools)

        # Run the agent
        prompt = "How many repositories are in 'i-am-bee' org?"
        print("User:", prompt)
        response = await agent.run(prompt)
        print("Agent 🤖 : ", response.last_message.text)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())
```

```typescript
import "dotenv/config.js";
import { ReActAgent } from "beeai-framework/agents/react/agent";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { OpenAPITool } from "beeai-framework/tools/openapi";
import * as fs from "fs";
import { dirname } from "node:path";
import { fileURLToPath } from "node:url";
import { ChatModel } from "beeai-framework/backend/chat";

const llm = await ChatModel.fromName("ollama:granite4:micro");

const __dirname = dirname(fileURLToPath(import.meta.url));
const openApiSchema = await fs.promises.readFile(
  `${__dirname}/assets/github_openapi.json`,
  "utf-8",
);

const agent = new ReActAgent({
  llm,
  memory: new TokenMemory(),
  tools: [new OpenAPITool({ openApiSchema })],
});

const response = await agent
  .run({ prompt: 'How many repositories are in "i-am-bee" org?' })
  .observe((emitter) => {
    emitter.on("update", async ({ data, update, meta }) => {
      console.log(`Agent (${update.key}) 🤖 : `, update.value);
    });
  });

console.log(`Agent 🤖 : `, response.result.text);
```
Tokens:
Raw