beeai

BeeAI Framework Integration with ContextForge

BeeAI Framework is a dual-language framework for building agentic applications in Python and TypeScript. It can use tools exposed by any MCP server through BeeAI's MCPTool, which discovers remote tools from an MCP client and makes them available to BeeAI agents.

ContextForge can be used in two common ways:

• Start mcpgateway-wrapper as a local stdio MCP server for clients that spawn subprocesses.
• Connect BeeAI directly to ContextForge over Streamable HTTP, or to a legacy SSE endpoint when required.

---

🛠 Installation

Install the Python packages when building BeeAI agents in Python:

pip install "beeai-framework[mcp]" mcp-contextforge-gateway

Install the TypeScript packages when building BeeAI agents in Node.js:

npm install beeai-framework @modelcontextprotocol/sdk zod

Prepare the ContextForge connection values used in the examples:

export MCPGATEWAY_BEARER_TOKEN="<gateway-token>"
export MCP_AUTH="Bearer ${MCPGATEWAY_BEARER_TOKEN}" # pragma: allowlist secret

# Virtual server endpoint used by mcpgateway-wrapper.
export MCP_SERVER_URL="http://localhost:4444/servers/UUID_OF_SERVER_1/mcp"

# Direct MCP endpoints used by HTTP/SSE clients.
export MCP_GATEWAY_MCP_URL="http://localhost:4444/mcp"
export MCP_GATEWAY_SSE_URL="http://localhost:4444/sse"

See mcpgateway-wrapper for details on generating tokens and choosing the correct virtual server URL.

---

🔗 Python: stdio via mcpgateway-wrapper

Use the wrapper when you want BeeAI to spawn ContextForge as a local stdio MCP server while the wrapper forwards calls to the gateway.

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.tools.mcp import MCPTool

server_params = StdioServerParameters(
    command="python3",
    args=["-m", "mcpgateway.wrapper"],
    env={
        "MCP_SERVER_URL": os.environ["MCP_SERVER_URL"],
        "MCP_AUTH": os.environ["MCP_AUTH"],
        "MCP_WRAPPER_LOG_LEVEL": "OFF",
    },
)


async def main() -> None:
    client = stdio_client(server_params)
    tools = await MCPTool.from_client(client)

    agent = RequirementAgent(
        llm=ChatModel.from_name(os.getenv("BEEAI_CHAT_MODEL", "ollama:granite4:micro")),
        tools=tools,
    )

    response = await agent.run("List the tools available through ContextForge.")
    print(response.last_message.text)


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