AsyncSubAgentExtension#

Build extensions, choose hooks, and manage request-scoped state. See the guide.

class zett_agent.extensions.async_subagent.AsyncSubAgentExtension(subagents=None, *, max_concurrent=4, max_retained=100, notify_parent=True)[源代码]#

Run subagents in the background and let the model manage them with tools.

The foreground SubAgentExtension blocks its parent until one report arrives. This extension returns at once and registers five tools instead:

task_start

Start a task from a configured subagent; returns its task_id.

task_list

Show every task of the session with its status.

task_status

Show one task without its report.

task_output

Return the report, optionally waiting up to wait_seconds for it.

task_cancel

Stop a running task.

Children are ordinary isolated agents with their own session, built from the same SubAgentDefinition objects the foreground extension uses. Tasks live in memory for the lifetime of this extension object and are visible only to the session that started them. The event loop that started a task must keep running for it to finish.

A finished task is announced to its parent at the next opportunity: before the next model call of a request that is still running, or when the session's next request starts. The notice is an ordinary AgentMessage, so it needs no budget of internal continuations and can never fail the parent's request. A task whose result the model already read through a task tool is not announced again. An application that wants to wake an idle parent publishes its own InternalMessageEvent from task_finished().

参数:

  • subagents (Sequence[SubAgentDefinition] | None) -- Profiles the parent may choose; defaults to the built-in reasoning and explore profiles on the parent's model.

  • max_concurrent (int) -- Running tasks allowed per session at once.

  • max_retained (int) -- Finished tasks kept per session for later inspection; the oldest finished tasks are forgotten first.

  • notify_parent (bool) -- Announce finished tasks to the parent.

示例

Offer background delegation next to the foreground tool:

extensions = [SubAgentExtension(), AsyncSubAgentExtension(), ToolGuidelinesExtension()]

Change what a finished task tells the parent:

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

备注

Customize through the overridable methods instead of copying the extension: build_tools() chooses the tools, task_info() shapes what a tool returns, completion_notice() writes the notification, task_started() and task_finished() observe the lifecycle, and _create_child() builds each child agent.

警告

Children run with the tools and permissions their definition grants. Background work keeps running after the request that started it ended, so call close() when the application shuts down.

async on_tool(context)[源代码]#

Register the task tools for this request.

参数:

context (AgentRunContext)

返回类型:

None

async on_state(context)[源代码]#

Announce tasks that finished while no request was active.

参数:

context (AgentRunContext)

返回类型:

None

async before_model(context, request)[源代码]#

Announce tasks that finished since the previous model call of this request.

参数:
返回类型:

None

tasks(session_id)[源代码]#

Return the tasks one session started, oldest first.

参数:

session_id (str)

返回类型:

tuple[SubAgentTask, ...]

get_task(session_id, task_id)[源代码]#

Return one task of a session.

抛出:

ValueError -- If the session has no such task. The message lists the known identifiers so the model can correct itself.

参数:

  • session_id (str)

  • task_id (str)

返回类型:

SubAgentTask

async start(context, definition, *, description, prompt)[源代码]#

Start one background task and return it while it runs.

抛出:

ValueError -- If the session already runs max_concurrent tasks.

参数:
返回类型:

SubAgentTask

async wait(task, timeout=None)[源代码]#

Wait until a task finishes or timeout seconds pass, then return it.

Cancelling the caller does not cancel the task.

参数:
返回类型:

SubAgentTask

async cancel(task)[源代码]#

Cancel a running task and wait for it to settle; a finished task is unchanged.

参数:

task (SubAgentTask)

返回类型:

SubAgentTask

async close()[源代码]#

Cancel every running task of every session and wait for them to settle.

返回类型:

None

build_tools(context, subagents)[源代码]#

Return the tools to register for one request.

Override to drop a tool, add one, or replace one, calling super().build_tools(...) for the rest.

参数:
返回类型:

tuple[AgentTool, ...]

task_info(task, *, include_output=False)[源代码]#

Describe one task for the model.

Override to add detail. include_output is true only for the tool that returns a report, so listings stay short.

参数:
返回类型:

SubAgentTaskInfo

completion_notice(task)[源代码]#

Return the message that tells the parent a task finished.

Return None to stay silent for this task; the model can still find it with task_list.

参数:

task (SubAgentTask)

返回类型:

str | None

async task_started(task)[源代码]#

Observe a task that just started; the default does nothing.

参数:

task (SubAgentTask)

返回类型:

None

async task_finished(task)[源代码]#

Observe a task that reached a terminal status; the default does nothing.

