在后台运行子智能体

启动一个子智能体而不等待它，让父智能体继续工作，等报告准备好后再读取。对子智能体的工具限制并不是操作系统层面的沙箱。

运行示例

在仓库根目录执行：

uv run python docs/_examples/async_subagent.py

输出包含：

Background task finished; child session is separate

这个程序使用确定性的离线模型，不需要 API 密钥，并由文档测试套件执行。

模型如何管理任务

AsyncSubAgentExtension 会注册五个工具：

工具	用途
task_start	用已配置的子智能体配置启动一个任务，并立即返回它的 task_id。
task_list	列出本会话的所有任务及其状态。
task_status	查看单个任务的状态，不含报告；从不等待。
task_output	返回报告；需要时最多等待 wait_seconds 秒。
task_cancel	停止一个正在运行的任务。

任务结束后，会在父智能体下一次调用模型之前，或会话的下一个请求开始时，以普通智能体消息的形式通知它。模型已经读取过结果的任务不会再次通知。

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()

通过子类化来定制

只重写你需要的部分，而不是复制整个扩展：

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 决定每个工具返回的内容，task_started 和 task_finished 用来观察生命周期（它们的错误只会被记录，不会抛进任务里），_create_child 则改变每个子智能体的构建方式。任务在扩展对象的生命周期内保存在内存中，并且只对启动它们的会话可见。应用关闭时请调用 await extension.close()。

完整源码

下载 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())

下一步：阻塞式的版本见 运行一个隔离的子智能体，或参阅 组合内置扩展 与其他能力对比。