From cdabbea262bae2319eff41ec4ad8069932fc54bf Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 15 Apr 2025 17:07:53 +0200 Subject: [PATCH] Match tests --- libs/langgraph-api/src/queue.mts | 28 +++++- libs/langgraph-api/src/schemas.mts | 4 +- libs/langgraph-api/src/server.mts | 4 + libs/langgraph-api/src/webhook.mts | 47 ++++++++++ libs/langgraph-api/tests/api.test.mts | 87 +++++++++++++++++++ .../tests/graphs/agent_simple.mts | 9 +- libs/langgraph-api/tests/graphs/http.mts | 65 +++++++++++++- .../tests/graphs/langgraph.http.json | 16 ---- .../langgraph-api/tests/graphs/langgraph.json | 6 +- 9 files changed, 240 insertions(+), 26 deletions(-) create mode 100644 libs/langgraph-api/src/webhook.mts delete mode 100644 libs/langgraph-api/tests/graphs/langgraph.http.json diff --git a/libs/langgraph-api/src/queue.mts b/libs/langgraph-api/src/queue.mts index 2ce821d..2fe9712 100644 --- a/libs/langgraph-api/src/queue.mts +++ b/libs/langgraph-api/src/queue.mts @@ -1,4 +1,4 @@ -import { type Run, Runs, Threads } from "./storage/ops.mjs"; +import { type Run, type RunStatus, Runs, Threads } from "./storage/ops.mjs"; import { type StreamCheckpoint, type StreamTaskResult, @@ -6,6 +6,7 @@ import { } from "./stream.mjs"; import { logError, logger } from "./logging.mjs"; import { serializeError } from "./utils/serde.mjs"; +import { callWebhook } from "./webhook.mjs"; const MAX_RETRY_ATTEMPTS = 3; @@ -23,10 +24,13 @@ export const queue = async () => { const worker = async (run: Run, attempt: number, abortSignal: AbortSignal) => { const startedAt = new Date(); + let endedAt: Date | undefined = undefined; let checkpoint: StreamCheckpoint | undefined = undefined; let exception: Error | undefined = undefined; + let status: RunStatus | undefined = undefined; const temporary = run.kwargs.temporary; + const webhook = run.kwargs.webhook as string | undefined; logger.info("Starting background run", { run_id: run.run_id, @@ -68,7 +72,7 @@ const worker = async (run: Run, attempt: number, abortSignal: AbortSignal) => { throw error; } - const endedAt = new Date(); + endedAt = new Date(); logger.info("Background run succeeded", { run_id: run.run_id, run_attempt: attempt, @@ -77,9 +81,11 @@ const worker = async (run: Run, attempt: number, abortSignal: AbortSignal) => { run_ended_at: endedAt, run_exec_ms: endedAt.valueOf() - startedAt.valueOf(), }); - await Runs.setStatus(run.run_id, "success"); + + status = "success"; + await Runs.setStatus(run.run_id, status); } catch (error) { - const endedAt = new Date(); + endedAt = new Date(); if (error instanceof Error) exception = error; logError(error, { @@ -93,6 +99,8 @@ const worker = async (run: Run, attempt: number, abortSignal: AbortSignal) => { run_exec_ms: endedAt.valueOf() - startedAt.valueOf(), }, }); + + status = "error"; await Runs.setStatus(run.run_id, "error"); } finally { if (temporary) { @@ -100,5 +108,17 @@ const worker = async (run: Run, attempt: number, abortSignal: AbortSignal) => { } else { await Threads.setStatus(run.thread_id, { checkpoint, exception }); } + + if (webhook) { + await callWebhook({ + checkpoint, + status, + exception, + run, + webhook, + run_started_at: startedAt, + run_ended_at: endedAt, + }); + } } }; diff --git a/libs/langgraph-api/src/schemas.mts b/libs/langgraph-api/src/schemas.mts index 6db8402..629d09e 100644 --- a/libs/langgraph-api/src/schemas.mts +++ b/libs/langgraph-api/src/schemas.mts @@ -98,7 +98,7 @@ export const CronCreate = z .describe("Metadata for the run.") .optional(), config: AssistantConfig.optional(), - webhook: z.string().url().optional(), + webhook: z.string().optional(), interrupt_before: z.union([z.enum(["*"]), z.array(z.string())]).optional(), interrupt_after: z.union([z.enum(["*"]), z.array(z.string())]).optional(), multitask_strategy: z @@ -188,7 +188,7 @@ export const RunCreate = z .describe("Metadata for the run.") .optional(), config: AssistantConfig.optional(), - webhook: z.string().url().optional(), + webhook: z.string().optional(), interrupt_before: z.union([z.enum(["*"]), z.array(z.string())]).optional(), interrupt_after: z.union([z.enum(["*"]), z.array(z.string())]).optional(), on_disconnect: z diff --git a/libs/langgraph-api/src/server.mts b/libs/langgraph-api/src/server.mts index 687070d..c5b0731 100644 --- a/libs/langgraph-api/src/server.mts +++ b/libs/langgraph-api/src/server.mts @@ -20,6 +20,7 @@ import { auth } from "./auth/custom.mjs"; import { registerAuth } from "./auth/index.mjs"; import { registerHttp } from "./http/custom.mjs"; import { cors, ensureContentType } from "./http/middleware.mjs"; +import { bindLoopbackFetch } from "./webhook.mjs"; export const StartServerSchema = z.object({ port: z.number(), @@ -129,6 +130,9 @@ export async function startServer(options: z.infer) { app.route("/", api); } + // Loopback fetch used by webhooks + bindLoopbackFetch(app); + logger.info(`Starting ${options.nWorkers} workers`); for (let i = 0; i < options.nWorkers; i++) queue(); diff --git a/libs/langgraph-api/src/webhook.mts b/libs/langgraph-api/src/webhook.mts new file mode 100644 index 0000000..b621de8 --- /dev/null +++ b/libs/langgraph-api/src/webhook.mts @@ -0,0 +1,47 @@ +import { Hono } from "hono"; +import type { Run } from "./storage/ops.mjs"; +import type { StreamCheckpoint } from "./stream.mjs"; +import { serializeError } from "./utils/serde.mjs"; + +let LOOPBACK_FETCH: + | ((url: string, init?: RequestInit) => Promise | undefined) + | undefined; + +export const bindLoopbackFetch = (app: Hono) => { + LOOPBACK_FETCH = async (url: string, init?: RequestInit) => + app.request(url, init); +}; + +export async function callWebhook(result: { + checkpoint: StreamCheckpoint | undefined; + status: string | undefined; + exception: Error | undefined; + run: Run; + webhook: string; + run_started_at: Date; + run_ended_at: Date | undefined; +}) { + const payload = { + ...result.run, + status: result.status, + run_started_at: result.run_started_at.toISOString(), + run_ended_at: result.run_ended_at?.toISOString(), + webhook_sent_at: new Date().toISOString(), + values: result.checkpoint?.values, + ...(result.exception + ? { error: serializeError(result.exception).message } + : undefined), + }; + + if (result.webhook.startsWith("/")) { + await LOOPBACK_FETCH?.(result.webhook, { + method: "POST", + body: JSON.stringify(payload), + }); + } else { + await fetch(result.webhook, { + method: "POST", + body: JSON.stringify(payload), + }); + } +} diff --git a/libs/langgraph-api/tests/api.test.mts b/libs/langgraph-api/tests/api.test.mts index a55528e..d68895d 100644 --- a/libs/langgraph-api/tests/api.test.mts +++ b/libs/langgraph-api/tests/api.test.mts @@ -2418,3 +2418,90 @@ it("generative ui", async () => { client["~ui"].getComponent("non-existent", "none"), ).rejects.toThrow(); }); + +it("custom routes", async () => { + const fetcher = async (...args: Parameters) => { + const res = await fetch(...args); + if (!res.ok) throw new Error(`${res.status} ${res.statusText}`); + return { json: await res.json(), headers: res.headers }; + }; + + let res = await fetcher(new URL("/custom/my-route?aCoolParam=13", API_URL), { + headers: { "x-custom-input": "hey" }, + }); + expect(res.json).toEqual({ foo: "bar" }); + expect(res.headers.get("x-custom-output")).toEqual("hey"); + expect(res.headers.get("x-js-middleware")).toEqual("true"); + + res = await fetcher(new URL("/runs/afakeroute", API_URL)); + expect(res.json).toEqual({ foo: "afakeroute" }); + + await expect(() => + fetcher(new URL("/does/not/exist", API_URL)), + ).rejects.toThrow("404"); + + await expect(() => + fetcher(new URL("/custom/error", API_URL)), + ).rejects.toThrow("400"); + + if (!IS_MEMORY) { + await expect(() => + fetcher(new URL("/__langgraph_check", API_URL), { method: "OPTIONS" }), + ).rejects.toThrow("404"); + } + + const stream = await fetch(new URL("/custom/streaming", API_URL)); + const reader = stream.body?.getReader(); + if (!reader) throw new Error("No reader"); + + const chunks: string[] = []; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(new TextDecoder().decode(value)); + } + + expect(chunks.length).toBeGreaterThanOrEqual(4); // Must actually stream + expect(chunks.join("")).toEqual("Count: 0\nCount: 1\nCount: 2\nCount: 3\n"); + + const thread = await client.threads.create(); + await client.runs.wait(thread.thread_id, "agent_simple", { + input: { messages: [{ role: "human", content: "foo" }] }, + webhook: "/custom/webhook", + }); + + await expect + .poll(() => fetcher(new URL("/custom/webhook-payload", API_URL)), { + interval: 500, + timeout: 3000, + }) + .toMatchObject({ json: { status: "success" } }); + + // check if custom middleware is applied even for python routes + res = await fetcher(new URL("/info", API_URL)); + expect(res.headers.get("x-js-middleware")).toEqual("true"); + + // ... and if we can intercept a request targeted for Python API + res = await fetcher(new URL("/info?interrupt", API_URL)); + expect(res.json).toEqual({ status: "interrupted" }); +}); + +it("custom routes - mutate request body", async () => { + const client = new Client({ + apiUrl: API_URL, + defaultHeaders: { + "x-configurable-header": "extra-client", + }, + }); + + const thread = await client.threads.create(); + const res = await client.runs.wait(thread.thread_id, "agent_simple", { + input: { messages: [{ role: "human", content: "input" }] }, + }); + + expect(res).toEqual({ + messages: expect.arrayContaining([ + expect.objectContaining({ content: "end: extra-client" }), + ]), + }); +}); diff --git a/libs/langgraph-api/tests/graphs/agent_simple.mts b/libs/langgraph-api/tests/graphs/agent_simple.mts index e8b2002..a0af19c 100644 --- a/libs/langgraph-api/tests/graphs/agent_simple.mts +++ b/libs/langgraph-api/tests/graphs/agent_simple.mts @@ -41,6 +41,12 @@ async function callModel( userId = user?.identity; } + if (config.configurable?.["x-configurable-header"] != null) { + return { + messages: [`end: ${config.configurable?.["x-configurable-header"]}`], + }; + } + const model = getStableModel(config.configurable?.thread_id ?? "$"); const existing = await config.store?.get([userId ?? "ALL"], "key_one"); if (!existing) { @@ -49,8 +55,7 @@ async function callModel( } const response = await model.invoke(state.messages); - const result: typeof AgentState.Update = { messages: [response] }; - return result; + return { messages: [response] }; } async function callTool( diff --git a/libs/langgraph-api/tests/graphs/http.mts b/libs/langgraph-api/tests/graphs/http.mts index 1e4b745..a79d62d 100644 --- a/libs/langgraph-api/tests/graphs/http.mts +++ b/libs/langgraph-api/tests/graphs/http.mts @@ -1,3 +1,66 @@ import { Hono } from "hono"; +import { HTTPException } from "hono/http-exception"; +import { streamText } from "hono/streaming"; -export const app = new Hono().get("/info", (c) => c.json({ random: true })); +let WEBHOOK_PAYLOAD: Record; + +export const app = new Hono<{ + Variables: { body: string | ArrayBuffer | ReadableStream | null }; +}>() + .use(async (c, next) => { + if (c.req.query("interrupt") != null) { + return c.json({ status: "interrupted" }); + } + + await next(); + c.header("x-js-middleware", "true"); + }) + .use(async (c, next) => { + const runsQuery = new RegExp( + "^(/runs(/stream|/wait)?$|/runs/batch$|/threads/[^/]+/runs(/stream|/wait)?)$", + ); + + if (c.req.method === "POST" && c.req.path.match(runsQuery)) { + const value = c.req.header("x-configurable-header"); + + if (value != null) { + const body = await c.req.json(); + + body["config"] ??= {}; + body["config"]["configurable"] ??= {}; + body["config"]["configurable"]["x-configurable-header"] ??= value; + } + } + + await next(); + }) + .get("/custom/my-route", (c) => + c.json( + { foo: "bar" }, + { + headers: { + "x-custom-output": c.req.header("x-custom-input") as string, + }, + }, + ), + ) + .get("/runs/afakeroute", (c) => c.json({ foo: "afakeroute" })) + .get("/custom/error", () => { + throw new HTTPException(400, { message: "Bad request" }); + }) + .get("/custom/streaming", (c) => + streamText(c, async (stream) => { + for (let i = 0; i < 4; i++) { + await stream.writeln(`Count: ${i}`); + await new Promise((resolve) => setTimeout(resolve, 10)); + } + + await stream.close(); + }), + ) + .post("/custom/webhook", async (c) => { + WEBHOOK_PAYLOAD = await c.req.json(); + return c.json({ status: "success" }); + }) + .get("/custom/webhook-payload", (c) => c.json(WEBHOOK_PAYLOAD)) + .notFound((c) => c.json({ status: "not-found" })); diff --git a/libs/langgraph-api/tests/graphs/langgraph.http.json b/libs/langgraph-api/tests/graphs/langgraph.http.json deleted file mode 100644 index aafd032..0000000 --- a/libs/langgraph-api/tests/graphs/langgraph.http.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "node_version": "20", - "graphs": { - "agent": "./agent.mts:graph", - "agent_simple": "./agent_simple.mts:graph" - }, - "ui": { - "agent": "./agent/ui.tsx" - }, - "http": { - "app": "./http.mts:app" - }, - "env": { - "LANGGRAPH_CONFIG": "{\"agent\": {\"configurable\": {\"model_name\": \"openai\"}}}" - } -} diff --git a/libs/langgraph-api/tests/graphs/langgraph.json b/libs/langgraph-api/tests/graphs/langgraph.json index af498f2..ea8f854 100644 --- a/libs/langgraph-api/tests/graphs/langgraph.json +++ b/libs/langgraph-api/tests/graphs/langgraph.json @@ -6,11 +6,15 @@ "weather": "./weather.mts:graph", "error": "./error.mts:graph", "delay": "./delay.mts:graph", - "dynamic": "./dynamic.mts:graph" + "dynamic": "./dynamic.mts:graph", + "agent_simple": "./agent_simple.mts:graph" }, "ui": { "agent": "./agent/ui.tsx" }, + "http": { + "app": "./http.mts:app" + }, "env": { "LANGGRAPH_CONFIG": "{\"agent\": {\"configurable\": {\"model_name\": \"openai\"}}}" }