Run subagents in the background

Start a child agent without waiting for it, keep the parent working, and read the report when it is ready. Tool restrictions on a child are not an operating-system sandbox.

Run it

From the repository root:

uv run python docs/_examples/async_subagent.py

The output includes:

Background task finished; child session is separate

This program uses deterministic offline models. It requires no API key and is executed by the documentation test suite.

How the model manages a task

AsyncSubAgentExtension registers five tools:

Tool	Purpose
task_start	Start a task from a configured profile; returns its task_id at once.
task_list	Show every task of the session with its status.
task_status	Show one task without its report; never waits.
task_output	Return the report, waiting up to wait_seconds when asked.
task_cancel	Stop a running task.

A finished task is announced before the parent’s next model call, or when the session’s next request starts, as an ordinary agent message. A task whose result the model already read is not announced again.

async def main() -> None:
    definition = SubAgentDefinition(
        name="audit",
        description="audit a code area and report what it finds.",
        system_prompt="You audit code and return a short report.",
        model=ChildModel(),
    )
    extension = AsyncSubAgentExtension([definition])
    client = await create_agent(
        ParentModel(),
        config=AgentRunConfig(session_id="parent"),
        extensions=[extension, ToolGuidelinesExtension()],
    )
    try:
        print((await client.run("Audit the login code")).content)
        [task] = extension.tasks("parent")
        assert task.status is SubAgentTaskStatus.COMPLETED and task.session_id != "parent"
        print("Background task finished; child session is separate")
    finally:
        await extension.close()

Customize it by subclassing

Override only what you need instead of copying the extension:

class TerseNotices(AsyncSubAgentExtension):
    def build_tools(self, context, subagents):
        # Drop cancellation for this deployment.
        return tuple(t for t in super().build_tools(context, subagents) if t.name != "task_cancel")

    def completion_notice(self, task):
        return f"{task.task_id} is {task.status.value}."

    async def task_finished(self, task):
        metrics.record(task.subagent_type, task.elapsed)

task_info shapes what each tool returns, task_started and task_finished observe the lifecycle (their errors are logged, never raised into the task), and _create_child changes how each child agent is built. Tasks live in memory for the lifetime of the extension object and are visible only to the session that started them. Call await extension.close() when the application shuts down.

Complete source

Download async_subagent.py.

"""Start a subagent in the background, keep working, and read its report later."""

import asyncio
import json
from collections.abc import AsyncIterator

from zett_agent.agent import AgentRunConfig
from zett_agent.client import create_agent
from zett_agent.extensions.async_subagent import AsyncSubAgentExtension, SubAgentTaskStatus
from zett_agent.extensions.subagent import SubAgentDefinition
from zett_agent.extensions.tool_guidelines import ToolGuidelinesExtension
from zett_agent.messages import AssistantMessage, ToolCall, ToolMessage
from zett_agent.model import ModelEvent, ModelRequest, ModelResponse, RetryOptions


class ParentModel:
    """Start a task, do other work, then wait for the report."""

    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        results = [message for message in request.messages if isinstance(message, ToolMessage)]
        if not results:
            call = ToolCall(
                "start-1",
                "task_start",
                {"description": "Audit auth", "prompt": "List the auth entry points", "subagent_type": "audit"},
            )
            message = AssistantMessage(tool_calls=(call,))
        elif len(results) == 1:
            # The task is running; ask for its report and wait up to ten seconds.
            task_id = json.loads(results[0].text)["task_id"]
            call = ToolCall("output-1", "task_output", {"task_id": task_id, "wait_seconds": 10})
            message = AssistantMessage(tool_calls=(call,))
        else:
            report = json.loads(results[-1].text)["content"]
            message = AssistantMessage(content=f"Background report: {report}")
        yield ModelEvent.completed(ModelResponse(message))


class ChildModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        await asyncio.sleep(0.05)  # stands in for real work the parent does not wait for
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content="login, logout, refresh")))


async def main() -> None:
    definition = SubAgentDefinition(
        name="audit",
        description="audit a code area and report what it finds.",
        system_prompt="You audit code and return a short report.",
        model=ChildModel(),
    )
    extension = AsyncSubAgentExtension([definition])
    client = await create_agent(
        ParentModel(),
        config=AgentRunConfig(session_id="parent"),
        extensions=[extension, ToolGuidelinesExtension()],
    )
    try:
        print((await client.run("Audit the login code")).content)
        [task] = extension.tasks("parent")
        assert task.status is SubAgentTaskStatus.COMPLETED and task.session_id != "parent"
        print("Background task finished; child session is separate")
    finally:
        await extension.close()


if __name__ == "__main__":
    asyncio.run(main())

Next: Run an isolated child agent for the blocking variant, or Compose built-in extensions to compare it with the other capabilities.