INTEGRITY Dokumentace

← Vytvářejte agenty na Cloudflare / agents / runtime / operations / observability

Diagnostické kanály

Agents publikují strukturované události do diagnostické kanály pro každou významnou operaci: volání RPC, změny stavu, spouštění plánu, přechody workflow, připojení MCP a další. Publikování nemá žádnou režii, pokud nikdo neposlouchá.

Struktura událostí

Každá událost má tato pole:

{
  type: "rpc",                        // what happened
  agent: "MyAgent",                   // which agent class emitted it
  name: "user-123",                   // which agent instance (Durable Object name)
  payload: { method: "getWeather" },  // details
  timestamp: 1758005142787            // when (ms since epoch)
}

agent a name identifikují zdrojového agenta: agent je název třídy a name je název instance Durable Object.

Kanály

Události se směrují do pojmenovaných kanálů podle svého typu:

Kanál Typy událostí Popis
agents:state state:update Události synchronizace stavu
agents:rpc rpc, rpc:error Volání metod RPC a jejich selhání
agents:message message:request, message:response, message:clear, message:cancel, message:error, tool:result, tool:approval, submission:create, submission:status, submission:error Životní cyklus odeslání zprávy chatu, nástroje a Think
agents:chat chat:request:failed, chat:recovery:*, chat:stream:stalled, chat:context:compacted Životní cyklus požadavku chatu, obnovení, zaseknutí streamu a komprimace kontextu
agents:transcript chat:transcript:repaired Události opravy transkriptu
agents:fiber fiber:run:*, fiber:recovery:* Životní cyklus trvalého fiberu
agents:agent_tool agent_tool:recovery:* Recovery agent-tool mezi nadřazeným a podřízeným agentem
agents:schedule schedule:create, schedule:execute, schedule:cancel, schedule:retry, schedule:error, schedule:duplicate_warning, queue:create, queue:retry, queue:error Životní cyklus naplánovaných a zařazených úloh
agents:lifecycle connect, disconnect, destroy Připojení a ukončení agenta
agents:workflow workflow:start, workflow:event, workflow:approved, workflow:rejected, workflow:terminated, workflow:paused, workflow:resumed, workflow:restarted Přechody stavů workflow
agents:mcp mcp:client:preconnect, mcp:client:connect, mcp:client:authorize, mcp:client:discover Operace klienta MCP
agents:email email:receive, email:reply, email:send Zpracování e-mailu

Přihlašování k odběru událostí

Typovaný pomocník pro odběr

subscribe() funkce z agents/observability poskytuje typově bezpečný přístup k událostem na konkrétním kanálu:

import { subscribe } from "agents/observability";

const unsub = subscribe("rpc", (event) => {
	if (event.type === "rpc") {
		console.log(`RPC call: ${event.payload.method}`);
	}
	if (event.type === "rpc:error") {
		console.error(
			`RPC failed: ${event.payload.method} — ${event.payload.error}`,
		);
	}
});

// Clean up when done
unsub();
import { subscribe } from "agents/observability";

const unsub = subscribe("rpc", (event) => {
	if (event.type === "rpc") {
		console.log(`RPC call: ${event.payload.method}`);
	}
	if (event.type === "rpc:error") {
		console.error(
			`RPC failed: ${event.payload.method} — ${event.payload.error}`,
		);
	}
});

// Clean up when done
unsub();

Callback je plně typovaný: event se zúží jen na typy událostí, které tímto kanálem procházejí.

Typovaný helper používá klíče v camelCase, takže obnovení agent-tool je subscribe("agentTool", ...). Odběratelé surového diagnostického kanálu by měli použít vydaný název kanálu, agents:agent_tool.

Surový diagnostics_channel

Můžete se také přihlásit k odběru přímo pomocí Node.js API:

import { subscribe } from "node:diagnostics_channel";

subscribe("agents:schedule", (event) => {
	console.log(event);
});
import { subscribe } from "node:diagnostics_channel";

subscribe("agents:schedule", (event) => {
	console.log(event);
});

Tail Workers (produkce)

V produkčním prostředí se všechny zprávy diagnostického kanálu automaticky přeposílají do Tail Workers. V samotném agentovi není potřeba žádný kód pro odběr, stačí připojit Tail Worker a přistupovat k událostem přes event.diagnosticsChannelEvents:

export default {
	async tail(events) {
		for (const event of events) {
			for (const msg of event.diagnosticsChannelEvents) {
				// msg.channel is "agents:rpc", "agents:workflow", etc.
				// msg.message is the typed event payload
				console.log(msg.timestamp, msg.channel, msg.message);
			}
		}
	},
};
export default {
	async tail(events) {
		for (const event of events) {
			for (const msg of event.diagnosticsChannelEvents) {
				// msg.channel is "agents:rpc", "agents:workflow", etc.
				// msg.message is the typed event payload
				console.log(msg.timestamp, msg.channel, msg.message);
			}
		}
	},
};

