← Создание агентов на Cloudflare / agents / runtime / lifecycle

Внутреннее устройство класса Agent

Ядро agents библиотека представляет собой Agent класс. Вы расширяете его, переопределяете несколько методов и получаете управление состоянием, WebSockets, планирование, RPC и многое другое бесплатно. На этой странице объясняется, как Agent строится слой за слоем, чтобы вы понимали, что происходит под капотом.

Приведённые здесь сниппеты носят иллюстративный характер и не обязательно отражают лучшие практики. Полное описание API см. в Справочник по API и исходный код ↗.

Agent класс является расширением DurableObject : агенты являются Durable Objects. Если вы не знакомы с Durable Objects, ознакомьтесь с Что такое Durable Objects сначала. По своей сути Durable Objects представляют собой глобально адресуемые (у каждого экземпляра есть уникальный ID) однопоточные вычислительные инстансы с долговременным хранилищем (по типу ключ-значение и SQLite).

Agent не расширяет DurableObject напрямую. Он расширяет Server из partyserver ↗ пакет, который расширяет DurableObject. Представьте это как слои: DurableObject > Сервер > Agent.

Кратко рассмотрим, какие примитивы предоставляют Durable Objects, чтобы понять, как их используют внешние уровни. Класс Durable Object включает:

constructor(ctx: DurableObjectState, env: Env) {}

Рантайм Workers всегда вызывает конструктор для внутренней обработки. Это означает две вещи:

1. Конструктор вызывается при каждой инициализации Durable Object, но его сигнатура фиксирована: разработчики не могут добавлять или изменять параметры конструктора.
2. Вместо ручного создания экземпляра класса разработчики должны использовать binding API и делать это через DurableObjectNamespace.

Написав класс Durable Object, наследующий встроенный тип DurableObject, публичные методы становятся доступны как RPC-методы, которые разработчики могут вызывать с помощью DurableObjectStub из Worker.

// This instance could've been active, hibernated,
// not initialized or maybe had never even been created!
const stub = env.MY_DO.getByName("foo");

// We can call any public method on the class. The runtime
// ensures the constructor is called if the instance was not active.
await stub.bar();

Пробуждение Durable Objects может занять Request из Worker и отправить Response обратно. Это можно сделать только через fetch метод (который должен реализовать разработчик).

Durable Objects включают полноценную поддержку WebSockets. Durable Object может принять WebSocket, полученный от Request в fetch и забыть о нём. Базовый класс предоставляет методы, которые разработчики могут реализовать и которые вызываются как колбэки. По сути, они заменяют необходимость в слушателях событий.

Базовый класс предоставляет webSocketMessage(ws, message), webSocketClose(ws, code, reason, wasClean) и webSocketError(ws , error) (API).

export class MyDurableObject extends DurableObject {
	async fetch(request) {
		// Creates two ends of a WebSocket connection.
		const webSocketPair = new WebSocketPair();
		const [client, server] = Object.values(webSocketPair);

		// Calling `acceptWebSocket()` connects the WebSocket to the Durable Object, allowing the WebSocket to send and receive messages.
		this.ctx.acceptWebSocket(server);

		return new Response(null, {
			status: 101,
			webSocket: client,
		});
	}

	async webSocketMessage(ws, message) {
		ws.send(message);
	}
}

HTTP-запросы и RPC-запросы не единственные точки входа для Durable Object. Будильники позволяют разработчикам запланировать срабатывание события в более позднее время. Как только наступит время следующего будильника, рантайм вызовет alarm() метод, реализация которого остаётся на усмотрение разработчика.

Чтобы запланировать alarm, можно использовать this.ctx.storage.setAlarm() метод. Дополнительную информацию см. в Будильники.

Базовый DurableObject класс задаёт DurableObjectState в this.ctx. Здесь много интересных методов и свойств, но мы сосредоточимся на this.ctx.storage.

DurableObjectStorage это основной интерфейс к механизмам сохранения состояния Durable Object, которые включают как KV, так и SQLITE синхронный API.

const sql = this.ctx.storage.sql;

// Synchronous SQL query
const rows = sql.exec("SELECT * FROM contacts WHERE country = ?", "US");

// Key-value storage
const token = this.ctx.storage.get("someToken");

Наконец, стоит отметить, что у Durable Object также есть Worker Env в this.env. Подробнее см. в Bindings.

Теперь, когда вы увидели, что Durable Objects предоставляют «из коробки», Server класс из partyserver ↗ будет иметь больше смысла. Это выполнено в духе четкой философии DurableObject обертка, которая заменяет низкоуровневые примитивы удобными для разработчика колбэками.

