跳转到内容
本站为社区维护的非官方中文翻译,与 Cloudflare, Inc. 无关联。本站不提供账户登录,请勿在任何非官方站点输入您的 Cloudflare 账户信息。官方文档:developers.cloudflare.com

Webhook

最后更新 查看 MarkdownAgent 设置

接收来自外部服务的 webhook 事件并将其路由到专用 Agent 实例。 每个 webhook 来源(仓库、客户、设备)可有独立 state、持久化存储与实时客户端连接的 Agent。

快速入门

import { Agent, getAgentByName, routeAgentRequest } from "agents";

export class WebhookAgent extends Agent {
	async onRequest(request) {
		if (request.method !== "POST") {
			return new Response("Method not allowed", { status: 405 });
		}

		const rawBody = await request.text();
		const signature = request.headers.get("X-Hub-Signature-256");
		if (
			!(await verifyGitHubWebhook(rawBody, signature, this.env.WEBHOOK_SECRET))
		) {
			return new Response("Invalid signature", { status: 401 });
		}

		let payload;
		try {
			payload = JSON.parse(rawBody);
		} catch {
			return new Response("Invalid payload", { status: 400 });
		}

		await this.processEvent(payload);
		return new Response("OK");
	}

	async processEvent(payload) {
		// Store event, update state, trigger actions...
	}
}

async function verifyGitHubWebhook(rawBody, signature, secret) {
	if (!signature || !/^sha256=[0-9a-f]{64}$/i.test(signature)) return false;

	const encoder = new TextEncoder();
	const key = await crypto.subtle.importKey(
		"raw",
		encoder.encode(secret),
		{ name: "HMAC", hash: "SHA-256" },
		false,
		["verify"],
	);
	const signatureBytes = Uint8Array.from(
		signature.slice("sha256=".length).match(/.{2}/g) ?? [],
		(byte) => Number.parseInt(byte, 16),
	);

	return crypto.subtle.verify(
		"HMAC",
		key,
		signatureBytes,
		encoder.encode(rawBody),
	);
}

export default {
	async fetch(request, env) {
		const url = new URL(request.url);

		if (url.pathname === "/webhooks/github" && request.method === "POST") {
			const rawBody = await request.clone().text();
			const signature = request.headers.get("X-Hub-Signature-256");
			if (
				!(await verifyGitHubWebhook(rawBody, signature, env.WEBHOOK_SECRET))
			) {
				return new Response("Invalid signature", { status: 401 });
			}

			let payload;
			try {
				payload = JSON.parse(rawBody);
			} catch {
				return new Response("Invalid payload", { status: 400 });
			}

			const repository = payload.repository?.full_name;
			if (!repository) {
				return new Response("Missing repository", { status: 400 });
			}

			const agentName = repository.toLowerCase().replace(/\//g, "-");
			const agent = await getAgentByName(env.WebhookAgent, agentName);
			return agent.fetch(request);
		}

		return (
			(await routeAgentRequest(request, env)) ||
			new Response("Not found", { status: 404 })
		);
	},
};
import { Agent, getAgentByName, routeAgentRequest } from "agents";

export class WebhookAgent extends Agent<Env> {
	async onRequest(request: Request): Promise<Response> {
		if (request.method !== "POST") {
			return new Response("Method not allowed", { status: 405 });
		}

		const rawBody = await request.text();
		const signature = request.headers.get("X-Hub-Signature-256");
		if (
			!(await verifyGitHubWebhook(
				rawBody,
				signature,
				this.env.WEBHOOK_SECRET,
			))
		) {
			return new Response("Invalid signature", { status: 401 });
		}

		let payload: unknown;
		try {
			payload = JSON.parse(rawBody);
		} catch {
			return new Response("Invalid payload", { status: 400 });
		}

		await this.processEvent(payload);
		return new Response("OK");
	}

	private async processEvent(payload: unknown) {
		// Store event, update state, trigger actions...
	}
}

async function verifyGitHubWebhook(
	rawBody: string,
	signature: string | null,
	secret: string,
): Promise<boolean> {
	if (!signature || !/^sha256=[0-9a-f]{64}$/i.test(signature)) return false;

	const encoder = new TextEncoder();
	const key = await crypto.subtle.importKey(
		"raw",
		encoder.encode(secret),
		{ name: "HMAC", hash: "SHA-256" },
		false,
		["verify"],
	);
	const signatureBytes = Uint8Array.from(
		signature.slice("sha256=".length).match(/.{2}/g) ?? [],
		(byte) => Number.parseInt(byte, 16),
	);

	return crypto.subtle.verify(
		"HMAC",
		key,
		signatureBytes,
		encoder.encode(rawBody),
	);
}

export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		const url = new URL(request.url);

		if (url.pathname === "/webhooks/github" && request.method === "POST") {
			const rawBody = await request.clone().text();
			const signature = request.headers.get("X-Hub-Signature-256");
			if (
				!(await verifyGitHubWebhook(
					rawBody,
					signature,
					env.WEBHOOK_SECRET,
				))
			) {
				return new Response("Invalid signature", { status: 401 });
			}

			let payload: { repository?: { full_name?: string } };
			try {
				payload = JSON.parse(rawBody);
			} catch {
				return new Response("Invalid payload", { status: 400 });
			}

			const repository = payload.repository?.full_name;
			if (!repository) {
				return new Response("Missing repository", { status: 400 });
			}

			const agentName = repository.toLowerCase().replace(/\//g, "-");
			const agent = await getAgentByName(env.WebhookAgent, agentName);
			return agent.fetch(request);
		}

		return (
			(await routeAgentRequest(request, env)) ||
			new Response("Not found", { status: 404 })
		);
	},
} satisfies ExportedHandler<Env>;

