使用 Webhooks¶

当使用 LangGraph Cloud 时，您可能希望使用 webhook 以在 API 调用完成后接收更新。Webhook 对于在运行完成后触发服务中的操作非常有用。要实现这一点，您需要公开一个可以接受 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 调用之前，请设置你的助手和线程。

Python

from langgraph_sdk import get_client

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

JavaScript

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

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 Cloud 会向指定的 webhook URL 发送 POST 请求。

例如，如果您的服务器在 https://my-server.app/my-webhook-endpoint 监听 webhook 事件，请将此地址包含在您的请求中

Python

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

JavaScript

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

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 Cloud 以 Run 的格式发送 webhook 通知。有关详细信息，请参阅 API 参考文档。请求载荷包括运行输入、配置以及 kwargs 字段中的其他元数据。

保护 Webhooks 安全¶

为确保只有授权的请求访问您的 webhook 端点，请考虑添加安全令牌作为查询参数

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

您的服务器应提取并验证此令牌，然后再处理请求。

测试 Webhooks¶

您可以使用在线服务（如）测试您的 webhook

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

这些工具可帮助您验证 LangGraph Cloud 是否正确触发并向您的服务发送 webhook。

通过遵循这些步骤，您可以将 webhook 集成到您的 LangGraph Cloud 工作流程中，从而根据已完成的运行自动执行操作。