跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

Webhook

最后更新 查看 MarkdownAgent 设置

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

快速入门

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

// Agent that handles webhooks for a specific entity
export class WebhookAgent extends Agent {
	async onRequest(request) {
		if (request.method !== "POST") {
			return new Response("Method not allowed", { status: 405 });
		}

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

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

		// Process the webhook payload
		const payload = JSON.parse(body);
		await this.processEvent(payload);

		return new Response("OK", { status: 200 });
	}

	async verifySignature(payload, signature, secret) {
		if (!signature) return false;

		const encoder = new TextEncoder();
		const key = await crypto.subtle.importKey(
			"raw",
			encoder.encode(secret),
			{ name: "HMAC", hash: "SHA-256" },
			false,
			["sign"],
		);

		const signatureBytes = await crypto.subtle.sign(
			"HMAC",
			key,
			encoder.encode(payload),
		);
		const expected = `sha256=${Array.from(new Uint8Array(signatureBytes))
			.map((b) => b.toString(16).padStart(2, "0"))
			.join("")}`;

		return signature === expected;
	}

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

// Route webhooks to the right agent instance
export default {
	async fetch(request, env) {
		const url = new URL(request.url);

		// Webhook endpoint: POST /webhooks/:entityId
		if (url.pathname.startsWith("/webhooks/") && request.method === "POST") {
			const entityId = url.pathname.split("/")[2];
			const agent = await getAgentByName(env.WebhookAgent, entityId);
			return agent.fetch(request);
		}

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

// Agent that handles webhooks for a specific entity
export class WebhookAgent extends Agent {
	async onRequest(request: Request): Promise<Response> {
		if (request.method !== "POST") {
			return new Response("Method not allowed", { status: 405 });
		}

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

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

		// Process the webhook payload
		const payload = JSON.parse(body);
		await this.processEvent(payload);

		return new Response("OK", { status: 200 });
	}

	private async verifySignature(
		payload: string,
		signature: string | null,
		secret: string,
	): Promise<boolean> {
		if (!signature) return false;

		const encoder = new TextEncoder();
		const key = await crypto.subtle.importKey(
			"raw",
			encoder.encode(secret),
			{ name: "HMAC", hash: "SHA-256" },
			false,
			["sign"],
		);

		const signatureBytes = await crypto.subtle.sign(
			"HMAC",
			key,
			encoder.encode(payload),
		);
		const expected = `sha256=${Array.from(new Uint8Array(signatureBytes))
			.map((b) => b.toString(16).padStart(2, "0"))
			.join("")}`;

		return signature === expected;
	}

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

// Route webhooks to the right agent instance
export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		const url = new URL(request.url);

		// Webhook endpoint: POST /webhooks/:entityId
		if (url.pathname.startsWith("/webhooks/") && request.method === "POST") {
			const entityId = url.pathname.split("/")[2];
			const agent = await getAgentByName(env.WebhookAgent, entityId);
			return agent.fetch(request);
		}

		// Default routing for WebSocket connections
		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

关键模式是从 webhook 提取实体标识符，并用 getAgentByName() 路由到专用 Agent 实例。

从 payload 提取实体

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

export default {
	async fetch(request, env) {
		if (request.method === "POST" && url.pathname === "/webhooks/github") {
			const payload = await request.clone().json();

			// Extract entity ID from payload
			const repoFullName = payload.repository?.full_name;
			if (!repoFullName) {
				return new Response("Missing repository", { status: 400 });
			}

			// Sanitize for use as agent name
			const agentName = repoFullName.toLowerCase().replace(/\//g, "-");

			// Route to dedicated agent
			const agent = await getAgentByName(env.RepoAgent, agentName);
			return agent.fetch(request);
		}
	},
};
export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		if (request.method === "POST" && url.pathname === "/webhooks/github") {
			const payload = await request.clone().json();

			// Extract entity ID from payload
			const repoFullName = payload.repository?.full_name;
			if (!repoFullName) {
				return new Response("Missing repository", { status: 400 });
			}

			// Sanitize for use as agent name
			const agentName = repoFullName.toLowerCase().replace(/\//g, "-");

			// Route to dedicated agent
			const agent = await getAgentByName(env.RepoAgent, agentName);
			return agent.fetch(request);
		}
	},
} satisfies ExportedHandler<Env>;

从 URL 提取实体

或者，在 webhook URL 中包含实体 ID：

// Webhook URL: https://your-worker.dev/webhooks/stripe/cus_123456
if (url.pathname.startsWith("/webhooks/stripe/")) {
	const customerId = url.pathname.split("/")[3]; // "cus_123456"
	const agent = await getAgentByName(env.StripeAgent, customerId);
	return agent.fetch(request);
}
// Webhook URL: https://your-worker.dev/webhooks/stripe/cus_123456
if (url.pathname.startsWith("/webhooks/stripe/")) {
	const customerId = url.pathname.split("/")[3]; // "cus_123456"
	const agent = await getAgentByName(env.StripeAgent, customerId);
	return agent.fetch(request);
}

从 header 提取实体

部分服务在 header 中包含标识符：

// Slack sends workspace info in headers
const teamId = request.headers.get("X-Slack-Team-Id");
if (teamId) {
	const agent = await getAgentByName(env.SlackAgent, teamId);
	return agent.fetch(request);
}
// Slack sends workspace info in headers
const teamId = request.headers.get("X-Slack-Team-Id");
if (teamId) {
	const agent = await getAgentByName(env.SlackAgent, teamId);
	return agent.fetch(request);
}

签名验证

始终验证 webhook 签名以确保请求真实。大多数 provider 使用 HMAC-SHA256。

HMAC-SHA256 模式

async function verifySignature(payload, signature, secret) {
	if (!signature) return false;

	const encoder = new TextEncoder();
	const key = await crypto.subtle.importKey(
		"raw",
		encoder.encode(secret),
		{ name: "HMAC", hash: "SHA-256" },
		false,
		["sign"],
	);

	const signatureBytes = await crypto.subtle.sign(
		"HMAC",
		key,
		encoder.encode(payload),
	);

	const expected = `sha256=${Array.from(new Uint8Array(signatureBytes))
		.map((b) => b.toString(16).padStart(2, "0"))
		.join("")}`;

	// Use timing-safe comparison in production
	return signature === expected;
}
async function verifySignature(
	payload: string,
	signature: string | null,
	secret: string,
): Promise<boolean> {
	if (!signature) return false;

	const encoder = new TextEncoder();
	const key = await crypto.subtle.importKey(
		"raw",
		encoder.encode(secret),
		{ name: "HMAC", hash: "SHA-256" },
		false,
		["sign"],
	);

	const signatureBytes = await crypto.subtle.sign(
		"HMAC",
		key,
		encoder.encode(payload),
	);

	const expected = `sha256=${Array.from(new Uint8Array(signatureBytes))
		.map((b) => b.toString(16).padStart(2, "0"))
		.join("")}`;

	// Use timing-safe comparison in production
	return signature === expected;
}

提供商特定 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：

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 event type from headers
		const eventType = request.headers.get("X-Event-Type");

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

		if (!(await this.verifySignature(body, signature))) {
			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 event type from headers
		const eventType = request.headers.get("X-Event-Type");

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

		if (!(await this.verifySignature(body, signature))) {
			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 线程并发布可见回复），请用 startFiber() 包裹该任务。Managed fiber 保留状态、对 provider 重试去重，并让 onFiberRecovered() 或 resolveFiber() 记录应用级恢复结果。

多 provider 路由

在一个 Worker 中处理来自多个服务的 webhook：

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

		if (request.method === "POST") {
			// GitHub webhooks
			if (url.pathname.startsWith("/webhooks/github/")) {
				const payload = await request.clone().json();
				const repoName = payload.repository?.full_name?.replace("/", "-");
				const agent = await getAgentByName(env.GitHubAgent, repoName);
				return agent.fetch(request);
			}

			// Stripe webhooks
			if (url.pathname.startsWith("/webhooks/stripe/")) {
				const payload = await request.clone().json();
				const customerId = payload.data?.object?.customer;
				const agent = await getAgentByName(env.StripeAgent, customerId);
				return agent.fetch(request);
			}

			// Slack webhooks
			if (url.pathname === "/webhooks/slack") {
				const teamId = request.headers.get("X-Slack-Team-Id");
				const agent = await getAgentByName(env.SlackAgent, teamId);
				return agent.fetch(request);
			}
		}

		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
};
export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		const url = new URL(request.url);

		if (request.method === "POST") {
			// GitHub webhooks
			if (url.pathname.startsWith("/webhooks/github/")) {
				const payload = await request.clone().json();
				const repoName = payload.repository?.full_name?.replace("/", "-");
				const agent = await getAgentByName(env.GitHubAgent, repoName);
				return agent.fetch(request);
			}

			// Stripe webhooks
			if (url.pathname.startsWith("/webhooks/stripe/")) {
				const payload = await request.clone().json();
				const customerId = payload.data?.object?.customer;
				const agent = await getAgentByName(env.StripeAgent, customerId);
				return agent.fetch(request);
			}

			// Slack webhooks
			if (url.pathname === "/webhooks/slack") {
				const teamId = request.headers.get("X-Slack-Team-Id");
				const agent = await getAgentByName(env.SlackAgent, teamId);
				return agent.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 提供商

提供商 文档
GitHub Webhook 事件与载荷 ↗
Stripe Webhook 签名 ↗
Twilio 验证 webhook 请求 ↗
Slack 验证请求 ↗
Shopify Webhook 验证 ↗
SendGrid 事件 webhook ↗
Linear Webhooks ↗

后续步骤

这篇文档对您有帮助吗？