Server не добавляет собственных операций хранения: он лишь оборачивает жизненный цикл Durable Object.

partyserver предоставляет вспомогательные функции для обращения к Durable Objects по имени вместо ручного использования привязок. Сюда входит и схема маршрутизации URL (<your-worker>/servers/:durableClass/:durableName) на которых строится слой Agent.

// Note the await here!
const stub = await getServerByName(env.MY_DO, "foo");

// We can still call RPC methods.
await stub.bar();

Схема URL также позволяет использовать маршрутизатор запросов. На уровне Agent он повторно экспортируется как routeAgentRequest:

  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const res = await routeAgentRequest(request, env);

    if (res) return res;

    return new Response("Not found", { status: 404 });
  }

Уровень адресации позволяет Server чтобы предоставить onStart колбэк, который выполняется каждый раз при запуске Durable Object (после вытеснения, гибернации или первого создания) и перед любым fetch или вызов RPC.

class MyServer extends Server {
	onStart() {
		// Some initialization logic that you wish
		// to run every time the DO is started up.
		const sql = this.ctx.storage.sql;
		sql.exec(`...`);
	}
}

Server уже реализует fetch для базового Durable Object и предоставляет два разных колбэка, которые могут использовать разработчики, onRequest и onConnect для HTTP-запросов и входящих WS-соединений соответственно (WebSocket-соединения принимаются по умолчанию).

class MyServer extends Server {
	async onRequest(request: Request) {
		const url = new URL(request.url);

		return new Response(`Hello from ${url.origin}!`);
	}

	async onConnect(conn, ctx) {
		const { request } = ctx;
		const url = new URL(request.url);

		// Connections are a WebSocket wrapper
		conn.send(`Hello from ${url.origin}!`);
	}
}

Так же, как и onConnect это колбэк для каждого нового подключения, Server также предоставляет обёртки поверх колбэков по умолчанию из DurableObject класс: onMessage, onClose и onError.

Также есть this.broadcast который отправляет WS-сообщение всем подключённым клиентам (никакой магии, просто цикл по this.getConnections()!).

Сложно получить у Durable Object name изнутри него. partyserver пытается сделать это доступным в this.name но это не идеальное решение. Подробнее об этом см. в этот GitHub issue ↗.

И наконец, Agent класс. Agent расширяет Server и предоставляет продуманные примитивы для агентов с сохранением состояния, планированием и наблюдаемостью, которые могут взаимодействовать через RPC, WebSockets и (даже!) электронную почту.

Одна из ключевых возможностей Agent это автоматическое сохранение состояния. Разработчики определяют форму своего состояния через generic-параметр и initialState (используется только если в хранилище отсутствует состояние), а Agent берёт на себя загрузку, сохранение и рассылку изменений состояния (см. Server: this.broadcast() выше).

this.state это геттер, который лениво загружает состояние из хранилища (SQL). Состояние сохраняется при вытеснении Durable Object, если оно обновляется с помощью this.setState(), которая автоматически сериализует состояние и записывает его обратно в хранилище.

Также есть this.onStateChanged который можно переопределить, чтобы реагировать на изменения состояния.

class MyAgent extends Agent<Env, { count: number }> {
	initialState = { count: 0 };

	increment() {
		this.setState({ count: this.state.count + 1 });
	}

	onStateChanged(state, source) {
		console.log("State updated:", state);
	}
}

Состояние хранится в cf_agents_state SQL-таблицу. Сообщения состояния отправляются с type: "cf_agent_state" (как со стороны клиента, так и со стороны сервера). Поскольку agents предоставляет Клиенты на JS и React, обновления состояния в реальном времени доступны из коробки.

Agent предоставляет удобный sql template-тег для выполнения запросов к SQL-хранилищу Durable Object. Он формирует параметризованные запросы и выполняет их. Здесь используется синхронный SQL API из this.ctx.storage.sql.

class MyAgent extends Agent {
	onStart() {
		this.sql`
      CREATE TABLE IF NOT EXISTS users (
        id TEXT PRIMARY KEY,
        name TEXT
      )
    `;

		const userId = "1";
		const userName = "Alice";
		this.sql`INSERT INTO users (id, name) VALUES (${userId}, ${userName})`;

		const users = this.sql<{ id: string; name: string }>`
      SELECT * FROM users WHERE id = ${userId}
    `;
		console.log(users); // [{ id: "1", name: "Alice" }]
	}
}

