Skip to content

使用 Webhook

在使用 LangGraph Cloud 时，你可能希望在 API 调用完成后使用 Webhook 来接收更新。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 调用之前，先设置好你的助手和线程。

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

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

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 Cloud 以 运行 的格式发送 Webhook 通知。有关详细信息，请参阅 API 参考。请求有效负载在 kwargs 字段中包含运行输入、配置和其他元数据。

保护 Webhook

为确保只有经过授权的请求才能访问你的 Webhook 端点，可考虑添加一个安全令牌作为查询参数：

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

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

测试 Webhook

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

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

这些工具可帮助你验证 LangGraph Cloud 是否正确触发 Webhook 并将其发送到你的服务。

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

Comments