用例

Webhook 与 Agent 结合,使每个外部实体拥有隔离、有 state 的 Agent 实例。

开发者工具

用例 描述
GitHub 仓库监控 每个仓库一个 Agent,跟踪 commit、PR、issue 与 star
CI/CD 流水线 Agent 响应构建/部署事件,失败时通知,跟踪部署历史
Linear/Jira 跟踪器 自动分类 issue,按内容分配,跟踪解决时间

电商与支付

用例 描述
Stripe 客户 Agent 每个客户一个 Agent,跟踪支付、订阅与争议
Shopify 订单 Agent 从创建到履约的订单生命周期,含库存同步
支付对账 将 webhook 事件与内部记录匹配,标记差异

通信与通知

用例 描述
Twilio 短信/语音 由入站消息或通话触发的对话 Agent
Slack Bot 响应斜杠命令、按钮点击与交互消息
邮件跟踪 SendGrid/Mailgun 投递事件、退信处理、互动分析

IoT 与基础设施

用例 描述
设备遥测 每个设备一个 Agent,处理传感器数据流
告警聚合 从 PagerDuty、Datadog 或自定义监控收集告警
家庭自动化 响应 IFTTT/Zapier 触发器并持久化 state

SaaS 集成

用例 描述
CRM 同步 Salesforce/HubSpot 联系人与交易更新
日历 Agent Google Calendar 事件通知与日程安排
表单提交 Typeform、Tally 或自定义表单 webhook 及后续操作

将 webhook 路由到 Agent

关键模式是先验证原始请求再解析,然后仅从已认证的 payload 数据推导 Agent 身份。正文签名不能认证无关的 URL 片段或任意 header。

从 payload 提取实体

大多数 webhook 在 payload 中包含标识符:

export default {
	async fetch(request, env) {
		const url = new URL(request.url);
		if (request.method === "POST" && url.pathname === "/webhooks/github") {
			const rawBody = await request.clone().text();
			const signature = request.headers.get("X-Hub-Signature-256");
			if (
				!(await verifyGitHubWebhook(rawBody, signature, env.WEBHOOK_SECRET))
			) {
				return new Response("Invalid signature", { status: 401 });
			}

			let payload;
			try {
				payload = JSON.parse(rawBody);
			} catch {
				return new Response("Invalid payload", { status: 400 });
			}

			const repository = payload.repository?.full_name;
			if (!repository) {
				return new Response("Missing repository", { status: 400 });
			}

			const agentName = repository.toLowerCase().replace(/\//g, "-");
			const agent = await getAgentByName(env.RepoAgent, agentName);
			return agent.fetch(request);
		}
	},
};
export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		const url = new URL(request.url);
		if (request.method === "POST" && url.pathname === "/webhooks/github") {
			const rawBody = await request.clone().text();
			const signature = request.headers.get("X-Hub-Signature-256");
			if (
				!(await verifyGitHubWebhook(
					rawBody,
					signature,
					env.WEBHOOK_SECRET,
				))
			) {
				return new Response("Invalid signature", { status: 401 });
			}

			let payload: { repository?: { full_name?: string } };
			try {
				payload = JSON.parse(rawBody);
			} catch {
				return new Response("Invalid payload", { status: 400 });
			}

			const repository = payload.repository?.full_name;
			if (!repository) {
				return new Response("Missing repository", { status: 400 });
			}

			const agentName = repository.toLowerCase().replace(/\//g, "-");
			const agent = await getAgentByName(env.RepoAgent, agentName);
			return agent.fetch(request);
		}
	},
} satisfies ExportedHandler<Env>;

校验 URL 中的实体 ID

提供商的正文签名不能认证 webhook URL。若 URL 包含实体 ID,请将其与已验证提供商 payload 中的对应身份比较,并在调用 getAgentByName() 之前拒绝不匹配。

从已验证正文推导 Slack 身份

Slack 不会发送经过认证的 X-Slack-Team-Id 路由 header。请根据原始正文验证 Slack 带时间戳的签名与重放窗口,然后从已验证的 event 或 form body 读取 team_id。

签名验证

始终在信任或处理 payload 之前验证 webhook 签名。

GitHub HMAC-SHA256 模式

快速入门中的 verifyGitHubWebhook() 辅助函数使用 crypto.subtle.verify() 验证 GitHub 覆盖原始正文的 sha256=<hex> 签名。此格式为 GitHub 特有。其他提供商使用不同的签名编码、签名输入、时间戳检查与重放保护。请遵循 常见 webhook 提供商 下链接的提供商文档。

提供商特定 header

提供商 签名 Header 算法
GitHub X-Hub-Signature-256 HMAC-SHA256
Stripe Stripe-Signature HMAC-SHA256(含时间戳)
Twilio X-Twilio-Signature HMAC-SHA1
Slack X-Slack-Signature HMAC-SHA256(含时间戳)
Shopify X-Shopify-Hmac-Sha256 HMAC-SHA256(base64)

处理 webhook

onRequest 处理程序

在 Agent 中使用 onRequest() 处理入站 webhook。若 Worker 尚未验证请求,请在解析正文之前验证。本示例复用快速入门中的 verifyGitHubWebhook() 辅助函数:

export class WebhookAgent extends Agent {
	async onRequest(request) {
		// 1. Validate method
		if (request.method !== "POST") {
			return new Response("Method not allowed", { status: 405 });
		}

		// 2. Get the GitHub event type
		const eventType = request.headers.get("X-GitHub-Event") ?? "unknown";

		// 3. Verify the GitHub signature
		const signature = request.headers.get("X-Hub-Signature-256");
		const body = await request.text();

		if (
			!(await verifyGitHubWebhook(body, signature, this.env.WEBHOOK_SECRET))
		) {
			return new Response("Invalid signature", { status: 401 });
		}

		// 4. Parse and process
		const payload = JSON.parse(body);
		await this.handleEvent(eventType, payload);

		// 5. Respond quickly
		return new Response("OK", { status: 200 });
	}

	async handleEvent(type, payload) {
		// Update state (broadcasts to connected clients)
		this.setState({
			...this.state,
			lastEventType: type,
			lastEventTime: new Date().toISOString(),
		});

		// Store in SQL for history
		this
			.sql`INSERT INTO events (type, payload, timestamp) VALUES (${type}, ${JSON.stringify(payload)}, ${Date.now()})`;
	}
}
export class WebhookAgent extends Agent {
	async onRequest(request: Request): Promise<Response> {
		// 1. Validate method
		if (request.method !== "POST") {
			return new Response("Method not allowed", { status: 405 });
		}

		// 2. Get the GitHub event type
		const eventType = request.headers.get("X-GitHub-Event") ?? "unknown";

		// 3. Verify the GitHub signature
		const signature = request.headers.get("X-Hub-Signature-256");
		const body = await request.text();

		if (
			!(await verifyGitHubWebhook(body, signature, this.env.WEBHOOK_SECRET))
		) {
			return new Response("Invalid signature", { status: 401 });
		}

		// 4. Parse and process
		const payload = JSON.parse(body);
		await this.handleEvent(eventType, payload);

		// 5. Respond quickly
		return new Response("OK", { status: 200 });
	}