agents делает ещё один шаг вперёд по сравнению с RPC Durable Objects, реализуя RPC через WebSocket, поэтому клиенты могут вызывать методы Agent напрямую. Чтобы сделать метод доступным через WebSocket, используйте @callable() декоратор. Методы могут возвращать сериализуемое значение или поток (при использовании @callable({ streaming: true })).

class MyAgent extends Agent {
	@callable({ description: "Add two numbers" })
	async add(a: number, b: number) {
		return a + b;
	}
}

Клиенты могут вызывать этот метод, отправив сообщение WebSocket:

{
	"type": "rpc",
	"id": "unique-request-id",
	"method": "add",
	"args": [2, 3]
}

Например, с предоставленным React клиента это делается так же просто:

const { stub } = useAgent({ name: "my-agent" });
const result = await stub.add(2, 3);
console.log(result); // 5

Agents включают встроенную очередь задач для отложенного выполнения. Это удобно для переноса работы на потом или повторных попыток операций. Доступны следующие методы: this.queue, this.dequeue, this.dequeueAll, this.dequeueAllByCallback, this.getQueue, а также this.getQueues.

class MyAgent extends Agent {
	async onConnect() {
		// Queue a task to be executed later
		await this.queue("processTask", { userId: "123" });
	}

	async processTask(payload: { userId: string }, queueItem: QueueItem) {
		console.log("Processing task for user:", payload.userId);
	}
}

Задачи хранятся в cf_agents_queues SQL-таблицу и автоматически сбрасываются по очереди. Если задача выполняется успешно, она автоматически удаляется из очереди.

Agents поддерживают запланированное выполнение методов, оборачивая Durable Object alarm(). Доступные методы: this.schedule, this.getSchedule, this.getSchedules, this.cancelSchedule. Расписания могут быть разовыми, отложенными или повторяющимися (с использованием cron-выражений).

Поскольку Durable Objects допускают не более одного alarm одновременно, Agent класс решает эту задачу тем, что управляет несколькими расписаниями в SQL и использует единственный будильник.

class MyAgent extends Agent {
	async foo() {
		// Schedule at a specific time
		await this.schedule(new Date("2025-12-25T00:00:00Z"), "sendGreeting", {
			message: "Merry Christmas!",
		});

		// Schedule with a delay (in seconds)
		await this.schedule(60, "checkStatus", { check: "health" });

		// Schedule with a cron expression
		await this.schedule("0 0 * * *", "dailyTask", { type: "cleanup" });
	}

	async sendGreeting(payload: { message: string }) {
		console.log(payload.message);
	}

	async checkStatus(payload: { check: string }) {
		console.log("Running check:", payload.check);
	}

	async dailyTask(payload: { type: string }) {
		console.log("Daily task:", payload.type);
	}
}

Расписания хранятся в cf_agents_schedules SQL-таблицу. Задачи по расписанию cron автоматически планируются заново после выполнения, а одноразовые задачи удаляются.

Agent включает MCP-клиент для нескольких серверов. Это позволяет вашему агенту взаимодействовать с внешними сервисами, которые предоставляют интерфейсы MCP. MCP-клиент подробно описан в API MCP-клиента.

class MyAgent extends Agent {
	async onStart() {
		// Add an HTTP MCP server (callbackHost only needed for OAuth servers)
		await this.addMcpServer("GitHub", "https://mcp.github.com/mcp", {
			callbackHost: "https://my-worker.example.workers.dev",
		});

		// Add an MCP server via RPC (Durable Object binding, no HTTP overhead)
		await this.addMcpServer("internal-tools", this.env.MyMCP);
	}
}

Agents могут получать электронные письма и отвечать на них с помощью Email Routing.

class MyAgent extends Agent {
	async onEmail(email: AgentEmail) {
		console.log("Received email from:", email.from);
		console.log("Subject:", email.headers.get("subject"));

		const raw = await email.getRaw();
		console.log("Raw email size:", raw.length);

		// Reply to the email
		await this.replyToEmail(email, {
			fromName: "My Agent",
			subject: "Re: " + email.headers.get("subject"),
			body: "Thanks for your email!",
			contentType: "text/plain",
		});
	}
}

Чтобы направлять письма вашему Agent, используйте routeAgentEmail в обработчике электронной почты вашего Worker:

export default {
	async email(message, env, ctx) {
		await routeAgentEmail(message, env, {
			resolver: createAddressBasedEmailResolver("my-agent"),
		});
	},
} satisfies ExportedHandler<Env>;

agents оборачивает все ваши методы с помощью AsyncLocalStorage чтобы сохранять контекст на протяжении всего жизненного цикла запроса. Это позволяет получать доступ к текущему агенту, подключению, запросу или письму (в зависимости от обрабатываемого события) из любого места вашего кода:

