Agentbyte

Agentbyte is an observability-first agentic AI framework for building and studying multiagent systems with a learning-first, implementation-oriented workflow.

Current release: 0.26.1

Building an Agent

Every example below builds on the same domain: a customer support agent. Start with a plain agent and a model client — no tools, no middleware:

from agentbyte import Agent
from agentbyte.llm import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient.from_api_key(model="gpt-4.1-mini")

support_agent = Agent(
    name="support_agent",
    description="Answers customer support questions about orders and shipping.",
    instructions="You are a helpful support agent. Be concise and accurate.",
    model_client=model_client,
)

response = await support_agent.run("Where is my order #1234?")

print(response.final_message.content)
print(response.usage)          # tokens, cost, cache hits
print(response.finish_reason)  # "stop" | "max_iterations" | ...

run() executes to completion and returns one AgentResponse. For live progress — token-by-token streaming, tool calls as they happen — use run_stream() instead, which yields events and finishes with that same AgentResponse:

async for item in support_agent.run_stream("Where is my order #1234?", verbose=True):
    print(item)

Adding Tools

A support agent is only useful once it can look things up. Turn any function into a tool with @tool; gate risky ones with approval_mode:

from agentbyte import Agent
from agentbyte.tools import ApprovalMode, tool

@tool
def get_order_status(order_id: str) -> str:
    """Look up the status of an order."""
    return f"Order {order_id} is in transit."

@tool(approval_mode=ApprovalMode.ALWAYS)
def issue_refund(order_id: str, amount: float) -> str:
    """Issue a refund for an order (requires human approval)."""
    return f"Refunded ${amount} for order {order_id}"

support_agent = Agent(
    name="support_agent",
    description="Answers order/shipping questions and can issue refunds.",
    instructions="Look up orders before answering. Refunds always need approval.",
    model_client=model_client,
    tools=[get_order_status, issue_refund],
)

Agentbyte also ships ready-made tools you can drop in without writing any code — pass them straight into tools=[...]:

• Core: ThinkTool, TaskStatusTool, CalculatorTool, DateTimeTool, JSONParserTool, RegexTool (all at once via create_core_tools())
• Coding: ReadFileTool, WriteFileTool, ListDirectoryTool, GrepSearchTool, BashExecuteTool, PythonREPLTool
• Memory: MemoryTool — lets the agent read/write a memory backend (ListMemory or FileMemory) as a tool call, on top of the automatic context injection every agent already gets

Agentic Workflows

When support handling is a fixed multi-step pipeline rather than a single agent call — triage, then route — model it as a Workflow instead:

import asyncio
from pydantic import BaseModel
from agentbyte.workflow import FunctionStep, StepMetadata, Workflow, WorkflowConfig, WorkflowRunner

class TicketInput(BaseModel):
    text: str

class TriagedTicket(BaseModel):
    text: str
    priority: str

async def triage(input_data: TicketInput, context) -> TriagedTicket:
    priority = "high" if "urgent" in input_data.text.lower() else "normal"
    return TriagedTicket(text=input_data.text, priority=priority)

async def route(input_data: TriagedTicket, context) -> TriagedTicket:
    return input_data  # e.g. assign to a queue here

workflow = Workflow(WorkflowConfig(name="support_ticket_pipeline"))
workflow.chain(
    FunctionStep("triage", StepMetadata(name="triage"), TicketInput, TriagedTicket, triage),
    FunctionStep("route", StepMetadata(name="route"), TriagedTicket, TriagedTicket, route),
)

execution = asyncio.run(
    WorkflowRunner().run(workflow, {"text": "urgent: order not received"})
)
print(execution.state["route_output"])

Steps can also wrap an agent (AgentStep), call HTTP endpoints (HttpStep), transform data (TransformStep), or nest another workflow (SubWorkflowStep) — with conditional routing, parallel branches, checkpoint/resume, and human-in-the-loop suspend/resume.

Orchestrator Patterns

Two different ways to combine multiple agents — pick based on who's in control:

• AgentAsTool — one agent decides if and when to delegate. The support agent stays the single decision-maker and calls the billing agent like any other tool.
• An orchestrator (e.g. RoundRobinOrchestrator) — a separate controller drives the conversation between agents in turns, until a termination condition fires. Neither agent decides when the other speaks.

Agent as a tool — the support agent delegates billing questions:

from agentbyte import Agent
from agentbyte.agents import AgentAsTool

billing_agent = Agent(
    name="billing_agent",
    description="Handles billing and refund questions.",
    instructions="Answer billing questions and process refund requests.",
    model_client=model_client,
)

support_agent = Agent(
    name="support_agent",
    description="Front-line support agent that can delegate billing issues.",
    instructions="Handle general support; delegate billing questions to the billing tool.",
    model_client=model_client,
    tools=[AgentAsTool(agent=billing_agent)],
)

Orchestrated turns — support and escalation agents collaborate until the ticket is resolved:

from agentbyte import (
    MaxMessageTermination,
    RoundRobinOrchestrator,
    TextMentionTermination,
    UserMessage,
)

orchestrator = RoundRobinOrchestrator(
    agents=[support_agent, escalation_agent],
    termination=TextMentionTermination("RESOLVED") | MaxMessageTermination(6),
)

