构建功能模块

构建功能模块，以复用参与 Agent 启动、请求处理或后台工作的功能。Lifecycle 会启动该功能模块，并调用它实现的钩子。

本指南将构建一个 UptimeMonitor 功能模块。它会定期检查 URL、保存每次检查结果，并通过 HTTP 提供历史记录。检查失败时，它会通过回调通知 Agent。

监控器使用 Lifecycle 的三项功能：

表将检查结果存储在对象的 SQLite 数据库中。
作业记录下一次检查的运行时间。Lifecycle 会管理 alarm，并在到达检查时间时调用监控器的 onJob() 钩子。
请求处理会在 /uptime 提供检查历史记录。其他请求会继续传递给下一个功能模块或 Agent 的 onRequest()。

功能模块只需实现所需的钩子。有关功能模块如何共享启动过程、存储和 alarm，请参阅为什么要使用功能模块。

选择基类

编写 capability 有两种方式：

扩展 LifecycleCapability。 capability 会获得稳定 ID 和标准 Lifecycle 服务：存储、就绪状态、自有任务、事件，以及在主机上下文中调用用户代码的方式。几乎所有 capability 都适用此方式。
实现 DurableObjectCapability 接口。 capability 是一个仅包含所需钩子的普通对象。它不会获得服务或 ID，因此无法推送任务或发出事件。需要的内容请通过构造函数传入。

对于简单的中间件，基于该接口构建的 capability 就足够了：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";

class HealthCheck {
	storage;

	constructor(storage) {
		this.storage = storage;
	}

	onStart() {
		this.storage.sql.exec(
			"CREATE TABLE IF NOT EXISTS health_checks (checked_at INTEGER NOT NULL)",
		);
	}

	onRequest({ request }) {
		if (!new URL(request.url).pathname.endsWith("/health")) return;
		this.storage.sql.exec(
			"INSERT INTO health_checks (checked_at) VALUES (?)",
			Date.now(),
		);
		return new Response("ok");
	}
}

export class MyAgent extends DurableObject {
	health = new HealthCheck(this.ctx.storage);
	lifecycle = Lifecycle.install(this).use(this.health);

	onRequest() {
		return new Response("application response");
	}
}

import { DurableObject } from "cloudflare:workers";
import {
	Lifecycle,
	type CapabilityRequestContext,
	type DurableObjectCapability,
} from "agents/lifecycle";

class HealthCheck implements DurableObjectCapability {
	readonly storage: DurableObjectStorage;

	constructor(storage: DurableObjectStorage) {
		this.storage = storage;
	}

	onStart(): void {
		this.storage.sql.exec(
			"CREATE TABLE IF NOT EXISTS health_checks (checked_at INTEGER NOT NULL)",
		);
	}

	onRequest({ request }: CapabilityRequestContext): Response | undefined {
		if (!new URL(request.url).pathname.endsWith("/health")) return;
		this.storage.sql.exec(
			"INSERT INTO health_checks (checked_at) VALUES (?)",
			Date.now(),
		);
		return new Response("ok");
	}
}

export class MyAgent extends DurableObject<Env> {
	readonly health = new HealthCheck(this.ctx.storage);
	readonly lifecycle = Lifecycle.install(this).use(this.health);

	onRequest(): Response {
		return new Response("application response");
	}
}

本指南其余部分将扩展 LifecycleCapability。

扩展 `LifecycleCapability`

将稳定且非空的 ID 传给 super()。Lifecycle 会使用该 ID 将任务和事件路由回此 capability。部署后不要更改它，否则待处理任务会失去所属方：

import { LifecycleCapability } from "agents/lifecycle";

export class UptimeMonitor extends LifecycleCapability {
	options;

	constructor(options) {
		super("uptime-monitor");
		this.options = options;
	}
}

import { LifecycleCapability } from "agents/lifecycle";

type UptimeMonitorOptions = {
	url: string;
	intervalMs: number;
	onDown?: (status: number) => void | Promise<void>;
};

export class UptimeMonitor extends LifecycleCapability {
	readonly options: UptimeMonitorOptions;

	constructor(options: UptimeMonitorOptions) {
		super("uptime-monitor");
		this.options = options;
	}
}

安装 capability 后，this.lifecycle 会提供 Lifecycle 服务。capability 永远不会直接收到主机对象。请通过构造函数传入绑定和回调。

