接收来自外部服务的 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 投递事件、退信处理、互动分析 |
| 用例 | 描述 |
|---|---|
| 设备遥测 | 每个设备一个 Agent,处理传感器数据流 |
| 告警聚合 | 从 PagerDuty、Datadog 或自定义监控收集告警 |
| 家庭自动化 | 响应 IFTTT/Zapier 触发器并持久化 state |
| 用例 | 描述 |
|---|---|
| CRM 同步 | Salesforce/HubSpot 联系人与交易更新 |
| 日历 Agent | Google Calendar 事件通知与日程安排 |
| 表单提交 | Typeform、Tally 或自定义表单 webhook 及后续操作 |
关键模式是先验证原始请求再解析,然后仅从已认证的 payload 数据推导 Agent 身份。正文签名不能认证无关的 URL 片段或任意 header。
大多数 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>;提供商的正文签名不能认证 webhook URL。若 URL 包含实体 ID,请将其与已验证提供商 payload 中的对应身份比较,并在调用 getAgentByName() 之前拒绝不匹配。
Slack 不会发送经过认证的 X-Slack-Team-Id 路由 header。请根据原始正文验证 Slack 带时间戳的签名与重放窗口,然后从已验证的 event 或 form body 读取 team_id。
始终在信任或处理 payload 之前验证 webhook 签名。
快速入门中的 verifyGitHubWebhook() 辅助函数使用 crypto.subtle.verify() 验证 GitHub 覆盖原始正文的 sha256=<hex> 签名。此格式为 GitHub 特有。其他提供商使用不同的签名编码、签名输入、时间戳检查与重放保护。请遵循 常见 webhook 提供商 下链接的提供商文档。
| 提供商 | 签名 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) |
在 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()})`;
}
}使用 SQLite 持久化 webhook 事件以供历史与 replay。
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>;
}使用 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 结果。
为提供商特定的验证与解析使用一个带类型的辅助函数。它必须按 常见 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>;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,
});
}
}- 始终验证签名 — 切勿信任未验证的 webhook。
- 使用环境 secrets — 使用
wrangler secret put存储密钥,不要写在代码中。 - 快速响应 — 数秒内返回 200/202 以避免重试。
- 校验载荷 — 处理前检查必填字段。
- 记录拒绝 — 跟踪无效签名以进行安全监控。
- 使用 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;| 提供商 | 文档 |
|---|---|
| GitHub | Webhook 事件与载荷 ↗ |
| Stripe | Webhook 签名 ↗ |
| Twilio | 验证 webhook 请求 ↗ |
| Slack | 验证请求 ↗ |
| Shopify | Webhook 验证 ↗ |
| SendGrid | 事件 webhook ↗ |
| Linear | Webhook ↗ |