Its errors are logged, never raised into the task. This is the place to publish an InternalMessageEvent to a parent that should wake up instead of waiting for its next turn.

参数:

task (SubAgentTask)

返回类型:

None

accept(config, event)#

Handle one external event and report whether it was accepted.

The Agent broadcasts each event to every registered extension. Override this synchronous hook when an extension waits for input from another thread, HTTP request, TUI, or Web UI. The default ignores the event. Config is passed separately and can be None before initialization. Each extension decides whether the supplied identity and payload match its protocol. Return True to include this extension's name in the broadcast result; returning False does not stop later receivers.

参数:
返回类型:

bool

async after_model(context, response)#

Run after the complete assistant message is appended.

参数:
返回类型:

None

async after_run(context, result)#

Run after a successful final answer is appended.

参数:
返回类型:

None

async after_tool(context, call, result, error)#

Inspect or modify a tool result before it is appended.

error is the original execution exception for a failed result and None after successful execution. Changes made to result are included in context, persistence, and the next model request. Raising prevents the result from being appended.

参数:
返回类型:

None

async after_turn(context, result)#

Run once the turn's assistant message and tool results are settled.

result is the assistant message that opened the turn: either the final answer, or the tool-calling message whose ToolMessages were appended. Every requested call is settled first, including calls that failed, and calls that steering skipped, which arrive as unsuccessful or skipped ToolMessages.

Steering that interrupts a turn is accepted while the runtime settles that batch, so its UserMessage is already appended when this hook runs and the steering turn follows as the next turn.

A turn that never completed, because the request failed or was cancelled, has no after_turn call; use on_error and RunCancelledEvent for those paths.

参数:
返回类型:

None

async before_run(context)#

Run after the transformed user input is appended and published.

参数:

context (AgentRunContext)

返回类型:

None

async before_tool(context, call)#

Run before one requested tool is invoked.

参数:
返回类型:

None

async before_turn(context)#

Run before this turn's model call.

Runs after before_run() on the first turn and after the previous turn's after_turn() on later ones. Model request assembly, compaction, and before_model() follow this hook, so context edited here reaches the provider.

参数:

context (AgentRunContext)

返回类型:

None

property name: str#

Unique registration identity, defaulting to the concrete class name.

Override with a class attribute or assign an instance name to register independently configured instances of the same extension type.

async on_compact(context)#

Summarize restored context when a caller asks for compaction.

Agent.compact restores the session through on_tool and on_state exactly as a request does, appends no input message, and makes no primary model call: the only work a compaction pass does is what this hook performs. The default does nothing, so an extension with nothing to compact pays nothing.

示例

Usage:

async def on_compact(self, context):
    await self.compact(context)
参数:

context (AgentRunContext)

返回类型:

None

async on_error(context, error)#

Run before an agent error is propagated to the caller.

参数:
返回类型:

None

async on_event(context, event)#

Process a published notification; inspect its concrete type with match.

参数:
返回类型:

None

async on_message(context)#

Transform context.input_message after state restoration, before append.

Replace the pending UserMessage to implement externally selected modes. Do not append or persist it here: the runtime appends the final message exactly once after all input hooks finish. before_run sees that message through AgentRunContext.add_message() or AgentRunContext.replace_messages(), and persistence subscribers have already received it.

参数:

context (AgentRunContext)

返回类型:

None

async on_model_request(context, request, call_next)#

Wrap one provider request and its complete streamed response.

Pass request (or a dataclass replacement) to call_next. The default closes the inner stream so cancellation reaches the provider.

参数:
返回类型:

AsyncIterator[ModelEvent]

async on_success(context, result)#

Run once after all after_run hooks succeed, before RUN_COMPLETED.

This callback observes a completed request, not an intermediate model step. Failed or cancelled requests do not trigger it. Callback errors propagate through on_error and prevent RUN_COMPLETED from being emitted.

示例

Usage:

async def on_success(self, context, result):
    await save_answer(context.session_id, result.content)
参数:
返回类型:

None

async on_tool_call(context, call, call_next)#

Wrap one registered local tool handler and return its ToolResult.

An extension may adjust call.arguments, return without calling the next layer, transform its result, or raise into normal tool-failure handling.

参数:
返回类型:

ToolResult

property pending: bool#

Return whether this extension is participating in an active request.

The Agent updates this state for every registered extension, including extensions whose hooks are currently idle. One extension instance may be shared by multiple Agents and sessions; pending remains true until all requests using that instance have finished, failed, or been cancelled. Persistent data retained between requests does not count.