Agent 作为工具允许一个聊天 agent 将另一个具备聊天能力的子 agent 作为工作的一部分进行调度。子 agent 是真实的子 agent,拥有独立的 Durable Object 存储、消息、工具、可恢复流和深入查看 URL。父 agent 维护小型运行注册表,供客户端渲染子时间线、刷新后重放以及后续清理。
Agent 作为工具支持 @cloudflare/think agent 与 AIChatAgent 子类。AIChatAgent 子 agent 通过 saveMessages() 无界面运行,因此应使用服务端工具。在 agent-tool 轮次期间,浏览器提供的客户端工具不可用,除非你将该交互建模为服务端状态或由父 agent 中介的独立工作流。
当父代码需要对特定子级进行直接流式 RPC,且你的代码负责转发、取消与重放策略时,使用 subAgent(...).chat()。
当父模型或 workflow 将工作委派给子 agent,且需要保留子运行、事件重放、中止桥接与 UI 深入查看时,使用 agentTool() 或 runAgentTool()。Think 特定的轮次选择请参阅选择轮次 API。
当父模型应决定何时调用辅助 agent 时,使用 agentTool()。
import { Think } from "@cloudflare/think";
import { agentTool } from "agents/agent-tools";
import { z } from "zod";
export class Researcher extends Think {
getSystemPrompt() {
return "Research the user's topic and end with a concise summary.";
}
}
export class Assistant extends Think {
getTools() {
return {
research: agentTool(Researcher, {
description: "Research one topic in depth.",
displayName: "Researcher",
inputSchema: z.object({
query: z.string().min(3),
}),
}),
};
}
}import { Think } from "@cloudflare/think";
import { agentTool } from "agents/agent-tools";
import { z } from "zod";
export class Researcher extends Think<Env> {
getSystemPrompt() {
return "Research the user's topic and end with a concise summary.";
}
}
export class Assistant extends Think<Env> {
getTools() {
return {
research: agentTool(Researcher, {
description: "Research one topic in depth.",
displayName: "Researcher",
inputSchema: z.object({
query: z.string().min(3),
}),
}),
};
}
}子 agent 也可以是 AIChatAgent:
import { AIChatAgent } from "@cloudflare/ai-chat";
import { agentTool } from "agents/agent-tools";
import { convertToModelMessages, stepCountIs, streamText } from "ai";
import { z } from "zod";
export class Summarizer extends AIChatAgent {
formatAgentToolInput(input, request) {
return {
id: `agent-tool-${request.runId}-input`,
role: "user",
parts: [{ type: "text", text: `Summarize:\n\n${input.text}` }],
};
}
async onChatMessage() {
const result = streamText({
model: this.env.MODEL,
messages: await convertToModelMessages(this.messages),
});
return result.toUIMessageStreamResponse();
}
}
export class Assistant extends AIChatAgent {
async onChatMessage() {
const result = streamText({
model: this.env.MODEL,
messages: await convertToModelMessages(this.messages),
tools: {
summarize: agentTool(Summarizer, {
description: "Summarize long text in a separate retained agent.",
inputSchema: z.object({ text: z.string() }),
}),
},
stopWhen: stepCountIs(5),
});
return result.toUIMessageStreamResponse();
}
}import { AIChatAgent } from "@cloudflare/ai-chat";
import { agentTool } from "agents/agent-tools";
import { convertToModelMessages, stepCountIs, streamText } from "ai";
import { z } from "zod";
export class Summarizer extends AIChatAgent<Env> {
protected override formatAgentToolInput(input: { text: string }, request) {
return {
id: `agent-tool-${request.runId}-input`,
role: "user",
parts: [{ type: "text", text: `Summarize:\n\n${input.text}` }],
};
}
async onChatMessage() {
const result = streamText({
model: this.env.MODEL,
messages: await convertToModelMessages(this.messages),
});
return result.toUIMessageStreamResponse();
}
}
export class Assistant extends AIChatAgent<Env> {
async onChatMessage() {
const result = streamText({
model: this.env.MODEL,
messages: await convertToModelMessages(this.messages),
tools: {
summarize: agentTool(Summarizer, {
description: "Summarize long text in a separate retained agent.",
inputSchema: z.object({ text: z.string() }),
}),
},
stopWhen: stepCountIs(5),
});
return result.toUIMessageStreamResponse();
}
}生成的工具调用 this.runAgentTool(ChildAgent, ...),在父 WebSocket 上流式传输 agent-tool-event 帧,并将子 agent 摘要返回给父模型。若运行失败、中止或中断,工具返回结构化的 AgentToolFailure,而非空的成功值:
type AgentToolFailure = {
ok: false;
status: "error" | "aborted" | "interrupted";
error: string; // human-readable, safe to surface
retryable: boolean;
// Present only when `status` is "interrupted":
reason?: AgentToolInterruptedReason;
childStillRunning?: boolean;
};
type AgentToolInterruptedReason =
| "no-progress"
| "window-exceeded"
| "not-tailable"
| "inspect-timeout"
| "inspect-failed"
| "recovery-deadline"
| "budget-exceeded";retryable 仅对 interrupted 运行为 true——子 agent 被部署或父 agent 恢复重置或取代,且从未达到逻辑结果,因此重新派发相同调用可能成功。真正的 error 或有意 aborted 为 retryable: false。这使父提示词约定或编排工具可重试瞬时中断,而非向用户报告为最终失败。AgentToolFailure 从 agents 导出。
在 interrupted 运行中,reason 提供机器可读原因,childStillRunning 报告父 agent 停止等待时子 agent 是否仍在工作(true),或是否已被拆除(false)。应基于这些字段分支,而非解析 error 文本——例如,对 no-progress 中断重新派发(子 agent 可能自行恢复),但对 window-exceeded 中断重新连接或展示(子 agent 已被拆除)。reason 与 childStillRunning 也会镜像到 agent-tool-event 线帧和 useAgentToolEvents() 运行状态。
对于不做面向用户 assistant 文本、而是 workflow 式工作的 Think 子 agent,可覆盖 getAgentToolOutput(),必要时覆盖 getAgentToolSummary()。存在 assistant 文本时仍默认为摘要,但 Think agent-tool 运行可在不发出文本分块的情况下成功完成。
在子轮次结束前持久化任何结构化输出,因为 getAgentToolOutput() 在 saveMessages() 解析后立即读取。保持 getAgentToolSummary() 简洁以便展示;完整结构化值单独存储为工具输出。
export class Extractor extends Think {
getAgentToolOutput(runId) {
const rows = this.sql`
SELECT result_json FROM extraction_runs WHERE id = ${runId}
`;
return rows[0] ? JSON.parse(rows[0].result_json) : undefined;
}
getAgentToolSummary(_runId, output) {
return output ? "Extraction complete" : "";
}
}export class Extractor extends Think<Env> {
protected override getAgentToolOutput(runId: string) {
const rows = this.sql<{ result_json: string }>`
SELECT result_json FROM extraction_runs WHERE id = ${runId}
`;
return rows[0] ? JSON.parse(rows[0].result_json) : undefined;
}
protected override getAgentToolSummary(_runId: string, output: unknown) {
return output ? "Extraction complete" : "";
}
}对确定性 workflow、定时任务、HTTP 处理程序或扇出代码,使用 runAgentTool()。
const [a, b] = await Promise.allSettled([
this.runAgentTool(Researcher, {
input: { query: "HTTP/3" },
parentToolCallId: toolCallId,
displayOrder: 0,
}),
this.runAgentTool(Researcher, {
input: { query: "gRPC" },
parentToolCallId: toolCallId,
displayOrder: 1,
}),
]);const [a, b] = await Promise.allSettled([
this.runAgentTool(Researcher, {
input: { query: "HTTP/3" },
parentToolCallId: toolCallId,
displayOrder: 0,
}),
this.runAgentTool(Researcher, {
input: { query: "gRPC" },
parentToolCallId: toolCallId,
displayOrder: 1,
}),
]);runAgentTool() 按 runId 幂等。传入相同 runId 不会启动重复的子轮次。已完成、失败、中止和中断的运行会保留,直到显式清除。
默认 runAgentTool() 等待子 agent 到达终态后返回。对于不想阻塞派发轮次的长时工作——大型导入、视频渲染、深度研究——传入 detached。运行被派发,当前轮次继续,runAgentTool() 立即返回句柄:
type DetachedRunAgentToolResult = {
runId: string;
agentType: string;
status: "running" | "error"; // "error" only if dispatch itself was rejected
};detached: true 为即发即忘——通过 agent-tool-event 帧(与 useAgentToolEvents() 消费相同)和全局 onAgentToolFinish() 钩子观察运行。传入对象可配置有针对性、持久化的完成回调:
export class Importer extends Think {
async startImport(input) {
const { runId } = await this.runAgentTool(ImportAgent, {
input,
detached: { onFinish: "onImportDone", maxBudgetMs: 60 * 60 * 1000 },
});
return runId;
}
// Fires once, even if the Durable Object was evicted and rehydrated while the
// child ran. Referenced by METHOD NAME (like schedule()) — never a closure,
// which cannot survive eviction.
async onImportDone(run, result) {
switch (result.status) {
case "completed":
await this.markImportReady(run.runId, result.summary);
break;
case "error":
await this.markImportFailed(run.runId, result.error);
break;
case "interrupted":
// reason "budget-exceeded" ⇒ the run hit its maxBudgetMs ceiling.
// interrupted is soft: a child that finishes anyway re-fires this
// hook with "completed", so make the handler idempotent.
break;
}
}
}export class Importer extends Think<Env> {
async startImport(input: ImportInput) {
const { runId } = await this.runAgentTool(ImportAgent, {
input,
detached: { onFinish: "onImportDone", maxBudgetMs: 60 * 60 * 1000 },
});
return runId;
}
// Fires once, even if the Durable Object was evicted and rehydrated while the
// child ran. Referenced by METHOD NAME (like schedule()) — never a closure,
// which cannot survive eviction.
async onImportDone(run: AgentToolRunInfo, result: AgentToolLifecycleResult) {
switch (result.status) {
case "completed":
await this.markImportReady(run.runId, result.summary);
break;
case "error":
await this.markImportFailed(run.runId, result.error);
break;
case "interrupted":
// reason "budget-exceeded" ⇒ the run hit its maxBudgetMs ceiling.
// interrupted is soft: a child that finishes anyway re-fires this
// hook with "completed", so make the handler idempotent.
break;
}
}
}关键行为:
- 持久化完成。 交付在驱逐和部署后仍有效:热路径在 isolate 存活时低延迟交付,自调度协调骨干最终化热路径遗漏的内容。正常路径为恰好一次;崩溃下为至少一次,因此
onFinish处理程序必须幂等。 - 放弃与完成相互独立。 预算放弃以
status: "interrupted"、reason: "budget-exceeded"交付。因interrupted为软状态,放弃后仍完成的子 agent 会以真实结果再次触发onFinish——过早放弃不会隐藏延迟完成。 - 有界。 每个分离运行有绝对
maxBudgetMs上限(按运行,或detachedMaxBudgetMs静态选项;默认 24 小时)。过期后父 agent 停止监视并拆除子 agent,避免被遗弃的运行永久占用maxConcurrentAgentTools槽位。 - 不继承 signal。 分离运行必须比派发轮次存活更久,因此不继承
options.signal。需显式取消:
await this.cancelAgentTool(runId); // 幂等;以 onFinish "aborted" 交付await this.cancelAgentTool(runId); // 幂等;以 onFinish "aborted" 交付在聊天 agent(@cloudflare/think 或 AIChatAgent)上,通常希望模型_响应_已完成的后台运行。无需手动配置 onFinish,传入 notify: true——运行完成时 agent 向聊天注入消息(按运行 + 状态幂等,恰好一次完成不会重复),模型在上下文中携带结果进行下一轮次:
await this.runAgentTool(ResearchAgent, { input, detached: { notify: true } });await this.runAgentTool(ResearchAgent, { input, detached: { notify: true } });若应用按 metadata.source 路由或隐藏合成消息,可传入自定义来源:
await this.runAgentTool(ResearchAgent, {
input,
detached: { notify: { source: "research-background" } },
});await this.runAgentTool(ResearchAgent, {
input,
detached: { notify: { source: "research-background" } },
});覆盖 formatDetachedCompletion(run, result) 可自定义注入文本,或对特定结果返回空字符串以抑制通知。显式 onFinish 优先于 notify。
子 agent 的 inspectAgentToolRun(runId) 返回运行当前状态快照,或 null。null 不表示「失败」——表示子 agent 尚无该运行记录。这在派发后立即出现是正常的(子 agent 可能仍在持久化首行),也是刚重新水合的子 agent 在惰性协调过时的 running 行之前返回的内容。调用方——以及框架自身的协调骨干——将 null 视为「未达终态,在预算内继续监视」,而非终态失败。仅非 null 且 status 为终态(completed / error / aborted)的检查才会最终化运行。
作为 agent 工具运行的子 agent——无论等待还是分离——可报告运行中进度,供父 agent 渲染实时状态行、服务端计量运行,或在运行完成前响应命名检查点。在子 agent 内部调用 reportProgress()(例如从工具的 execute):
export class ImportAgent extends Think {
getTools() {
return {
ingest: tool({
inputSchema: z.object({ url: z.string() }),
execute: async ({ url }) => {
// Ephemeral progress: drives a generic bar / phase / status line.
await this.reportProgress({
fraction: 0.6,
phase: "ingesting",
message: "Ingested 40k/80k rows",
});
// ...
},
}),
};
}
}export class ImportAgent extends Think<Env> {
getTools() {
return {
ingest: tool({
inputSchema: z.object({ url: z.string() }),
execute: async ({ url }) => {
// Ephemeral progress: drives a generic bar / phase / status line.
await this.reportProgress({
fraction: 0.6,
phase: "ingesting",
message: "Ingested 40k/80k rows",
});
// ...
},
}),
};
}
}reportProgress() 在聊天 agent(@cloudflare/think 与 AIChatAgent)上可用。在基础 Agent 类上及在非 agent-tool 运行外调用时为无操作并输出开发警告,因此相同子 agent 代码可安全独立运行。框架从当前轮次解析活跃运行——无需传递运行 ID。
reportProgress<T>(
progress: {
fraction?: number; // 0..1 — drives a progress bar
message?: string; // human-readable status line
phase?: string; // coarse phase label, e.g. "ingesting"
milestone?: string; // present ⇒ a durable milestone (see below)
data?: T; // app-specific payload; live-only unless persisted
},
options?: { persist?: boolean },
): Promise<void>;瞬时信号通过子 agent 自身轮次流作为瞬时 data-agent-progress 部分传输,因此会重新广播到父 agent 已连接客户端,并通过 useAgentToolEvents() 出现在 AgentToolRunState.progress 上——后台运行托盘可渲染实时进度条、阶段和状态行,无需深入查看。突发信号会被合并(后者胜出;fraction >= 1 帧始终刷新)。data 字段仅实时有效,除非传入 { persist: true }。
覆盖 onProgress() 可在服务端计量、引导或展示进度。每当子 agent 进度信号经父 agent 转发时尽力触发,适用于等待与分离运行:
export class Assistant extends Think {
async onProgress(run, progress) {
if (progress.milestone) {
// 持久化里程碑已到达——据此分支。
}
console.log(run.runId, progress.phase, progress.fraction);
}
}export class Assistant extends Think<Env> {
override async onProgress(
run: AgentToolRunInfo,
progress: AgentToolProgressSnapshot,
) {
if (progress.milestone) {
// 持久化里程碑已到达——据此分支。
}
console.log(run.runId, progress.phase, progress.fraction);
}
}onProgress() 非持久化:驱逐后分离运行的最新快照从协调时的 inspectAgentToolRun().progress 重建,而非重新触发钩子。最新快照也会持久化在子运行行上,因此重新水合后的父 agent 可回答「该运行进展如何」,而无需跟踪实时流。
命名 milestone 将信号从瞬时层提升为持久化层——仍只有一个发出方法:
await this.reportProgress({
milestone: "sources-gathered",
data: { sources: 2 },
});await this.reportProgress({
milestone: "sources-gathered",
data: { sources: 2 },
});里程碑作为子 agent 上一行持久化,带有单调递增的每运行 sequence,并以持久化 data-agent-milestone 部分传输(与瞬时进度不同)。因此可在驱逐后存活、在深入查看时重放,并按 sequence 去重后出现在 AgentToolRunState.milestones 与 inspectAgentToolRun().milestones 上。onProgress() 对里程碑同样触发,且 progress.milestone 已设置,消费者可区分里程碑与瞬时进度。
对聊天 agent 上的分离运行,detached: { onMilestones } 在配置的里程碑到达时(运行完成_之前_)展示聊天消息。每个 (runId, name) 最多触发一次——无论实时观察还是驱逐后协调——因此确定性 ID 将热路径与冷路径交付合并为至多一次:
// "narrate"(默认):注入合成 assistant 状态行——不触发 model turn。
await this.runAgentTool(Researcher, {
input,
detached: { onMilestones: ["sources-gathered"] },
});
// "react":发布 user 角色 turn 让 model 响应(引导、启动依赖工作)。消耗一次 model turn。
await this.runAgentTool(Researcher, {
input,
detached: { onMilestones: { names: ["needs-approval"], mode: "react" } },
});// "narrate"(默认):注入合成 assistant 状态行——不触发 model turn。
await this.runAgentTool(Researcher, {
input,
detached: { onMilestones: ["sources-gathered"] },
});
// "react":发布 user 角色 turn 让 model 响应(引导、启动依赖工作)。消耗一次 model turn。
await this.runAgentTool(Researcher, {
input,
detached: { onMilestones: { names: ["needs-approval"], mode: "react" } },
});覆盖 formatDetachedMilestone(run, milestone) 可自定义措辞,或对特定里程碑返回空字符串以抑制。合成叙述消息携带 metadata.source,客户端可将其渲染为 agent 事件而非人类轮次。
分离子 agent 至少报告一次信号后,若运行随后静默超过 detachedNoProgressBudgetMs(默认 1 小时;按运行通过 detached: { noProgressBudgetMs } 覆盖),协调骨干会放弃。表现为 status: "interrupted"、reason: "no-progress"。从未报告的子 agent 仅受绝对 detachedMaxBudgetMs 上限约束——运行不会因慢而被放弃。将 noProgressBudgetMs 设为 0 或 Infinity 可禁用该重置窗口。
useAgentToolEvents() 是无头钩子。它订阅现有父连接、去重重放/实时竞争、将子 UIMessageChunk 正文应用到消息部分,并按父工具调用 ID 分组兄弟运行。每个运行状态携带 progress 与 milestones,后台运行托盘可渲染实时进度条、阶段和里程碑标记,无需深入查看。
import { useAgent, useAgentToolEvents } from "agents/react";
import { useAgentChat } from "@cloudflare/ai-chat/react";
const agent = useAgent({ agent: "Assistant", name: userId });
const { messages } = useAgentChat({ agent });
const agentTools = useAgentToolEvents({ agent });
for (const message of messages) {
for (const part of message.parts) {
if (part.type === "tool-call") {
const runs = agentTools.getRunsForToolCall(part.toolCallId);
// 在此 tool call 旁渲染子运行。
}
}
}import { useAgent, useAgentToolEvents } from "agents/react";
import { useAgentChat } from "@cloudflare/ai-chat/react";
const agent = useAgent({ agent: "Assistant", name: userId });
const { messages } = useAgentChat({ agent });
const agentTools = useAgentToolEvents({ agent });
for (const message of messages) {
for (const part of message.parts) {
if (part.type === "tool-call") {
const runs = agentTools.getRunsForToolCall(part.toolCallId);
// 在此 tool call 旁渲染子运行。
}
}
}无父工具调用的命令式运行可通过 agentTools.unboundRuns 访问。
Agent 作为工具即为普通子 agent。通过父路由连接保留的子 agent:
useAgent({
agent: "Assistant",
name: userId,
sub: [{ agent: "Researcher", name: runId }],
});useAgent({
agent: "Assistant",
name: userId,
sub: [{ agent: "Researcher", name: runId }],
});通过父注册表控制外部访问,防止猜测运行 ID 创建新的子 facet:
override async onBeforeSubAgent(_request, child) {
if (!this.hasAgentToolRun(child.className, child.name)) {
return new Response("Not found", { status: 404 });
}
}运行与子 facet 默认保留,供刷新、深入查看与后续检查。清除聊天历史或应用自定义保留策略时显式删除:
await this.clearAgentToolRuns();
await this.clearAgentToolRuns({
status: ["completed", "error", "aborted", "interrupted"],
});
await this.clearAgentToolRuns({ olderThan: Date.now() - 7 * 24 * 60 * 60_000 });await this.clearAgentToolRuns();
await this.clearAgentToolRuns({
status: ["completed", "error", "aborted", "interrupted"],
});
await this.clearAgentToolRuns({ olderThan: Date.now() - 7 * 24 * 60 * 60_000 });若保留的运行仍为 starting 或 running,清理会在删除 facet 前取消子 agent。
Agent-tool 运行保留在父 agent 中。若父 agent 在子运行仍为 starting 或 running 时重启(部署或驱逐),不会放弃子 agent。启动恢复会重新挂接到活跃子 agent 并跟踪其流直至终态结果。子 agent 作为子 agent 拥有自身 chatRecovery,在父 agent 转发输出的同时自行修复中断轮次。已完成的子 agent 直接最终化,无需重新执行已完成工作。
重新挂接等待以进度为键,而非固定墙钟。两个静态 options 可调:
| 选项 | 默认值 | 行为 |
|---|---|---|
agentToolReattachNoProgressTimeoutMs |
120000(2 分钟) |
在无转发进度时父 agent 等待多久后放弃。每收到转发分块重置,因此流式传输的子 agent 可一直跟随至终态。 |
agentToolReattachMaxWindowMs |
Infinity |
单次重新挂接的可选硬墙钟上限。默认无上限(镜像聊天恢复的 maxRecoveryWork),健康的长时子 agent 不会被截断。设有限值可施加上限。 |
放弃结果映射到 AgentToolFailure 字段:
- 子 agent 在整个无进度窗口内静默时,封存为
reason: "no-progress"、childStillRunning: true。该封存为软状态:子 agent 继续运行,因此重新派发相同runId可重新挂接并在自行恢复后收集结果。 - 若设置有限
agentToolReattachMaxWindowMs且触发,运行封存为reason: "window-exceeded"、childStillRunning: false,子 agent 被拆除(已用完完整窗口,视为耗尽)。 - 无法跟踪或检查的子 agent,或超过整体恢复截止时间的子 agent,以对应
reason封存,使父工具调用返回结构化失败而非无限挂起。
挂起的子 agent 不会永久阻塞恢复。无进度预算约束静默子 agent。内容失控由子 agent 自身 chatRecovery(maxRecoveryWork 与 shouldKeepRecovering)约束,而非仅父 agent 计时器。
通过 agentTool 可观测性通道监视父协调:
import { subscribe } from "agents/observability";
const unsubscribe = subscribe("agentTool", (event) => {
if (event.type === "agent_tool:recovery:row") {
console.log("Recovered agent-tool row", event.payload);
}
});import { subscribe } from "agents/observability";
const unsubscribe = subscribe("agentTool", (event) => {
if (event.type === "agent_tool:recovery:row") {
console.log("Recovered agent-tool row", event.payload);
}
});原始 diagnostics_channel 订阅者应使用通道名 agents:agent_tool。