如果尚未通过 use() 安装 capability，访问 this.lifecycle 会抛出异常。

在启动时创建表

Lifecycle 会先对每个 capability 调用 onStart()，然后调用主机的 onStart()。可在此创建或迁移 capability 的表。请为表名添加前缀，避免与主机或其他 capability 的表冲突：

class UptimeMonitor extends LifecycleCapability {
	// ...

	onStart() {
		this.lifecycle.storage.sql.exec(`
			CREATE TABLE IF NOT EXISTS uptime_checks (
				checked_at INTEGER NOT NULL,
				status INTEGER NOT NULL
			)
		`);
	}
}

class UptimeMonitor extends LifecycleCapability {
	// ...

	override onStart(): void {
		this.lifecycle.storage.sql.exec(`
			CREATE TABLE IF NOT EXISTS uptime_checks (
				checked_at INTEGER NOT NULL,
				status INTEGER NOT NULL
			)
		`);
	}
}

按计划运行任务

推送任务，以便稍后唤醒 capability，并在 onJob() 中处理该任务。返回 { rescheduleAt } 可安排再次运行：

import { LifecycleCapability } from "agents/lifecycle";

class UptimeMonitor extends LifecycleCapability {
	// ...

	async startMonitoring() {
		await this.lifecycle.ready();
		await this.lifecycle.jobs.push({
			id: "check",
			fn: "check",
			time: Date.now(),
		});
	}

	async onJob({ job }) {
		if (job.fn !== "check") return;
		const response = await fetch(this.options.url, { method: "HEAD" });
		this.lifecycle.storage.sql.exec(
			"INSERT INTO uptime_checks (checked_at, status) VALUES (?, ?)",
			Date.now(),
			response.status,
		);
		this.lifecycle.events.emit("uptime:check", { status: response.status });

		const { onDown } = this.options;
		if (!response.ok && onDown) {
			await this.lifecycle.runInHostContext(() => onDown(response.status));
		}
		return { rescheduleAt: Date.now() + this.options.intervalMs };
	}
}

import {
	LifecycleCapability,
	type LifecycleJobContext,
	type LifecycleJobOutcome,
} from "agents/lifecycle";

class UptimeMonitor extends LifecycleCapability {
	// ...

	async startMonitoring(): Promise<void> {
		await this.lifecycle.ready();
		await this.lifecycle.jobs.push({
			id: "check",
			fn: "check",
			time: Date.now(),
		});
	}

	async onJob({ job }: LifecycleJobContext): Promise<LifecycleJobOutcome> {
		if (job.fn !== "check") return;
		const response = await fetch(this.options.url, { method: "HEAD" });
		this.lifecycle.storage.sql.exec(
			"INSERT INTO uptime_checks (checked_at, status) VALUES (?, ?)",
			Date.now(),
			response.status,
		);
		this.lifecycle.events.emit("uptime:check", { status: response.status });

		const { onDown } = this.options;
		if (!response.ok && onDown) {
			await this.lifecycle.runInHostContext(() => onDown(response.status));
		}
		return { rescheduleAt: Date.now() + this.options.intervalMs };
	}
}

任务使用固定 ID，因此再次调用 startMonitoring() 会替换待处理的检查，而不是新增一项检查。请让 onJob() 尽可能简短。对于耗时较长的任务，请参阅运行长时间任务。

在公共方法中等待启动完成

startMonitoring() 会先等待 this.lifecycle.ready()。capability 的公共方法可能在启动前运行，因为 RPC 调用不会启动对象。启动期间，对象的输入门会保持关闭，直到 onStart() 运行完毕。若启动流程之外开始的任务又被 onStart() 等待，该任务就无法完成，直到启动在 30 秒后超时并重置对象。是否出现此问题取决于 RPC 调用是否先到达，因此表现为间歇性超时。

在任何使用存储或任务、或者启动 onStart() 同样会启动的工作的方法开头等待 ready()。在启动期间，ready() 会立即解析，因此 onStart() 可以调用同一方法。

在主机上下文中调用用户代码

capability 钩子在主机的调用上下文之外运行，因此在其中调用 getCurrentAgent() 会得到 undefined 值。请通过 runInHostContext() 调用用户代码，这样其中的 getCurrentAgent() 就会返回主机对象。Scheduler 也会以相同方式调用回调。

