BeeAI Framework

Build agents in BeeAI Framework that communicate with StackOne’s A2A agents using A2AAgent.

Overview

The BeeAI Framework is an open-source framework for building AI agents, developed under the Linux Foundation. BeeAI has built-in support for the A2A protocol through A2AAgent, allowing your agents to communicate with remote A2A agents like StackOne’s.

Use BeeAI Framework to build agents that:

• Consume StackOne’s A2A agents as remote agents
• Orchestrate multi-agent systems with local and remote agents
• Access StackOne platform actions without managing tool definitions

Python only: BeeAI Framework’s A2AAgent supports custom authentication headers required for StackOne in Python. For TypeScript, use the A2A JavaScript SDK directly.

Installation

Install BeeAI Framework with A2A support:

pip install beeai-framework[a2a]

Quick Start

This example creates an A2AAgent that connects to a StackOne A2A agent.

import asyncio
import base64
from beeai_framework.adapters.a2a.agents import A2AAgent, A2AAgentParameters, HttpxAsyncClientParameters
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory

# StackOne A2A configuration
STACKONE_API_KEY = "<stackone_api_key>"
STACKONE_ACCOUNT_ID = "<account_id>"
BASE64_API_KEY = base64.b64encode(f"{STACKONE_API_KEY}:".encode()).decode()

async def main():
    # Create A2A agent with StackOne authentication
    agent = A2AAgent(
        url="https://a2a.stackone.com",
        memory=UnconstrainedMemory(),
        parameters=A2AAgentParameters(
            httpx_async_client=HttpxAsyncClientParameters(
                headers={
                    "Authorization": f"Basic {BASE64_API_KEY}",
                    "x-account-id": STACKONE_ACCOUNT_ID
                }
            )
        )
    )

    # Send a message and get the response
    response = await agent.run("List the first 5 employees")
    print(f"Agent response: {response.last_message.text}")

asyncio.run(main())

See Authentication Guide for details on obtaining your API key and account ID.

Architecture

BeeAI’s A2AAgent allows your application to communicate with remote A2A agents:

Complete Example with Event Handling

Here’s a complete example that handles events and debug information:

import asyncio
import base64
from beeai_framework.adapters.a2a.agents import A2AAgent, A2AAgentUpdateEvent, A2AAgentParameters, HttpxAsyncClientParameters
from beeai_framework.emitter import EventMeta
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory
from beeai_framework.errors import FrameworkError

# Configuration
STACKONE_API_KEY = "<stackone_api_key>"
STACKONE_ACCOUNT_ID = "<account_id>"
BASE64_API_KEY = base64.b64encode(f"{STACKONE_API_KEY}:".encode()).decode()

async def main():
    # Create A2A agent with StackOne authentication
    agent = A2AAgent(
        url="https://a2a.stackone.com",
        memory=UnconstrainedMemory(),
        parameters=A2AAgentParameters(
            httpx_async_client=HttpxAsyncClientParameters(
                headers={
                    "Authorization": f"Basic {BASE64_API_KEY}",
                    "x-account-id": STACKONE_ACCOUNT_ID
                }
            )
        )
    )

    # Define event handler for progress updates
    def handle_update(data: A2AAgentUpdateEvent, event: EventMeta) -> None:
        value = data.value
        debug_info = value[1] if isinstance(value, tuple) else value
        print(f"Agent progress: {debug_info}")

    # Run the agent with event handling
    try:
        response = await agent.run("List the first 5 employees").on("update", handle_update)
        print(f"\nAgent response: {response.last_message.text}")
    except FrameworkError as e:
        print(f"Error: {e.explain()}")

asyncio.run(main())

Interactive Chat Loop

Create an interactive chat session with the StackOne agent:

import asyncio
import base64
from beeai_framework.adapters.a2a.agents import A2AAgent, A2AAgentParameters, HttpxAsyncClientParameters
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory

STACKONE_API_KEY = "<stackone_api_key>"
STACKONE_ACCOUNT_ID = "<account_id>"
BASE64_API_KEY = base64.b64encode(f"{STACKONE_API_KEY}:".encode()).decode()

