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.2

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.

Choosing Agents or Graph-Based Orchestration

Graph-based orchestration makes routing deterministic: after one defined step completes, the next defined step runs. It does not make an LLM's interpretation, tool selection, or output deterministic. Use a graph or Workflow when the business process is known, stable, and requires explicit control over every transition.

Start with an Agent when the work requires the system to decide how to solve an open-ended request: researching, interpreting documents, selecting tools, synthesizing results, or handling long-tail exceptions. A single agent can complete many such multi-step tasks without encoding every anticipated reasoning path as graph infrastructure.

Keep deterministic controls at operational boundaries. Validate structured output, apply policy checks and approvals, use idempotent writes, and emit audit records before an agent can produce external side effects. This separates deterministic process control from the model's inherently probabilistic reasoning, while allowing a workflow to be introduced where a stable business process truly needs one.

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