使用 webhooks¶

在使用 LangGraph Platform 时,你可能希望使用 webhooks 在 API 调用完成后接收更新。Webhooks 对于在运行完成处理后触发服务中的操作很有用。要实现这一点,你需要公开一个可以接受 POST 请求的端点,并在 API 请求中将此端点作为 webhook 参数传递。

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

支持的端点¶

以下 API 端点接受 webhook 参数:

操作	HTTP 方法	端点
创建运行	`POST`	`/thread/{thread_id}/runs`
创建线程定时任务	`POST`	`/thread/{thread_id}/runs/crons`
流式运行	`POST`	`/thread/{thread_id}/runs/stream`
等待运行	`POST`	`/thread/{thread_id}/runs/wait`
创建定时任务	`POST`	`/runs/crons`
无状态流式运行	`POST`	`/runs/stream`
无状态等待运行	`POST`	`/runs/wait`

在本指南中,我们将展示如何在流式运行后触发 webhook。

设置你的助手和线程¶

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

PythonJavaScriptCURL

from langgraph_sdk import get_client

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

import { Client } from "@langchain/langgraph-sdk";

const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantID = "agent";
const thread = await client.threads.create();
console.log(thread);

curl --request POST \
    --url <DEPLOYMENT_URL>/assistants/search \
    --header 'Content-Type: application/json' \
    --data '{ "limit": 10, "offset": 0 }' | jq -c 'map(select(.config == null or .config == {})) | .[0]' && \
curl --request POST \
    --url <DEPLOYMENT_URL>/threads \
    --header 'Content-Type: application/json' \
    --data '{}'

示例响应:

{
    "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 参数。当运行完成时,LangGraph Platform 会向指定的 webhook URL 发送 POST 请求。

例如,如果你的服务器在 https://my-server.app/my-webhook-endpoint 监听 webhook 事件,请在请求中包含此信息:

PythonJavaScriptCURL

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

const input = { messages: [{ role: "human", content: "Hello!" }] };

const streamResponse = client.runs.stream(
  thread["thread_id"],
  assistantID,
  {
    input: input,
    webhook: "https://my-server.app/my-webhook-endpoint"
  }
);

for await (const chunk of streamResponse) {
  // Handle stream output
}

curl --request POST \
    --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
    --header 'Content-Type: application/json' \
    --data '{
        "assistant_id": <ASSISTANT_ID>,
        "input": {"messages": [{"role": "user", "content": "Hello!"}]},
        "webhook": "https://my-server.app/my-webhook-endpoint"
    }'

Webhook 负载¶

LangGraph Platform 以运行的格式发送 webhook 通知。有关详细信息,请参阅 API 参考。请求负载在 kwargs 字段中包含运行输入、配置和其他元数据。

安全的 webhooks¶

为确保只有授权请求到达你的 webhook 端点,请考虑将安全令牌添加为查询参数:

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

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

禁用 webhooks¶

从 langgraph-api>=0.2.78 开始,开发人员可以在 langgraph.json 文件中禁用 webhooks:

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

此功能主要用于自托管部署,平台管理员或开发人员可能更喜欢禁用 webhooks 以简化其安全态势——特别是如果他们不配置防火墙规则或其他网络控制。禁用 webhooks 有助于防止不受信任的有效负载发送到内部端点。

有关完整的配置详细信息,请参阅配置文件参考。

测试 webhooks¶

你可以使用以下在线服务测试你的 webhook:

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

这些工具可帮助你验证 LangGraph Platform 是否正确触发并向你的服务发送 webhooks。