Skip to content

使用Webhooks

在使用LangGraph Cloud时，您可能希望使用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之前，请设置您的助手和线程。

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) {
  // 处理流输出
}

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