	private async handleEvent(type: string, payload: unknown) {
		// Update state (broadcasts to connected clients)
		this.setState({
			...this.state,
			lastEventType: type,
			lastEventTime: new Date().toISOString(),
		});

		// Store in SQL for history
		this
			.sql`INSERT INTO events (type, payload, timestamp) VALUES (${type}, ${JSON.stringify(payload)}, ${Date.now()})`;
	}
}

存储 webhook 事件

使用 SQLite 持久化 webhook 事件以供历史与 replay。

事件表 schema

class WebhookAgent extends Agent {
	async onStart() {
		this.sql`
      CREATE TABLE IF NOT EXISTS events (
        id TEXT PRIMARY KEY,
        type TEXT NOT NULL,
        action TEXT,
        title TEXT NOT NULL,
        description TEXT,
        url TEXT,
        actor TEXT,
        payload TEXT,
        timestamp TEXT NOT NULL
      )
    `;

		this.sql`
      CREATE INDEX IF NOT EXISTS idx_events_timestamp
      ON events(timestamp DESC)
    `;
	}
}
class WebhookAgent extends Agent {
	async onStart(): Promise<void> {
		this.sql`
      CREATE TABLE IF NOT EXISTS events (
        id TEXT PRIMARY KEY,
        type TEXT NOT NULL,
        action TEXT,
        title TEXT NOT NULL,
        description TEXT,
        url TEXT,
        actor TEXT,
        payload TEXT,
        timestamp TEXT NOT NULL
      )
    `;

		this.sql`
      CREATE INDEX IF NOT EXISTS idx_events_timestamp
      ON events(timestamp DESC)
    `;
	}
}

清理旧事件

仅保留最近事件以防止无限增长:

// Keep last 100 events
this.sql`
  DELETE FROM events WHERE id NOT IN (
    SELECT id FROM events ORDER BY timestamp DESC LIMIT 100
  )
`;

// Or delete events older than 30 days
this.sql`
  DELETE FROM events
  WHERE timestamp < datetime('now', '-30 days')
`;
// Keep last 100 events
this.sql`
  DELETE FROM events WHERE id NOT IN (
    SELECT id FROM events ORDER BY timestamp DESC LIMIT 100
  )
`;

// Or delete events older than 30 days
this.sql`
  DELETE FROM events
  WHERE timestamp < datetime('now', '-30 days')
`;

查询事件

import { Agent, callable } from "agents";

class WebhookAgent extends Agent {
	@callable()
	getEvents(limit = 20) {
		return [
			...this.sql`
      SELECT * FROM events
      ORDER BY timestamp DESC
      LIMIT ${limit}
    `,
		];
	}

	@callable()
	getEventsByType(type, limit = 20) {
		return [
			...this.sql`
      SELECT * FROM events
      WHERE type = ${type}
      ORDER BY timestamp DESC
      LIMIT ${limit}
    `,
		];
	}
}
import { Agent, callable } from "agents";

class WebhookAgent extends Agent {
	@callable()
	getEvents(limit = 20) {
		return [
			...this.sql`
      SELECT * FROM events
      ORDER BY timestamp DESC
      LIMIT ${limit}
    `,
		];
	}

	@callable()
	getEventsByType(type: string, limit = 20) {
		return [
			...this.sql`
      SELECT * FROM events
      WHERE type = ${type}
      ORDER BY timestamp DESC
      LIMIT ${limit}
    `,
		];
	}
}

实时广播

webhook 到达时更新 Agent 状态,自动向已连接 WebSocket 客户端广播。