task = UserMessage(content="Customer says their package never arrived.", source="user")
async for item in orchestrator.run_stream(task, verbose=True):
    print(item)

Other orchestrators follow the same run()/run_stream() shape: AIOrchestrator (a model picks the next speaker), HandoffOrchestrator (agents explicitly hand off control), PlanBasedOrchestrator (a plan is drafted, then executed step by step).

Middleware

Built in: LoggingMiddleware, PIIRedactionMiddleware, GuardrailMiddleware, MetricsMiddleware, RateLimitMiddleware, ApprovalMiddleware, ContextCompactionMiddleware, RetryMiddleware, OTelMiddleware. Attach any combination to the same support agent via middlewares=[...]:

from agentbyte import Agent
from agentbyte.middleware import ApprovalMiddleware, LoggingMiddleware, RateLimitMiddleware

support_agent = Agent(
    name="support_agent",
    description="Answers order/shipping questions and can issue refunds.",
    instructions="Look up orders before answering. Refunds always need approval.",
    model_client=model_client,
    tools=[get_order_status, issue_refund],
    middlewares=[
        LoggingMiddleware(),
        RateLimitMiddleware(max_requests=10, window_seconds=60),
        ApprovalMiddleware(tool_names=["issue_refund"]),
    ],
)

Observability-First Telemetry

Agentbyte exposes two complementary telemetry layers via OTelMiddleware:

• Per-call spans (chat <model>, tool <name>, embedding <model>) for model/tool/embedding-level diagnostics.
• Task-level root span (agent <name>) wrapping every per-call span in one run, carrying the final aggregated usage and outcome.

Enable telemetry:

export AGENTBYTE_ENABLE_OTEL=true

Per-call span attributes emitted by OTelMiddleware:

• gen_ai.system, gen_ai.operation.name, gen_ai.agent.name, gen_ai.session.id
• gen_ai.request.model, gen_ai.response.model, gen_ai.response.finish_reason
• gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.usage.total_tokens, gen_ai.usage.cost_estimate_usd
• gen_ai.tool.name, gen_ai.tool.success
• gen_ai.embedding.input_count, gen_ai.embedding.output_count
• gen_ai.input.messages, gen_ai.output.messages, gen_ai.tool.parameters, gen_ai.tool.result (opt-in content capture — only when AGENTBYTE_OTEL_CAPTURE_CONTENT=true, since these can carry PII)

Reading a trace: chat gpt-4.1-mini spans show per-call usage/cost/finish reason; the wrapping agent <name> span shows the final accumulated usage and task outcome for the whole run()/run_stream() call.

Installation

Python requirement: 3.11+

uv sync --all-groups

Optional extras:

uv sync --extra openai
uv sync --extra azureopenai
uv sync --extra otel
uv sync --extra webui

Install in another project (pip / uv add)

Use extras to enable provider + telemetry support:

pip install "agentbyte[azureopenai,otel]"

uv add "agentbyte[azureopenai,otel]"

For the browser WebUI:

pip install "agentbyte[webui]"
# or
uv add "agentbyte[webui]"

Install all optional features:

pip install "agentbyte[all]"
# or
uv add "agentbyte[all]"

Note: the Azure extra is azureopenai.

Run The WebUI

Option 1: Run the preset-backed app

This is the easiest way to see the WebUI working end to end with real preset entities:

• preset agents
• preset orchestrators
• preset workflow

Step 1. Install the WebUI extra:

uv sync --extra webui

Step 2. Start the preset-backed app:

uv run python examples/webui/presets_webui.py

Step 3. Open the browser:

http://127.0.0.1:8080

If auto-open is enabled in your environment, the browser may open automatically.

Option 2: Run the WebUI against your current project directory

Use this when you want Agentbyte to scan a directory for exported agent, workflow, or orchestrator objects.

Important: discovery is convention-based. The scanned directory must contain Python modules that expose top-level variables literally named agent, workflow, or orchestrator. If you point --dir at a folder that does not export those names, the UI will load but show No entities found.

Step 1. Install the WebUI extra:

uv sync --extra webui

Step 2. Launch the WebUI and scan the current directory:

uv run agentbyte webui --dir .

Step 3. Open the browser:

http://127.0.0.1:8080

Useful variants:

uv run agentbyte webui --dir . --port 8080 --host 127.0.0.1 --no-open
uv run agentbyte webui --dir examples --port 8090

For this repository, the most reliable first-run path is the preset-backed launcher:

uv run python examples/webui/presets_webui.py

Use agentbyte webui --dir ... when you have a directory of exportable demo modules, for example:

# my_entities.py
agent = ...
workflow = ...
orchestrator = ...

Option 3: Run it programmatically

Use this when you want to serve in-memory entities directly from Python.

from agentbyte.webui import serve

serve(entities=[agent], port=8080, auto_open=True)

Quick Troubleshooting

If the app does not start:

uv sync --extra webui

If port 8080 is already in use:

uv run agentbyte webui --dir . --port 8090

If you do not want the browser to open automatically:

uv run agentbyte webui --dir . --no-open

Development

uv run ruff check src tests
uv run pytest tests -v