サブエージェント RPC とプログラムからのターン

Think はトップレベルエージェントとしても、サブエージェントとしても使えます。サブエージェントとして使うとき、chat() メソッドはターン全体を実行し、コールバック経由でイベントをストリームします。

べき等な再試行とあとからの状態確認を伴う耐久的な受付はプログラムからの送信を参照してください。退避後のリカバリーは耐久リカバリーを参照してください。

chat

async chat(
	userMessage: string | UIMessage,
	callback: StreamCallback,
	options?: ChatOptions,
): Promise<void>

StreamCallback

メソッド	発火タイミング
`onStart(event)`	作業開始前。キャンセル用のリクエスト ID を公開します
`onEvent(json)`	各ストリーミングチャンク（JSON シリアライズされた `UIMessageChunk`）
`onDone()`	ターン完了後、アシスタントメッセージが永続化されたあと
`onError(message)`	ターン中のエラー
`onInterrupted()`	任意。試行が中断され、あとから別 isolate で動くスケジュール済み継続が最終結果を担当します。完了でも、終端エラーでもありません。デフォルトは no-op です

chat() 起点のターンが中断して復旧するとき、onInterrupted が重要です。RPC の Promise は正常に解決します（isolate はまだ生きています）。正常解決だけを成功とみなすと、途中までストリームした内容を確定してしまいます。扱い方は「未完了、失敗でもない。継続側が答えを持つ」です。部分結果を確定せず、チャネルを開いたままにする、復旧中の状態を出す、再接続する、のいずれかです。デプロイや退避による中断は、これが発火する前に isolate を落とします（呼び出し側はトランスポート切断を見ます）。onInterrupted が扱うのは、isolate 内でのストールからリカバリーへ入る経路です。

ChatOptions

フィールド	説明
`signal`	ストリーム途中でターンをキャンセルする `AbortSignal`

ツールは子エージェント側のものです。耐久的な能力は、子の getTools()、拡張、MCP ツール、またはクライアントツールスキーマで定義します。chat() に options.tools を渡す古い呼び出し元には警告が出て、値は無視されます。

例: 親から子を呼ぶ

import { Think } from "@cloudflare/think";

export class ParentAgent extends Think {
	getModel() {
		/* ... */
	}

	async delegateToChild(task) {
		const child = await this.subAgent(ChildAgent, "child-1");

		const chunks = [];
		await child.chat(task, {
			onStart: (event) => {
				console.log("Child started:", event.requestId);
			},
			onEvent: (json) => {
				chunks.push(json);
			},
			onDone: () => {
				console.log("Child completed");
			},
			onError: (error) => {
				console.error("Child failed:", error);
			},
		});

		return chunks;
	}
}

export class ChildAgent extends Think {
	getModel() {
		/* ... */
	}

	getSystemPrompt() {
		return "You are a research assistant. Analyze data and report findings.";
	}
}

import { Think } from "@cloudflare/think";
import type { StreamCallback } from "@cloudflare/think";

export class ParentAgent extends Think<Env> {
	getModel() {
		/* ... */
	}

	async delegateToChild(task: string) {
		const child = await this.subAgent(ChildAgent, "child-1");

		const chunks: string[] = [];
		await child.chat(task, {
			onStart: (event) => {
				console.log("Child started:", event.requestId);
			},
			onEvent: (json) => {
				chunks.push(json);
			},
			onDone: () => {
				console.log("Child completed");
			},
			onError: (error) => {
				console.error("Child failed:", error);
			},
		});

		return chunks;
	}
}

export class ChildAgent extends Think<Env> {
	getModel() {
		/* ... */
	}

	getSystemPrompt() {
		return "You are a research assistant. Analyze data and report findings.";
	}
}

サブエージェントのターンをキャンセルする

サブエージェント境界をまたぐ RPC 安全なキャンセルには、onStart と cancelChat() を使います。

let requestId;

const callback = {
	onStart(event) {
		requestId = event.requestId;
	},
	onEvent(json) {
		// Forward stream chunks.
	},
	onDone() {},
	onError(error) {
		console.error(error);
	},
};

const turn = child.chat("Long analysis task", callback);

// Later, from another RPC call or failure handler:
if (requestId) {
	await child.cancelChat(requestId, "client disconnected");
}

await turn;

let requestId: string | undefined;

const callback: StreamCallback = {
	onStart(event) {
		requestId = event.requestId;
	},
	onEvent(json) {
		// Forward stream chunks.
	},
	onDone() {},
	onError(error) {
		console.error(error);
	},
};

const turn = child.chat("Long analysis task", callback);