class WebhookAgent extends Agent {
	async processWebhook(eventType, payload) {
		// Update state - this automatically broadcasts to all connected clients
		this.setState({
			...this.state,
			stats: payload.stats,
			lastEvent: {
				type: eventType,
				timestamp: new Date().toISOString(),
			},
		});
	}
}
class WebhookAgent extends Agent {
	private async processWebhook(eventType: string, payload: WebhookPayload) {
		// Update state - this automatically broadcasts to all connected clients
		this.setState({
			...this.state,
			stats: payload.stats,
			lastEvent: {
				type: eventType,
				timestamp: new Date().toISOString(),
			},
		});
	}
}

客户端:

import { useAgent } from "agents/react";

function Dashboard() {
	const [state, setState] = useState(null);

	const agent = useAgent({
		agent: "webhook-agent",
		name: "my-entity-id",
		onStateUpdate: (newState) => {
			setState(newState); // Automatically updates when webhooks arrive
		},
	});

	return <div>Last event: {state?.lastEvent?.type}</div>;
}

模式

事件 dedupe

使用 event ID 防止处理重复事件:

class WebhookAgent extends Agent {
	async handleEvent(eventId, payload) {
		// Check if already processed
		const existing = [
			...this.sql`
      SELECT id FROM events WHERE id = ${eventId}
    `,
		];

		if (existing.length > 0) {
			console.log(`Event ${eventId} already processed, skipping`);
			return;
		}

		// Process and store
		await this.processPayload(payload);
		this.sql`INSERT INTO events (id, ...) VALUES (${eventId}, ...)`;
	}
}
class WebhookAgent extends Agent {
	async handleEvent(eventId: string, payload: unknown) {
		// Check if already processed
		const existing = [
			...this.sql`
      SELECT id FROM events WHERE id = ${eventId}
    `,
		];

		if (existing.length > 0) {
			console.log(`Event ${eventId} already processed, skipping`);
			return;
		}

		// Process and store
		await this.processPayload(payload);
		this.sql`INSERT INTO events (id, ...) VALUES (${eventId}, ...)`;
	}
}

快速响应,异步处理

Webhook provider 期望快速响应。用 queue 处理繁重工作:

class WebhookAgent extends Agent {
	async onRequest(request) {
		const payload = await request.json();

		// Quick validation
		if (!this.isValid(payload)) {
			return new Response("Invalid", { status: 400 });
		}

		// Queue heavy processing
		await this.queue("processWebhook", payload);

		// Respond immediately
		return new Response("Accepted", { status: 202 });
	}

	async processWebhook(payload) {
		// Heavy processing happens here, after response sent
		await this.enrichData(payload);
		await this.notifyDownstream(payload);
		await this.updateAnalytics(payload);
	}
}
class WebhookAgent extends Agent {
	async onRequest(request: Request): Promise<Response> {
		const payload = await request.json();

		// Quick validation
		if (!this.isValid(payload)) {
			return new Response("Invalid", { status: 400 });
		}

		// Queue heavy processing
		await this.queue("processWebhook", payload);

		// Respond immediately
		return new Response("Accepted", { status: 202 });
	}

	async processWebhook(payload: WebhookPayload) {
		// Heavy processing happens here, after response sent
		await this.enrichData(payload);
		await this.notifyDownstream(payload);
		await this.updateAnalytics(payload);
	}
}

若异步工作是单个 Think 聊天轮次,请改用 submitMessages()。它会立即返回持久提交 ID,并允许重试时使用相同幂等键,避免重复消息轮次:

const submission = await this.submitMessages(messages, {
	idempotencyKey: payload.id,
});

return Response.json(
	{ submissionId: submission.submissionId },
	{ status: 202 },
);
const submission = await this.submitMessages(messages, {
	idempotencyKey: payload.id,
});

return Response.json(
	{ submissionId: submission.submissionId },
	{ status: 202 },
);

如果 webhook 负责 turn 周围的应用程序副作用,例如恢复 provider thread 并发布可见回复,请在该任务外使用 startFiber()。Managed fiber 会保留状态、对 provider retry 进行去重,并允许 onFiberRecovered() 或 resolveFiber() 记录应用程序级 recovery 结果。

多 provider 路由

为提供商特定的验证与解析使用一个带类型的辅助函数。它必须按 常见 webhook 提供商 下链接的提供商文档验证原始请求,然后仅从已验证正文推导 agentName。