Díky tomu získáte strukturovanou, filtrovatelnou observabilitu v produkci bez jakékoli režie v hot path agenta.

Vlastní observabilita

Výchozí implementaci můžete přepsat poskytnutím vlastní Observability rozhraní:

import { Agent } from "agents";

const myObservability = {
	emit(event) {
		// Send to your logging service, filter events, etc.
		if (event.type === "rpc:error") {
			console.error(event.payload.method, event.payload.error);
		}
	},
};

class MyAgent extends Agent {
	observability = myObservability;
}
import { Agent } from "agents";
import type { Observability } from "agents/observability";

const myObservability: Observability = {
	emit(event) {
		// Send to your logging service, filter events, etc.
		if (event.type === "rpc:error") {
			console.error(event.payload.method, event.payload.error);
		}
	},
};

class MyAgent extends Agent {
	override observability = myObservability;
}

Nastavte observability na undefined k vypnutí veškerého vysílání událostí:

import { Agent } from "agents";

class MyAgent extends Agent {
	observability = undefined;
}
import { Agent } from "agents";

class MyAgent extends Agent {
	override observability = undefined;
}

Reference událostí

Události RPC

Typ Payload Kdy
rpc { method, streaming? } A @callable metoda se vyvolá
rpc:error { method, error } A @callable metoda vyvolá výjimku

Stavové události

Typ Payload Kdy
state:update {} setState() se volá

Události zpráv, nástrojů a odeslání

Tyto události sledují životní cyklus chatových zpráv, interakce nástrojů na straně klienta a trvalá odeslání Think.

Typ Payload Kdy
message:request {} Chatová zpráva je přijata
message:response {} Stream chatové odpovědi se dokončí
message:clear {} Historie chatu se vymaže
message:cancel { requestId } Streamovaný požadavek je zrušen
message:error { error } Chatový stream selže
tool:result { toolCallId, toolName } Výsledek nástroje na straně klienta je přijat
tool:approval { toolCallId, approved } Volání nástroje je schváleno nebo zamítnuto
submission:create { submissionId } Odeslání do Think je přijato
submission:status { submissionId, status } Stav odeslání do Think se změní
submission:error { submissionId, error } Odeslání do Think selže

Události obnovení chatu

Typ Payload Kdy
chat:request:failed { requestId?, stage, messagesPersisted?, error } Chatový požadavek Think selže při parsování, ukládání, spouštění nebo streamování
chat:recovery:detected { incidentId, requestId, attempt, maxAttempts, recoveryKind } Přerušený chatový fiber je poprvé zaznamenán
chat:recovery:attempt { incidentId, requestId, attempt, maxAttempts, recoveryKind } Framework zahajuje pokus o obnovení
chat:recovery:scheduled { incidentId, requestId, attempt, maxAttempts, recoveryKind } Callback pro opakování nebo pokračování je naplánován
chat:recovery:completed { incidentId, requestId, attempt, maxAttempts, recoveryKind } Obnovení bylo úspěšně dokončeno
chat:recovery:skipped { incidentId, requestId, attempt, maxAttempts, recoveryKind, reason? } Obnovení bylo přeskočeno, protože se konverzace změnila nebo již nebyla obnovitelná
chat:recovery:failed { incidentId, requestId, attempt, maxAttempts, recoveryKind, reason? } Obnovení proběhlo, ale selhalo
chat:recovery:exhausted { incidentId, requestId, attempt, maxAttempts, recoveryKind, reason } Obnovení překročilo nakonfigurovaný rozpočet pokusů
chat:stream:stalled { requestId, timeoutMs } Watchdog pro nečinnost zasáhl, protože žádný blok streamu nedorazil do chatStreamStallTimeoutMs. Kolo se přesměruje do trvalého zotavení

recoveryKind je "retry" když obnova přehraje nezodpovězené kolo uživatele a "continue" když navazuje na částečné kolo asistenta.

Události kontextu chatu

Typ Payload Kdy
chat:context:compacted { reason, shortened, requestId?, attempt? } Think komprimuje relaci, aby zvládl přetečení kontextového okna. reason je "proactive" ( contextOverflow.proactive guard spuštěný před krokem) nebo "reactive" (contextOverflow.reactive spuštěna po přetečení). shortened je, zda komprimace skutečně zmenšila historii: false znamená, že by opakování znovu přeteklo. Viz Obnovení po přetečení kontextového okna.

Události transkriptu

