Agents

This documentation is outdated. The platform is currently being upgraded to the latest version of ACP with major changes. Updated docs are expected in ~2 weeks. Apologies for the inconvenience.

Agents are implemented in BeeAI as stateless functions that process input objects according to an input schema and return results in the output schema format. During execution, agents have the ability to stream progress.

The registration of agents is facilitated through the Agent Communication Protocol SDK.

Logic for agents can be implemented using any framework, language, or runtime. For example:

Agent schema

Each agent has input and output schemas based on the beeai-sdk package:

In ACP Alpha, beeai-sdk is meging into a single library acp-sdk.

class Input(BaseModel, extra="allow"):
    config: Config | None = None

class Output(BaseModel, extra="allow"):
    logs: list[Log | None] = Field(default_factory=list)

The config input supports additional configuration, like a list of tools.
The logs field allows streaming progress details (e.g., thoughts, actions)

Streaming

Agents can use the ctx object to stream progress during execution. The progress schema is a subset of the output schema, typically using the logs field:

@server.agent(...)
async def run_langgraph_agent(input: TextInput, ctx: Context) -> TextOutput:
    inputs = SummaryStateInput(research_topic=input.text)
    output = None
    async for event in graph.astream(inputs, stream_mode="updates"):
        log = Log(message=f"🚶‍♂️{key}: {value}" for key, value in event.items())
        output = event
        await ctx.report_agent_run_progress(
            delta=TextOutput(logs=[None, log], text="")
        )
    output = output.get("finalize_summary", {}).get("running_summary", None)
    return TextOutput(text=str(output))

Standardized agent interfaces

Agents can define their own input/output schemas or opt into standardized schemas for automatic UI features.

Chat agents

Chat agents use the chat interface, which requires the MessageInput and MessageOutput schemas. They must declare this interface in their metadata.

In ACP Pre-alpha, chat agents are stateless and need to re-create any persistent state, like memory, on each invocation. ACP Alpha is introducing stateful agents.

import asyncio

from acp.server.highlevel import Context, Server
from beeai_sdk.providers.agent import run_agent_provider
from beeai_sdk.schemas.message import (
    MessageInput,
    MessageOutput,
    AssistantMessage
)
from beeai_sdk.schemas.metadata import (
    Metadata,
    Examples,
    UiDefinition,
    UiType
)

def main():
    server = Server("chat-agents")
    @server.agent(
        "chat-example",
        "Chat with the user",
        input=MessageInput, # Chat input schema
        output=MessageOutput, # Chat output schema
        **Metadata(
            framework="LangGraph",
            license="Apache 2.0",
            languages=["Python"],
            examples=Examples(cli=[...]),
            ui=UiDefinition(
                type=UiType.chat, # Hands off UI type
                userGreeting="Hello, how can I help you?"
            ),
            fullDescription="Chat with the user"
        ).model_dump(),
    )
    def run_chat(input: MessageInput, ctx: Context) -> MessageOutput:
        assistant_message = AssistantMessage(
            role="assistant",
            content=f"Hello {input.messages[0].content}"
        )
        return MessageOutput(messages=[assistant_message])

    asyncio.run(run_agent_provider(server))

Hands-off agents

Hands-off agents support tasks that don’t require continuous interaction. They use the TextInput and TextOutput schemas and declare their type as UiType.hands_off.

import asyncio

from acp.server.highlevel import Context, Server
from beeai_sdk.providers.agent import run_agent_provider
from beeai_sdk.schemas.text import TextInput, TextOutput
from beeai_sdk.schemas.metadata import (
    Metadata,
    Examples,
    UiDefinition,
    UiType
)
def main():
    server = Server("chat-agents")
    @server.agent(
        "chat-example",
        "Chat with the user",
        input=TextInput, # Text input schema
        output=TextOutput, # Text output schema
        **Metadata(
            framework="LangGraph",
            license="Apache 2.0",
            languages=["Python"],
            examples=Examples(cli=[...]),
            ui=UiDefinition(
                type=UiType.hands_off, # Chat UI type
                userGreeting="Hello, how can I help you?"
            ),
            fullDescription="Chat with the user"
        ).model_dump(),
    )
    def run_chat(input: TextInput, ctx: Context) -> TextOutput:
        return TextOutput(text=f"Hello {input}!")
    asyncio.run(run_agent_provider(server))

Introduction

How To

Core Concepts

Agent Communication Protocol

Observability & Telemetry

Agents

Agent schema

Streaming

Standardized agent interfaces

Chat agents

Hands-off agents

Introduction

How To

Core Concepts

Agents

Agent Communication Protocol

Observability & Telemetry

​Agent schema

​Streaming

​Standardized agent interfaces

​Chat agents

​Hands-off agents

Agent schema

Streaming

Standardized agent interfaces

Chat agents

Hands-off agents