使用 webhooks

NOTE

本页介绍 LangGraph API webhooks，一种在特定运行完成时接收 POST 回调的方式。如果你在寻找 LangSmith 自动化 webhooks（在自动化规则匹配运行或线程时触发），请参阅为规则配置 webhook 通知。

Webhooks 实现了从你的 LangSmith 应用到外部服务的事件驱动通信。例如，你可能想在某个对 LangSmith 的 API 调用完成后，向一个单独的服务发出更新。

许多 LangSmith 端点都接受一个 webhook 参数。如果该参数由一个能够接受 POST 请求的端点指定，LangSmith 会在一次运行完成时发送一个请求。

在使用 LangSmith 时，你可能想在某个 API 调用完成后使用 webhooks 来接收更新。Webhooks 对于在一次运行处理完毕后触发你服务中的动作很有用。要实现这一点，你需要暴露一个能够接受 POST 请求的端点，并在你的 API 请求中将该端点作为一个 webhook 参数传入。

目前，SDK 不提供用于定义 webhook 端点的内置支持，但你可以通过 API 请求手动指定它们。

支持的端点

以下 API 端点接受一个 webhook 参数：

操作	HTTP 方法	端点
Create Run	POST	/thread/{thread_id}/runs
Create Thread Cron	POST	/thread/{thread_id}/runs/crons
Stream Run	POST	/thread/{thread_id}/runs/stream
Wait Run	POST	/thread/{thread_id}/runs/wait
Create Cron	POST	/runs/crons
Stream Run Stateless	POST	/runs/stream
Wait Run Stateless	POST	/runs/wait

在本指南中，我们将展示如何在流式传输一次运行后触发一个 webhook。

设置你的助手和线程

在进行 API 调用之前，设置你的助手和线程。

from langgraph_sdk import get_client

client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
print(thread)

示例响应：

{
    "thread_id": "9dde5490-2b67-47c8-aa14-4bfec88af217",
    "created_at": "2024-08-30T23:07:38.242730+00:00",
    "updated_at": "2024-08-30T23:07:38.242730+00:00",
    "metadata": {},
    "status": "idle",
    "config": {},
    "values": null
}

在图运行中使用 webhook

要使用一个 webhook，请在你的 API 请求中指定 webhook 参数。当运行完成时，LangSmith 会向指定的 webhook URL 发送一个 POST 请求。

例如，如果你的服务器在 https://my-server.app/my-webhook-endpoint 监听 webhook 事件，请在你的请求中包含以下内容：

input = { "messages": [{ "role": "user", "content": "Hello!" }] }

async for chunk in client.runs.stream(
    thread_id=thread["thread_id"],
    assistant_id=assistant_id,
    input=input,
    stream_mode="events",
    webhook="https://my-server.app/my-webhook-endpoint"
):
    pass

Webhook payload

LangSmith 以一个运行的格式发送 webhook 通知。请求负载在 kwargs 字段中包含运行输入、配置和其他元数据。除了标准的运行字段外，webhook 负载还包含 values、webhook_sent_at 和 error 字段。

完整的 webhook 负载包含以下字段：

字段	类型	描述
run_id	string（UUID）	运行的唯一标识符。
thread_id	string（UUID）	该运行所属线程的标识符。
assistant_id	string	执行该运行的助手的标识符。
status	string	运行的最终状态（例如 "success"、"error"）。
created_at	string（datetime）	运行被创建时的时间戳。
updated_at	string（datetime）	运行最近被更新时的时间戳。
run_started_at	string（datetime）	运行开始执行时的时间戳。
run_ended_at	string（datetime）	运行完成时的时间戳。如果运行尚未结束则省略。
webhook_sent_at	string（datetime）	webhook 请求被发送时的时间戳。
metadata	JSON object	与该运行关联的自定义元数据。
kwargs	JSON object	运行输入、配置和其他调用参数。
values	JSON object	来自线程最新检查点的状态值。仅对有状态运行存在。
multitask_strategy	string	该运行所使用的多任务策略。
error	JSON object | null	仅当运行失败时存在。包含 error（错误类型）和 message（详情）字段。

示例负载：