async def main():
    agent = A2AAgent(
        url="https://a2a.stackone.com",
        memory=UnconstrainedMemory(),
        parameters=A2AAgentParameters(
            httpx_async_client=HttpxAsyncClientParameters(
                headers={
                    "Authorization": f"Basic {BASE64_API_KEY}",
                    "x-account-id": STACKONE_ACCOUNT_ID
                }
            )
        )
    )

    print("Connected to StackOne A2A Agent. Type 'quit' to exit.\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() == "quit":
            break

        response = await agent.run(user_input)
        print(f"Agent: {response.last_message.text}\n")

asyncio.run(main())

Multiple StackOne Accounts

Connect to multiple StackOne accounts by creating separate A2AAgent instances:

import base64
from beeai_framework.adapters.a2a.agents import A2AAgent, A2AAgentParameters, HttpxAsyncClientParameters
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory

STACKONE_API_KEY = "<stackone_api_key>"
BASE64_API_KEY = base64.b64encode(f"{STACKONE_API_KEY}:".encode()).decode()

# Agent for HiBob integration
hibob_agent = A2AAgent(
    url="https://a2a.stackone.com",
    memory=UnconstrainedMemory(),
    parameters=A2AAgentParameters(
        httpx_async_client=HttpxAsyncClientParameters(
            headers={
                "Authorization": f"Basic {BASE64_API_KEY}",
                "x-account-id": "<hibob_account_id>"
            }
        )
    )
)

# Agent for BambooHR integration
bamboo_agent = A2AAgent(
    url="https://a2a.stackone.com",
    memory=UnconstrainedMemory(),
    parameters=A2AAgentParameters(
        httpx_async_client=HttpxAsyncClientParameters(
            headers={
                "Authorization": f"Basic {BASE64_API_KEY}",
                "x-account-id": "<bamboohr_account_id>"
            }
        )
    )
)

async def query_multiple_systems():
    # Query HiBob
    hibob_response = await hibob_agent.run("List all employees")
    print(f"HiBob: {hibob_response.last_message.text}")

    # Query BambooHR
    bamboo_response = await bamboo_agent.run("List all employees")
    print(f"BambooHR: {bamboo_response.last_message.text}")

Memory Management

BeeAI Framework provides memory implementations for conversation history:

from beeai_framework.adapters.a2a.agents import A2AAgent, A2AAgentParameters, HttpxAsyncClientParameters
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory

# UnconstrainedMemory: Keeps all messages (default choice)
agent = A2AAgent(
    url="https://a2a.stackone.com",
    memory=UnconstrainedMemory(),
    parameters=A2AAgentParameters(
        httpx_async_client=HttpxAsyncClientParameters(
            headers={
                "Authorization": f"Basic {BASE64_API_KEY}",
                "x-account-id": STACKONE_ACCOUNT_ID
            }
        )
    )
)

# Clear memory to start a new conversation
async def start_new_conversation():
    agent.memory = UnconstrainedMemory()  # Memory must be empty before setting

Best Practices

Use Appropriate Memory

Choose the right memory implementation for your use case:

from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory

agent = A2AAgent(
    url="https://a2a.stackone.com",
    memory=UnconstrainedMemory(),
    # ...
)

Use UnconstrainedMemory for typical interactions. Memory must be empty when assigned to an agent.

Handle Events for Debugging

Subscribe to update events for debugging and progress tracking:

def handle_update(data: A2AAgentUpdateEvent, event: EventMeta) -> None:
    print(f"Debug: {data.value}")

response = await agent.run("Query").on("update", handle_update)

This is especially useful for understanding agent behavior during development.

Error Handling

Wrap agent calls in try-except blocks to handle errors gracefully:

from beeai_framework.errors import FrameworkError
from beeai_framework.agents import AgentError

try:
    response = await agent.run("List employees")
except AgentError as e:
    print(f"Agent error: {e}")
    # Handle agent-specific errors
except FrameworkError as e:
    print(f"Framework error: {e.explain()}")
    # Handle framework errors

Context Management

Use context IDs and task IDs to manage multi-turn conversations:

# Start a new conversation
response = await agent.run(
    "List employees",
    clear_context=True  # Clears previous context
)

# Continue an existing conversation (uses stored context)
follow_up = await agent.run("Tell me more about the first employee")

Next Steps

Python SDK

Build custom A2A tools with the Python SDK

JavaScript SDK

Use the A2A JavaScript SDK

Authentication

Learn about authentication and security

Quickstart

Learn A2A basics with cURL