export default {
	async fetch(request, env) {
		const url = new URL(request.url);
		if (request.method === "POST" && url.pathname.startsWith("/webhooks/")) {
			const verified = await verifyAndParseWebhook(request.clone(), env);
			if (!verified) {
				return new Response("Invalid signature", { status: 401 });
			}

			switch (verified.provider) {
				case "github":
					return (
						await getAgentByName(env.GitHubAgent, verified.agentName)
					).fetch(request);
				case "stripe":
					return (
						await getAgentByName(env.StripeAgent, verified.agentName)
					).fetch(request);
				case "slack":
					return (
						await getAgentByName(env.SlackAgent, verified.agentName)
					).fetch(request);
			}
		}

		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
};
type VerifiedWebhook =
	| { provider: "github"; agentName: string }
	| { provider: "stripe"; agentName: string }
	| { provider: "slack"; agentName: string };

declare function verifyAndParseWebhook(
	request: Request,
	env: Env,
): Promise<VerifiedWebhook | null>;

export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		const url = new URL(request.url);
		if (request.method === "POST" && url.pathname.startsWith("/webhooks/")) {
			const verified = await verifyAndParseWebhook(request.clone(), env);
			if (!verified) {
				return new Response("Invalid signature", { status: 401 });
			}

			switch (verified.provider) {
				case "github":
					return (
						await getAgentByName(env.GitHubAgent, verified.agentName)
					).fetch(request);
				case "stripe":
					return (
						await getAgentByName(env.StripeAgent, verified.agentName)
					).fetch(request);
				case "slack":
					return (
						await getAgentByName(env.SlackAgent, verified.agentName)
					).fetch(request);
			}
		}

		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
} satisfies ExportedHandler<Env>;

发送出站 webhook

Agent 也可向外部服务发送 webhook:

export class NotificationAgent extends Agent {
	async notifySlack(message) {
		const response = await fetch(this.env.SLACK_WEBHOOK_URL, {
			method: "POST",
			headers: { "Content-Type": "application/json" },
			body: JSON.stringify({ text: message }),
		});

		if (!response.ok) {
			throw new Error(`Slack notification failed: ${response.status}`);
		}
	}

	async sendSignedWebhook(url, payload) {
		const body = JSON.stringify(payload);
		const signature = await this.sign(body, this.env.WEBHOOK_SECRET);

		await fetch(url, {
			method: "POST",
			headers: {
				"Content-Type": "application/json",
				"X-Signature": signature,
			},
			body,
		});
	}
}
export class NotificationAgent extends Agent {
	async notifySlack(message: string) {
		const response = await fetch(this.env.SLACK_WEBHOOK_URL, {
			method: "POST",
			headers: { "Content-Type": "application/json" },
			body: JSON.stringify({ text: message }),
		});

		if (!response.ok) {
			throw new Error(`Slack notification failed: ${response.status}`);
		}
	}

	async sendSignedWebhook(url: string, payload: unknown) {
		const body = JSON.stringify(payload);
		const signature = await this.sign(body, this.env.WEBHOOK_SECRET);

		await fetch(url, {
			method: "POST",
			headers: {
				"Content-Type": "application/json",
				"X-Signature": signature,
			},
			body,
		});
	}
}

安全最佳实践

  1. 始终验证签名 — 切勿信任未验证的 webhook。
  2. 使用环境 secrets — 使用 wrangler secret put 存储密钥,不要写在代码中。
  3. 快速响应 — 数秒内返回 200/202 以避免重试。
  4. 校验载荷 — 处理前检查必填字段。
  5. 记录拒绝 — 跟踪无效签名以进行安全监控。
  6. 使用 HTTPS — Webhook URL 应始终使用 TLS。
// Store secrets securely
// wrangler secret put GITHUB_WEBHOOK_SECRET

// Access in agent
const secret = this.env.GITHUB_WEBHOOK_SECRET;
// Store secrets securely
// wrangler secret put GITHUB_WEBHOOK_SECRET

// Access in agent
const secret = this.env.GITHUB_WEBHOOK_SECRET;

常见 webhook provider

后续步骤

这篇文档对您有帮助吗?