Agents Module

The agents module lets you invoke agents programmatically from workflows. Use agents.run() when the workflow needs the result, or agents.enqueue() when it should return after the run is accepted.

Import

from bifrost import agents

Methods

Method	Behavior	Returns
agents.run()	Waits for the agent to finish	Agent output
agents.enqueue()	Waits only for HTTP acceptance	AgentRunHandle
agents.get_run()	Reads the durable run record	AgentRun

All three methods are async. Awaiting agents.enqueue() waits for the short enqueue request, not for agent execution.

agents.enqueue()

Queue an agent run and return as soon as the API accepts it with HTTP 202.

async def enqueue(
    agent_name: str,
    input: dict | None = None,
    *,
    output_schema: dict | None = None,
) -> AgentRunHandle

Parameters

Parameter	Type	Description
agent_name	str	Name of the agent to run
input	dict	Structured input data for the agent
output_schema	dict	JSON Schema for the expected output

Returns

An AgentRunHandle containing:

Field	Type	Description
run_id	str	ID to pass to agents.get_run()
status	"queued"	Initial durable status

The run row is committed before queue publication, so a returned run_id is immediately queryable even if a worker has not claimed it yet.

Raises

• AgentPausedError — if the target agent is paused
• BifrostAPIError — if the API rejects the request, including an unknown agent or queue publication failure

Example

from bifrost import workflow, agents

@workflow
async def queue_ticket_triage(ticket_id: str):
    handle = await agents.enqueue(
        "ticket-classifier",
        input={"ticket_id": ticket_id},
        output_schema={
            "type": "object",
            "properties": {
                "priority": {"type": "string"},
                "category": {"type": "string"},
            },
        },
    )
    return {"run_id": handle.run_id, "status": handle.status}

agents.get_run()

Read the current status and result of an agent run.

async def get_run(run_id: str) -> AgentRun

Parameters

Parameter	Type	Description
run_id	str	ID returned by agents.enqueue()

Returns

An AgentRun with status, input, output, error, usage, timing, and caller metadata. Status progresses from queued to running, then to a terminal status such as completed, failed, timeout, cancelled, or budget_exceeded.

Raises

• ValueError — if the run does not exist
• PermissionError — if the caller cannot access the run

Polling example

import asyncio
from bifrost import agents

TERMINAL_STATUSES = {
    "completed",
    "failed",
    "timeout",
    "cancelled",
    "budget_exceeded",
}

async def wait_for_agent(run_id: str):
    while True:
        run = await agents.get_run(run_id)
        if run.status in TERMINAL_STATUSES:
            return run
        await asyncio.sleep(1)

agents.run()

Execute an agent autonomously and wait for the result.

async def run(
    agent_name: str,
    input: dict | None = None,
    *,
    output_schema: dict | None = None,
    timeout: int = 1800,
) -> dict | str

Parameters

Parameter	Type	Description
agent_name	str	Name of the agent to run
input	dict	Structured input data for the agent
output_schema	dict	JSON Schema for the expected output (agent will conform its response)
timeout	int	Maximum seconds to wait (default 1800 = 30 min)

Returns

• If output_schema is provided: dict matching the schema
• Otherwise: str (the agent’s final text response)

Raises

• RuntimeError — if the agent run fails
• BifrostAPIError — if the API rejects the request, including an unknown agent
• AgentPausedError — if the target agent is paused

Examples

from bifrost import workflow, agents

@workflow
async def classify_ticket(description: str):
    """Use an AI agent to classify a support ticket."""
    result = await agents.run(
        "ticket-classifier",
        input={"description": description},
        output_schema={
            "type": "object",
            "properties": {
                "priority": {"type": "string", "enum": ["low", "normal", "high", "urgent"]},
                "category": {"type": "string"},
                "summary": {"type": "string"}
            }
        },
        timeout=120,
    )
    return result  # {"priority": "high", "category": "billing", "summary": "..."}


@workflow
async def generate_report():
    """Use an agent to generate a natural language report."""
    report = await agents.run(
        "report-writer",
        input={"report_type": "weekly", "department": "engineering"},
    )
    return {"report": report}  # report is a string

How It Works

1. The API commits a durable run row with status queued.
2. The run context is published for a worker to claim.
3. agents.enqueue() returns the run ID; agents.run() keeps waiting.
4. The agent loads its system prompt, tools, and knowledge sources.
5. An LLM tool-calling loop executes: LLM → tool → LLM → …
6. The durable row is updated with the terminal status and result.

Budget Limits

Agent runs are bounded by two limits set on the agent configuration:

Limit	Description	Default	Max
max_iterations	Max LLM call cycles per run	50	200
max_token_budget	Max total tokens per run	100,000	1,000,000

If either limit is reached, the run completes with status budget_exceeded.

An optional per-agent max_run_timeout can also be configured in the UI to set a hard time limit on runs.

Delegation

When an agent has Delegated Agents configured, it automatically receives delegate_to_<agent_name> tools. During execution, the agent can delegate tasks to child agents:

# Parent agent automatically gets delegation tools like:
# delegate_to_child_agent(task: str) -> str
# These are called by the LLM, not by your workflow code

Delegation is bounded by:

• Max depth: 5 levels of nested delegation
• Timeout: 600 seconds per delegation call
• Child tracking: Each delegation creates a child AgentRun linked via parent_run_id

Cancellation and Rerun

Agent runs support cancellation and rerun via the API:

import requests

# Cancel a running agent
requests.post(f"{api_url}/api/agent-runs/{run_id}/cancel", headers=headers)

# Rerun a completed/failed/cancelled run
response = requests.post(f"{api_url}/api/agent-runs/{run_id}/rerun", headers=headers)
new_run = response.json()

See Also

• Agents and Chat — Agent configuration and chat
• AI Tools — Making workflows callable as agent tools
• AI Module — Direct LLM completions