使用 Webhook¶
在使用 LangGraph Cloud 时,你可能希望使用 webhook 在 API 调用完成后接收更新。一旦运行处理完成,webhook 对于触发你的服务中的操作非常有用。要实现这一点,你需要公开一个可以接受 POST
请求的端点,并在你的 API 请求中将其作为 webhook
参数传递。
当前,SDK 不提供定义 webhook 端点的内置支持,但你可以通过 API 请求手动指定它们。
支持的端点¶
以下 API 端点接受 webhook
参数
操作 | HTTP 方法 | 端点 |
---|---|---|
创建运行 | POST |
/thread/{thread_id}/runs |
创建线程 Cron 作业 | POST |
/thread/{thread_id}/runs/crons |
流式传输运行 | POST |
/thread/{thread_id}/runs/stream |
等待运行 | POST |
/thread/{thread_id}/runs/wait |
创建 Cron 作业 | 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 以运行 (Run) 的格式发送 webhook 通知。详情请参阅API 参考。请求负载在 kwargs
字段中包含运行输入、配置和其他元数据。
保护 Webhook¶
为了确保只有授权的请求到达你的 webhook 端点,可以考虑添加一个安全令牌作为查询参数
你的服务器应在处理请求之前提取并验证此令牌。
测试 Webhook¶
你可以使用以下在线服务测试你的 webhook,例如
- Beeceptor – 快速创建测试端点并检查传入的 webhook 负载。
- Webhook.site – 实时查看、调试和记录传入的 webhook 请求。
这些工具可以帮助你验证 LangGraph Cloud 是否正确触发并将 webhook 发送到你的服务。
通过遵循这些步骤,你可以将 webhook 集成到你的 LangGraph Cloud 工作流程中,根据已完成的运行实现自动化操作。