{
  "run_id": "1ef6a5b8-4457-6db0-8b15-cffd3797fa04",
  "thread_id": "9dde5490-2b67-47c8-aa14-4bfec88af217",
  "assistant_id": "agent",
  "status": "success",
  "created_at": "2024-08-30T23:07:38.242730+00:00",
  "updated_at": "2024-08-30T23:07:40.120000+00:00",
  "run_started_at": "2024-08-30T23:07:38.300000+00:00",
  "run_ended_at": "2024-08-30T23:07:40.100000+00:00",
  "webhook_sent_at": "2024-08-30T23:07:40.150000+00:00",
  "metadata": {},
  "kwargs": {
    "input": {
      "messages": [{"role": "user", "content": "Hello!"}]
    }
  },
  "values": {
    "messages": [
      {"role": "user", "content": "Hello!"},
      {"role": "assistant", "content": "Hi there! How can I help you today?"}
    ]
  },
  "multitask_strategy": "reject",
  "error": null
}

当运行失败时，error 字段包含关于该失败的详情：

{
  "error": {
    "error": "TimeoutError",
    "message": "Run exceeded maximum execution time"
  }
}

保护 webhooks

为确保只有经过授权的请求到达你的 webhook 端点，可以考虑将一个安全令牌作为查询参数添加：

https://my-server.app/my-webhook-endpoint?token=YOUR_SECRET_TOKEN

你的服务器应在处理请求之前提取并验证该令牌。

向 webhook 请求添加标头

NOTE

在 langgraph-api>=0.5.36 中可用。

你可以配置静态标头，以包含在所有出站 webhook 请求中。这对于向你的 webhook 端点进行身份验证、路由或传递元数据很有用。

向你的 langgraph.json 文件添加一个 webhooks.headers 配置：

{
  "webhooks": {
    "headers": {
      "X-Custom-Header": "my-value",
      "X-Environment": "production"
    }
  }
}

在标头中使用环境变量

要在不将机密或环境特定值签入你的配置文件的情况下包含它们，请使用 ${ env.VAR } 模板语法：

{
  "webhooks": {
    "headers": {
      "Authorization": "Bearer ${{ env.LG_WEBHOOK_TOKEN }}"
    }
  }
}

出于安全考虑，默认只有以 LG_WEBHOOK_ 开头的环境变量才能被引用。这可以防止意外泄露无关的环境变量。你可以使用 env_prefix 来自定义该前缀：

{
  "webhooks": {
    "env_prefix": "MY_APP_",
    "headers": {
      "Authorization": "Bearer ${{ env.MY_APP_SECRET }}"
    }
  }
}

NOTE

缺失的必需环境变量会阻止服务器启动，从而确保你不会以不完整的配置进行部署。

限制 webhook 目的地

NOTE

在 langgraph-api>=0.5.36 中可用。

出于安全或合规目的，你可以使用 webhooks.url 配置来限制哪些 URL 是有效的 webhook 目的地：

{
  "webhooks": {
    "url": {
      "allowed_domains": ["*.mycompany.com", "api.trusted-service.com"],
      "require_https": true
    }
  }
}

可用选项：

选项	描述
allowed_domains	主机名允许列表。支持用于子域名的通配符（例如 *.mycompany.com）。
require_https	当为 true 时拒绝 http:// URL。
allowed_ports	显式端口允许列表。默认为 443（https）和 80（http）。
disable_loopback	当为 true 时禁止相对 URL（内部回环调用）。
max_url_length	允许的最大 URL 长度（以字符计）。

禁用 webhooks

从 langgraph-api>=0.2.78 起，开发者可以在 langgraph.json 文件中禁用 webhooks：

{
  "http": {
    "disable_webhooks": true
  }
}

此功能主要用于自托管部署，在那里平台管理员或开发者可能更倾向于禁用 webhooks 以简化其安全态势——尤其是当他们没有配置防火墙规则或其他网络控制时。禁用 webhooks 有助于防止不受信任的负载被发送到内部端点。

有关完整的配置详情，参阅配置文件参考。

测试 webhooks

你可以使用如下在线服务来测试你的 webhook：

• Beeceptor – 快速创建一个测试端点并检查传入的 webhook 负载。
• Webhook.site – 实时查看、调试和记录传入的 webhook 请求。

这些工具帮助你核实 LangSmith 是否正确触发并向你的服务发送 webhooks。

---