Sub-agents

agents=[...] on agent() gives a parent agent a pool of specialized children it can call. Flux appends each child’s name and description to the system prompt and injects a delegate tool. The LLM calls delegate to dispatch work; the result comes back as a structured dict.

Declare sub-agents

Each sub-agent is a regular agent() task with a name and a description. The parent reads both when choosing whether and where to delegate.

from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent

@task
async def search_web(query: str) -> str:
    """Search the web for information about the query."""
    ...

@workflow
async def review_workflow(ctx: ExecutionContext):
    researcher = await agent(
        "You are a thorough research specialist. Use search_web to gather "
        "information, then synthesize your findings into a clear summary.",
        model="ollama/qwen3",
        name="researcher",
        description="Deep research using web sources. Delegate when gathering "
                    "and synthesizing information from multiple sources.",
        tools=[search_web],
    )

    reviewer = await agent(
        "You are a technical reviewer. Evaluate quality, accuracy, and "
        "completeness of research summaries. Provide constructive feedback.",
        model="ollama/qwen3",
        name="reviewer",
        description="Reviews and critiques research output for quality and completeness.",
    )

    manager = await agent(
        "You are a senior engineering manager. Coordinate your team:\n"
        "1. Delegate research to the researcher agent.\n"
        "2. Send the output to the reviewer for feedback.\n"
        "3. Produce a final summary combining both perspectives.",
        model="ollama/qwen3",
        agents=[researcher, reviewer],
        max_tool_calls=10,
    )

    return await manager(f"Research and review: {ctx.input['topic']}")

The description controls when the parent delegates. Write it from the parent’s perspective: describe the situation that warrants the handoff, not just what the agent is capable of.

What happens at construction time

When agents is non-empty, agent() appends a ## Sub-Agents section to the system prompt listing each agent’s name and description, plus instructions for using delegate. It also creates a delegate @task and adds it to the tool list.

The parent LLM never sees sub-agent implementations. It routes based on the names and descriptions you provide.

The delegate tool

The injected delegate tool has these parameters:

Sub-agents start with a blank context. They cannot see the parent’s conversation history. Pass all necessary context inside instruction or input.

Delegation results

Every delegation returns a DelegationResult:

@dataclass
class DelegationResult:
    agent: str
    status: Literal["completed", "paused", "failed"]
    output: Any
    execution_id: str | None = None

The parent LLM receives this as a JSON dict and can inspect status to decide whether to continue, retry with a different agent, or surface an error. Delegation errors never raise exceptions. An unknown agent name or a sub-agent crash both come back as status="failed" with output describing what went wrong.

Workflow agents

Use workflow_agent() to delegate to a remote Flux workflow instead of an in-process agent. It is a synchronous factory, so no await:

from flux import workflow, ExecutionContext
from flux.tasks.ai import agent, workflow_agent

@workflow
async def release_workflow(ctx: ExecutionContext):
    deployer = workflow_agent(
        name="deployer",
        description="Handles deployment pipelines. May pause for human approval.",
        workflow="deploy_pipeline",
    )

    manager = await agent(
        "You are a release manager. Delegate deployment to the deployer agent. "
        "If a deployment pauses for approval, review the details and resume "
        "with your decision.",
        model="ollama/qwen3",
        agents=[deployer],
        max_tool_calls=10,
    )

    service = ctx.input.get("service", "api-gateway")
    version = ctx.input.get("version", "1.0.0")
    return await manager(f"Deploy {service} version {version} to production")

workflow_agent() takes three arguments: name, description, and workflow (the registered Flux workflow name). It uses FluxClient internally to call the remote workflow via run_workflow_sync or resume_execution_sync.

When a remote workflow pauses, the DelegationResult comes back with status="paused" and an execution_id. The parent LLM can resume by calling delegate again with the same agent name and that execution_id.

Mix local and workflow agents

Local agent() tasks and workflow_agent() instances go in the same agents=[...] list. The delegate tool dispatches to whichever type is registered under each name:

manager = await agent(
    "You are a senior engineering manager...",
    model="ollama/qwen3",
    agents=[researcher, reviewer, deployer],  # local + workflow agents together
    max_tool_calls=15,
)

Where researcher and reviewer are local agent() tasks and deployer is a workflow_agent(). See examples/ai/sub_agents_mixed.py for the full working example.

Nested hierarchies

Agents can have their own sub-agents, forming a tree. Each agent only sees its direct children. The manager delegates to the analyst; the analyst can delegate to the researcher.

researcher = await agent(
    "You are a research specialist.",
    model="ollama/qwen3",
    name="researcher",
    description="Gathers information from web sources.",
    tools=[search_web],
)

analyst = await agent(
    "You are a data analyst.",
    model="ollama/qwen3",
    name="analyst",
    description="Analyzes data and produces structured reports. Delegates "
                "web research to the researcher when raw data is needed.",
    agents=[researcher],
)

manager = await agent(
    "You are a project manager. Coordinate analysis work.",
    model="ollama/qwen3",
    agents=[analyst],
)

Validation

Flux validates all sub-agents at parent construction time and raises AgentValidationError before the workflow runs.

Agent names must be lowercase, use only letters, numbers, and single hyphens, must not start or end with a hyphen, and must not exceed 64 characters. Each sub-agent must be callable with non-empty name and description attributes. All names within one agents=[...] list must be unique.

# Missing description — fails at construction
broken = await agent("...", model="ollama/qwen3", name="broken")
await agent("...", model="ollama/qwen3", agents=[broken])
# AgentValidationError: Sub-agent 'broken' must have a non-empty description attribute.

# Duplicate names — fails at construction
agent_a = await agent("...", model="ollama/qwen3", name="worker",
                      description="Does work.")
await agent("...", model="ollama/qwen3", agents=[agent_a, agent_a])
# AgentValidationError: Duplicate agent name: 'worker'

Observability

Delegation shows up in the Flux event log as nested task events. Because delegate is a standard @task, each call gets its own started/completed event and the sub-agent’s tool calls nest inside:

TASK_STARTED    manager      {"instruction": "Research and review: async programming"}
TASK_STARTED    delegate     {"agent": "researcher", "instruction": "..."}
TASK_STARTED    researcher   {"instruction": "..."}
TASK_STARTED    search_web   {"query": "async programming in Python"}
TASK_COMPLETED  search_web   "Results: ..."
TASK_COMPLETED  researcher   "Research findings: ..."
TASK_COMPLETED  delegate     {"agent": "researcher", "status": "completed", ...}
TASK_COMPLETED  manager      "Final review: ..."

Next steps

• Planning — give the parent agent a structured plan before it begins delegating.
• Tool approval — gate any delegate call behind human approval.
• Agents in workflows — understand how agents checkpoint and replay within a Flux execution.