// Later, from another RPC call or failure handler:
if (requestId) {
	await child.cancelChat(requestId, "client disconnected");
}

await turn;

呼び出し元と呼び出し先が Workers RPC で分かれていない場合は、ストリーム途中のキャンセルに AbortSignal も渡せます。

const controller = new AbortController();
setTimeout(() => controller.abort(), 30_000);

await child.chat("Long analysis task", callback, {
	signal: controller.signal,
});

const controller = new AbortController();
setTimeout(() => controller.abort(), 30_000);

await child.chat("Long analysis task", callback, {
	signal: controller.signal,
});

ターンがすでに完了している、またはリクエスト ID が不明な場合、cancelChat(requestId, reason?) は no-op です。中止しても、途中までのアシスタントメッセージは永続化されます。

saveMessages

メッセージを注入し、WebSocket 接続なしでモデルターンを起動します。スケジュール応答、webhook 起点のターン、プロアクティブエージェント、onChatResponse からの連鎖に使います。

async saveMessages(
	messages:
		| UIMessage[]
		| ((current: UIMessage[]) => UIMessage[] | Promise<UIMessage[]>),
	options?: SaveMessagesOptions,
): Promise<SaveMessagesResult>

{ requestId, status, error? } を返します。status は "completed"、"error"、"skipped"、"aborted" のいずれかです。

`status`	条件
`"completed"`	ターンが完了するまで実行されました。
`"error"`	ターンは始まったが、ストリームがエラーを報告しました。取得できる場合、`error` にストリームのエラーメッセージが入ります。
`"skipped"`	ターンが途中で無効になりました。例は `chat-clear` です。ユーザーメッセージは永続化され、モデルは走りません。
`"aborted"`	`options.signal` または `chat-request-cancel` で完了前にキャンセルされました。途中までのアシスタントチャンクは永続化されます。

プログラムターンを開始した Durable Object からキャンセルするには options.signal を渡します。AbortSignal は Durable Object の RPC 境界を越えられません。シグナルはハイバネーションをまたいで永続化されません。

静的メッセージ

await this.saveMessages([
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Time for your daily summary." }],
	},
]);

await this.saveMessages([
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Time for your daily summary." }],
	},
]);

関数形式

複数の saveMessages 呼び出しがキューに並ぶと、関数形式はターン開始時点の最新メッセージで実行されます。

await this.saveMessages((current) => [
	...current,
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Continue your analysis." }],
	},
]);

await this.saveMessages((current) => [
	...current,
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Continue your analysis." }],
	},
]);

スケジュール応答

getScheduledTasks() で定期プロンプトターンを起動します。

export class MyAgent extends Think {
	getModel() {
		/* ... */
	}

	getScheduledTasks() {
		return {
			dailyReport: {
				schedule: "every day at 09:00",
				timezone: "UTC",
				prompt: "Generate the daily report.",
			},
		};
	}
}

export class MyAgent extends Think<Env> {
	getModel() {
		/* ... */
	}

	getScheduledTasks() {
		return {
			dailyReport: {
				schedule: "every day at 09:00",
				timezone: "UTC",
				prompt: "Generate the daily report.",
			},
		};
	}
}

onChatResponse からの連鎖

現在のターン完了後に、後続ターンを開始します。

async onChatResponse(result: ChatResponseResult) {
	if (result.status === "completed" && this.needsFollowUp(result.message)) {
		await this.saveMessages([{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: "Now summarize what you found." }],
		}]);
	}
}

continueLastTurn

新しいユーザーメッセージを注入せず、最新のアシスタントメッセージのあとに、もう一度モデル呼び出しを実行します。Think は結果を continuation: true 付きの新しいアシスタントメッセージとして永続化します。既存のアシスタントメッセージへチャンクを追記しません。

protected async continueLastTurn(
	body?: Record<string, unknown>,
	options?: SaveMessagesOptions,
): Promise<SaveMessagesResult>

最後のメッセージがアシスタントメッセージでない場合は { requestId, status: "skipped" } を返します。任意の body パラメーターは、この継続で保存済みの body を上書きします。実行中の継続をキャンセルするには options.signal を渡します。

abortRequest と abortAllRequests

Durable Object 内部から、進行中のチャットターンをキャンセルします。

protected abortRequest(requestId: string, reason?: unknown): void
protected abortAllRequests(): void

リクエスト ID が分かっているときは abortRequest() を使います。今動いているターンをキャンセルすべき単一目的のヘルパーでは abortAllRequests() を使います。呼び出し元でシグナルを渡せるプログラムターンでは、SaveMessagesOptions.signal を優先します。