import { getCurrentAgent } from "agents";

function someUtilityFunction() {
	const { agent, connection, request, email } = getCurrentAgent();

	if (agent) {
		console.log("Current agent:", agent.name);
	}

	if (connection) {
		console.log("WebSocket connection ID:", connection.id);
	}
}

Agent расширяет Server: onError чтобы его можно было использовать для обработки ошибок, которые не обязательно являются ошибками WebSocket. Он вызывается с Connection или unknown ошибка.

class MyAgent extends Agent {
	onError(connectionOrError: Connection | unknown, error?: unknown) {
		if (error) {
			// WebSocket connection error
			console.error("Connection error:", error);
		} else {
			// Server error
			console.error("Server error:", connectionOrError);
		}

		// Optionally throw to propagate the error
		throw connectionOrError;
	}
}

this.destroy() удаляет все таблицы, удаляет alarm, очищает хранилище и прерывает контекст. Чтобы гарантировать полное вытеснение Durable Object, this.ctx.abort() вызывается асинхронно с помощью setTimeout() чтобы дать любым выполняющимся в данный момент обработчикам (например, запланированным задачам) завершить операции очистки до того, как контекст будет прерван.

Это означает this.ctx.abort() выбрасывает неперехватываемое исключение, которое появится в ваших логах, но делает это уже после передачи управления циклу событий (подробнее об этом в abort()).

destroy() метод можно безопасно вызывать внутри запланированных задач. При вызове изнутри колбэка расписания Agent устанавливает внутренний флаг, чтобы пропустить оставшиеся обновления базы данных, и возвращает ctx.abort() в цикл событий, чтобы обработчик alarm корректно завершился до вытеснения Agent.

class MyAgent extends Agent {
	async onStart() {
		console.log("Agent is starting up...");
		// Initialize your agent
	}

	async cleanup() {
		// This wipes everything!
		await this.destroy();
	}

	async selfDestruct() {
		// Safe to call from within a scheduled task
		await this.schedule(60, "destroyAfterDelay", {});
	}

	async destroyAfterDelay() {
		// This will safely destroy the Agent even when
		// called from within the alarm handler
		await this.destroy();
	}
}

Настройте поведение агента, переопределив static options в вашем классе. Все поля необязательны: значения по умолчанию применяются во время выполнения.

export class MyAgent extends Agent {
	static options = {
		hibernate: true,
		sendIdentityOnConnect: false,
		retry: { maxAttempts: 5, baseDelayMs: 200, maxDelayMs: 5000 },
	};
}

Durable Objects вытесняются из памяти после периода бездействия (обычно от 70 до 140 секунд без входящих запросов, сообщений WebSocket или alarm). Во время долгих операций, таких как потоковая передача ответов LLM, ожидание внешних API или выполнение многошаговых вычислений, агент может быть вытеснен прямо в процессе работы.

keepAlive() создаёт alarm heartbeat, который предотвращает вытеснение. keepAliveWhile() оборачивает асинхронную функцию и гарантирует очистку.

class MyAgent extends Agent {
	async handleLongTask() {
		// Option 1: manual dispose
		const dispose = await this.keepAlive();
		try {
			await longRunningComputation();
		} finally {
			dispose();
		}

		// Option 2: automatic cleanup (recommended)
		const result = await this.keepAliveWhile(async () => {
			return await longRunningComputation();
		});
	}
}

AIChatAgent использует keepAliveWhile внутренне, чтобы поддерживать агент активным во время потоковой передачи ответов LLM. Подробнее см. Планирование задач: поддержание агента активным.

Agent класс повторно экспортирует вспомогательные функции для адресации как getAgentByName и routeAgentRequest.

const stub = await getAgentByName(env.MY_DO, "foo");
await stub.someMethod();

const res = await routeAgentRequest(request, env);
if (res) return res;

return new Response("Not found", { status: 404 });

AIChatAgent класс из @cloudflare/ai-chat расширяет Agent со слоем, построенным на четкой философии, для чата с ИИ. Он добавляет автоматическое сохранение сообщений в SQLite, возобновляемую потоковую передачу, поддержку инструментов (на стороне сервера, на стороне клиента и с участием человека) и React hook (useAgentChat) для создания пользовательских интерфейсов чата.

Полная иерархия: DurableObject > Сервер > Agent > AIChatAgent.

Если вы создаёте чат-агента, начните с AIChatAgent. Если вам нужен контроль более низкого уровня или вы не создаёте чат-интерфейс, используйте Agent напрямую.