Match tests

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,
+      });
+    }
   }
 };

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

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<typeof StartServerSchema>) {
     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();
 

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<Response> | 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),
+    });
+  }
+}

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<typeof fetch>) => {
+    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<any>({
+    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" }),
+    ]),
+  });
+});

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(

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<string, unknown>;
+
+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" }));

libs/langgraph-api/tests/graphs/langgraph.http.json

@@ -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\"}}}"
-  }
-}

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\"}}}"
   }