Typ Payload Kdy
chat:transcript:repaired { requestId?, removedToolCalls, normalizedInputs, toolCallIds? } Think před odesláním poskytovateli opraví uložený přepis konverzace. removedToolCalls počítá opravená osiřelá volání nástrojů; normalizedInputs počítá opravené serializované nebo chybějící vstupy nástrojů

Události fiberu

Typ Payload Kdy
fiber:run:started { fiberId, fiberName, managed? } Trvalý fiber se spustí
fiber:run:completed { fiberId, fiberName, managed?, elapsedMs? } Trvalý fiber se dokončí
fiber:run:failed { fiberId, fiberName, managed?, error, elapsedMs? } Trvalý fiber vyvolá výjimku
fiber:run:interrupted { fiberId, fiberName, managed?, recoveryReason, elapsedMs? } Při startu se najde přerušený fiber
fiber:recovery:detected { fiberId, fiberName, managed?, recoveryReason, elapsedMs? } Obnovení zaznamená přerušený fiber
fiber:recovery:attempt { fiberId, fiberName, managed?, recoveryReason } Hook pro obnovení se spustí
fiber:recovery:handled { fiberId, fiberName, managed?, recoveryReason, status, elapsedMs? } Zpracování obnovení se dokončí
fiber:recovery:skipped { fiberId, fiberName, managed?, reason, elapsedMs? } Sken obnovení přeskočí zbývající práci
fiber:recovery:failed { fiberId, fiberName, managed?, error, reason?, elapsedMs? } Hook pro obnovení selže

Události obnovení nástrojů agenta

Typ Payload Kdy
agent_tool:recovery:begin { runCount, totalTimeoutMs? } Recovery nadřazeného agenta zahájí skenování zastaralých běhů agent-tool
agent_tool:recovery:row { runId, agentType, status, reason?, elapsedMs? } Jeden zastaralý běh je sladěn
agent_tool:recovery:deadline { runId, agentType, elapsedMs? } Celková lhůta pro obnovení vyprší dříve, než dojde ke kontrole řádku
agent_tool:recovery:complete { runCount, elapsedMs? } Recovery nadřazeného agenta dokončí skenování řádků
agent_tool:recovery:failed { error } Recovery nadřazeného agenta neočekávaně selže

Plánování a řazení událostí do fronty

Typ Payload Kdy
schedule:create { callback, id } Plán je vytvořen
schedule:execute { callback, id } Naplánovaný callback se spustí
schedule:cancel { callback, id } Plán je zrušen
schedule:retry { callback, id, attempt, maxAttempts } Naplánovaný callback se opakuje
schedule:error { callback, id, error, attempts } Naplánovaný callback selže po vyčerpání všech opakování
schedule:duplicate_warning { callback } Neidempotentní plán může práci zdvojit
queue:create { callback, id } Úkol je zařazen do fronty
queue:retry { callback, id, attempt, maxAttempts } Zařazený callback se opakuje
queue:error { callback, id, error, attempts } Zařazený callback selže po vyčerpání všech opakování

Události životního cyklu

Typ Payload Kdy
connect { connectionId } Připojení WebSocket se naváže
disconnect { connectionId, code, reason } Připojení WebSocket se uzavře
destroy {} Agent je zničen

Události workflow

Typ Payload Kdy
workflow:start { workflowId, workflowName? } Spustí se instance workflow
workflow:event { workflowId, eventType? } Událost je odeslána do workflow
workflow:approved { workflowId, reason? } Workflow je schválen
workflow:rejected { workflowId, reason? } Workflow je zamítnut
workflow:terminated { workflowId, workflowName? } Workflow je ukončen
workflow:paused { workflowId, workflowName? } Workflow je pozastaven
workflow:resumed { workflowId, workflowName? } Workflow je obnoven
workflow:restarted { workflowId, workflowName? } Workflow je restartován

Události MCP

Typ Payload Kdy
mcp:client:preconnect { serverId } Před připojením k serveru MCP
mcp:client:connect { url, transport, state, error? } Pokus o připojení MCP se dokončí nebo selže
mcp:client:authorize { serverId, authUrl, clientId? } Zahájí se MCP OAuth flow
mcp:client:discover { url?, state?, error?, capability? } Úspěch nebo selhání zjišťování možností MCP

E-mailové události

Typ Payload Kdy
email:receive { from, to, subject? } E-mail je přijat
email:reply { from, to, subject? } Odpověď e-mailem je odeslána
email:send { from, to, subject? } E-mail je odeslán

Další kroky

Trasování

Trasujte volání modelu, běhy nástrojů a schválení pomocí trasování Workers.

Konfigurace

Nastavení a nasazení wrangler.jsonc.

Tail Workers

Přeposílejte události diagnostického kanálu do Tail Workeru pro monitorování provozu.