拥有连接或请求的 capability 可以将它们作为第二个参数传入：runInHostContext(fn, { connection, request })。这样用户代码就能通过 getCurrentAgent() 获取这些对象。

发出事件

this.lifecycle.events.emit(type, payload) 会使用 capability 的 ID 发布尽力而为的事件。在普通 Durable Object 上，事件会发送到 agents:* 诊断通道。具体通道由事件类型的前缀决定。例如，schedule: 事件会发送到 agents:schedule。若类型的前缀不在 SDK 识别范围内（例如 uptime:check），事件则会发送到 agents:lifecycle。

启动期间发出的事件会暂存到启动完成后再发送。它们属于遥测数据，而非持久化发件箱。如果业务逻辑依赖事件送达，请在 capability 自己的表中存储发件箱。

处理请求

实现 onRequest()，让 capability 在主机前充当中间件。返回 Response 可处理请求；返回 undefined 则将请求继续传递：

import { LifecycleCapability } from "agents/lifecycle";

class UptimeMonitor extends LifecycleCapability {
	// ...

	onRequest({ request }) {
		if (!new URL(request.url).pathname.endsWith("/uptime")) return;
		const checks = this.lifecycle.storage.sql
			.exec(
				"SELECT checked_at, status FROM uptime_checks ORDER BY checked_at DESC LIMIT 20",
			)
			.toArray();
		return Response.json(checks);
	}
}

import {
	LifecycleCapability,
	type CapabilityRequestContext,
} from "agents/lifecycle";

class UptimeMonitor extends LifecycleCapability {
	// ...

	onRequest({ request }: CapabilityRequestContext): Response | undefined {
		if (!new URL(request.url).pathname.endsWith("/uptime")) return;
		const checks = this.lifecycle.storage.sql
			.exec(
				"SELECT checked_at, status FROM uptime_checks ORDER BY checked_at DESC LIMIT 20",
			)
			.toArray();
		return Response.json(checks);
	}
}

只匹配 capability 所负责的路由。若 capability 要处理每个请求，请将 claims 设为 "catch-all"，这样 Lifecycle 会在所有其他 capability 运行后再调用它。更多信息请参阅Capability 顺序。

安装 capability

像安装其他 capability 一样安装此 capability：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";

export class SiteAgent extends DurableObject {
	uptime = new UptimeMonitor({
		url: "https://example.com/",
		intervalMs: 5 * 60_000,
		onDown: (status) => console.warn(`example.com returned ${status}`),
	});
	lifecycle = Lifecycle.install(this).use(this.uptime);

	async onRequest() {
		await this.uptime.startMonitoring();
		return new Response("Monitoring started", { status: 202 });
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";

export class SiteAgent extends DurableObject<Env> {
	readonly uptime = new UptimeMonitor({
		url: "https://example.com/",
		intervalMs: 5 * 60_000,
		onDown: (status) => console.warn(`example.com returned ${status}`),
	});
	readonly lifecycle = Lifecycle.install(this).use(this.uptime);

	async onRequest(): Promise<Response> {
		await this.uptime.startMonitoring();
		return new Response("Monitoring started", { status: 202 });
	}
}

对 /agents/site-agent/example/uptime 的请求会返回存储的检查结果。其他请求都会到达主机的 onRequest()。

测试 capability

像 SiteAgent 示例那样，将 capability 安装到带有真实 Lifecycle 的小型 Durable Object 上进行测试。不要伪造 Lifecycle 服务。请使用 Workers Vitest 集成运行测试，确保任务、警报和存储的行为与生产环境一致。

其他钩子

capability 还可以实现以下钩子。完整签名请参阅 Lifecycle API：

钩子	用途
`onWebSocketUpgrade(context)`	返回 `Response` 以接管 WebSocket upgrade
`onWebSocketMessage`, `onWebSocketClose`, `onWebSocketError`	处理 capability 所拥有套接字的事件，并返回 `true`
`onJobError(context, error)`	决定任务最后一次重试失败后如何处理
`onMemoryLimit(context)`	在警报因内存限制而重置后应用策略
`onRoute(context)`	处理路由到此 capability ID 的消息
`dispose()`	在显式拆除期间释放实时资源

在 `Agent` 类上使用

同一个 capability 也适用于 Agent 类，它会安装相同的 Lifecycle。将 capability 实例化为字段，并在 Agent 的构造函数中调用 super() 后，通过 this.lifecycle.use(this.uptime) 安装。