DX-3022: mcp-toolkit: isError fails the task, keep failure details from the model

.changeset/mcp-toolkit-initial.md

@@ -24,6 +24,10 @@ limit of live subscriptions per subscriber (8 by default), SSRF checks on callba
 secrets encrypted at rest. `QStashDelivery` signs each attempt with
 Standard Webhooks and lets QStash retry failures (`export const POST = events.createDeliveryHandler()`).
 
+A handler that returns a tool error (`isError: true`) fails its task without a retry, and
+`task_status` shows that error the way the synchronous tool would. The model only ever sees a
+failure's `code` and `message`: transport details stay in the store, and dispatch errors are logged.
+
 The package uses WebCrypto only, so it runs on Node and edge runtimes.
 
 `@upstash/mcp-toolkit/upstash` holds the Upstash backends for both: `RedisTaskStore`,

packages/mcp-toolkit/README.md

@@ -12,14 +12,15 @@ Durable building blocks for MCP servers built on the official TypeScript SDK
   `WorkflowDispatcher`, `RedisSubscriptionStore`, `QStashDelivery`.
 
 ```bash
-npm install @upstash/mcp-toolkit @modelcontextprotocol/server zod
+npm install @upstash/mcp-toolkit @modelcontextprotocol/server mcp-handler zod
 ```
 
 Environment variables:
 
 ```bash
 UPSTASH_REDIS_REST_URL=...
 UPSTASH_REDIS_REST_TOKEN=...
+QSTASH_URL=...                   # your QStash region's URL, from the console
 QSTASH_TOKEN=...
 QSTASH_CURRENT_SIGNING_KEY=...   # required: delivery routes refuse to run without them
 QSTASH_NEXT_SIGNING_KEY=...
@@ -30,14 +31,24 @@ APP_URL=https://your-app.com     # used in the snippets below. QStash has to be
 ## Who is calling
 
 Both layers need `principal`: a function that returns the id of the user making the call. Tasks
-belong to the user who started them, and subscriptions to the user who subscribed. Read it from the
-token your MCP route has verified, and throw when there is none:
+belong to the user who started them, and subscriptions to the user who subscribed. Keep it next to
+the function that verifies the token, so the whole path is in one file:
 
 ```ts
 // lib/auth.ts
 import type { AuthInfo } from "@modelcontextprotocol/server";
 
-// `auth` is the AuthInfo your MCP route verified and passed to the SDK (see below).
+// 1. Runs on every MCP request (wired up with withMcpAuth below). Returning undefined answers 401.
+export async function verifyToken(
+  req: Request,
+  bearerToken?: string,
+): Promise<AuthInfo | undefined> {
+  if (!bearerToken) return undefined;
+  const { sub, client_id } = await verifyJwt(bearerToken); // Clerk, WorkOS, Auth0, your own
+  return { token: bearerToken, clientId: client_id, scopes: [], extra: { userId: sub } };
+}
+
+// 2. The toolkit calls this with the AuthInfo above. Return the user id, or throw.
 export function principal({ auth }: { auth?: AuthInfo }): string {
   const userId = auth?.extra?.userId;
   if (typeof userId !== "string") throw new Error("Not authenticated");
@@ -52,8 +63,9 @@ export function principal({ auth }: { auth?: AuthInfo }): string {
   returns anything but a non-empty string, the call is refused as not authenticated.
 - **Use the user id, not `auth.clientId`.** The client id identifies the OAuth app, and every
   ChatGPT user shares the same one.
-- `auth` is only what your route passed to `handler.fetch(request, { authInfo })` (see below). The
-  SDK never fills it from headers.
+- `auth` is exactly what `verifyToken` returned: `withMcpAuth` runs it and the toolkit hands the
+  result to `principal`. Nothing is read from headers on its own. Without `mcp-handler`, pass it
+  yourself with the SDK's `createMcpHandler`: `handler.fetch(request, { authInfo })`.
 - `principal` also receives `request`, for cookie or session apps: type its argument as `Caller`
   (`{ auth, request }`, from `@upstash/mcp-toolkit/tasks` or `/events`). `request` is unverified,
   so check the session yourself, and never trust a header like `x-user-id`.
@@ -65,7 +77,6 @@ export function principal({ auth }: { auth?: AuthInfo }): string {
 
 ```ts
 // lib/tasks.ts
-import { McpServer } from "@modelcontextprotocol/server";
 import { createTaskLayer } from "@upstash/mcp-toolkit/tasks";
 import { QStashDispatcher, RedisTaskStore } from "@upstash/mcp-toolkit/upstash";
 import * as z from "zod";
@@ -86,32 +97,26 @@ tasks.define(
     return { content: [{ type: "text", text: await writeReport(topic) }] };
   },
 );
-
-export function createServer() {
-  const server = new McpServer({ name: "reports", version: "1.0.0" });
-  tasks.register(server); // adds generate_report, task_status and task_cancel
-  return server;
-}
 ```
 
 ```ts
 // app/api/mcp/route.ts
-import { createMcpHandler, type AuthInfo } from "@modelcontextprotocol/server";
-import { createServer } from "../../lib/tasks";
+import { createMcpHandler, withMcpAuth } from "mcp-handler";
+import { verifyToken } from "@/lib/auth";
+import { tasks } from "@/lib/tasks";
 
-const handler = createMcpHandler(() => createServer());
+const handler = createMcpHandler((server) => {
+  tasks.register(server); // adds generate_report, task_status and task_cancel
+});
 
-export async function POST(request: Request) {
-  // Clerk, WorkOS, Auth0, your own. `principal` reads the user id from `authInfo.extra`.
-  const authInfo: AuthInfo | undefined = await verifyToken(request);
-  if (!authInfo) return new Response("Unauthorized", { status: 401 });
-  return handler.fetch(request, { authInfo });
-}
+// Runs verifyToken before any tool sees the request; no AuthInfo means 401.
+const authHandler = withMcpAuth(handler, verifyToken, { required: true });
+export { authHandler as GET, authHandler as POST };
 ```
 
 ```ts
 // app/api/execute/route.ts: QStash delivers the work here
-import { tasks } from "../../lib/tasks";
+import { tasks } from "@/lib/tasks";
 
 export const POST = tasks.createExecuteHandler();
 ```
@@ -238,6 +243,12 @@ can take longer.
   Without signing keys the route throws instead of running anything unverified.
 - When your handler throws, the task is not marked failed. Only the dispatcher marks it `failed`,
   and only after QStash has stopped retrying. The failed message stays in the QStash DLQ.
+- When your handler _returns_ a tool error (`isError: true`), the task is `failed` at once, with no
+  retry, and `task_status` returns your content with `isError`, as the synchronous tool would.
+- What a failure looks like to the model is `error.code` and `error.message`. The transport's
+  details (`error.data`: the QStash DLQ id, the Workflow run id) stay in Redis for you. What your
+  handler threw (Workflow's failure response) and a dispatch error are logged, never stored or
+  returned.
 - By default QStash tries 5 times with backoff `min(pow(3, retried) * 1000, 300000)`, about two
   minutes in total, so a task survives a server restart. The free tier and the local dev server
   allow at most 5 retries.
@@ -279,12 +290,12 @@ export const commentCreated = events.define("comment.created", {
 });
 ```
 
-Call `events.register(server)` in `createServer()`, next to `tasks.register(server)`, and add the
+Call `events.register(server)` in the `createMcpHandler` callback, next to `tasks.register(server)`, and add the
 route QStash delivers to:
 
 ```ts
 // app/api/events/route.ts
-import { events } from "../../lib/events";
+import { events } from "@/lib/events";
 
 export const POST = events.createDeliveryHandler();
 ```

packages/mcp-toolkit/src/tasks/backends/workflow.test.ts

@@ -6,8 +6,8 @@
  * `task.run(...)` becomes a journaled step under Workflow and a plain call without it.
  */
 import { Client as QStashClient } from "@upstash/qstash";
-import { WorkflowContext } from "@upstash/workflow";
-import { describe, expect, it } from "vitest";
+import { Client as WorkflowClient, WorkflowContext } from "@upstash/workflow";
+import { describe, expect, it, vi } from "vitest";
 import * as z from "zod";
 import { insideStep, WorkflowDispatcher, workflowRoute } from "./workflow.js";
 import { createTaskLayer } from "../core.js";
@@ -95,7 +95,41 @@ describe("WorkflowDispatcher", () => {
     await dispatcher.cancel("task-1");
 
     // Unlike a queue, a workflow run can be stopped mid-flight rather than only un-queued.
-    expect(cancelled).toEqual(["task-1"]);
+    expect(cancelled).toEqual(["wfr_task-1"]);
+  });
+
+  it("cancels the run it triggered, with the real Workflow client", async () => {
+    // The real client, only its HTTP stubbed: a stubbed client can't catch how the real one
+    // names runs (`trigger` prefixes `wfr_`, `cancel` doesn't).
+    const requests: { method: string; url: string; body: string; headers: string }[] = [];
+    vi.stubGlobal(
+      "fetch",
+      async (input: string | URL | Request, init?: Parameters<typeof fetch>[1]) => {
+        const url = String(input instanceof Request ? input.url : input);
+        const method = init?.method ?? "GET";
+        const headers = JSON.stringify(Object.fromEntries(new Headers(init?.headers).entries()));
+        requests.push({ method, url, body: String(init?.body ?? ""), headers });
+        return method === "DELETE"
+          ? Response.json({ cancelled: 1 })
+          : Response.json([{ messageId: "msg_1" }]);
+      },
+    );
+    try {
+      const client = new WorkflowClient({ token: "test-token", baseUrl: "https://qstash.test" });
+      const dispatcher = new WorkflowDispatcher({ url: URL, client });
+      const taskId = "6f1c2c5e-0000-4000-8000-000000000000";
+
+      await dispatcher.dispatch(task({ taskId }));
+      await dispatcher.cancel(taskId);
+
+      const runId = `wfr_${taskId}`;
+      const trigger = requests.find((r) => r.method !== "DELETE");
+      const cancel = requests.find((r) => r.method === "DELETE");
+      expect(`${trigger?.body}${trigger?.headers}`).toContain(runId);
+      expect(new globalThis.URL(cancel!.url).searchParams.get("workflowRunIds")).toBe(runId);
+    } finally {
+      vi.unstubAllGlobals();
+    }
   });
 });
 

packages/mcp-toolkit/src/tasks/backends/workflow.ts

@@ -63,14 +63,14 @@ export class WorkflowDispatcher implements TaskDispatcher<WorkflowContext<Workfl
       url: this.config.url,
       body: { taskId: task.taskId } satisfies WorkflowPayload,
       retries: this.config.retries,
-      // One run per task: Workflow refuses a run id that was already used.
+      // One run per task: Workflow refuses a run id that was already used. It prefixes this with
+      // `wfr_` itself, so the run is `runIdOf(taskId)`.
       workflowRunId: task.taskId,
     });
   }
 
   async cancel(taskId: string): Promise<void> {
-    // The run id is the task id.
-    await this.client().cancel(taskId);
+    await this.client().cancel(runIdOf(taskId));
   }
 
   /** The workflow endpoint. Its `failureFunction` settles the task `failed` once retries run out. */
@@ -97,10 +97,16 @@ export class WorkflowDispatcher implements TaskDispatcher<WorkflowContext<Workfl
       failureFunction: async ({ context, failStatus, failResponse }) => {
         const taskId = (context.requestPayload as WorkflowPayload | undefined)?.taskId;
         if (!taskId) return;
+        // `failResponse` is the thrown error's message: database errors, internal hosts, upstream
+        // responses. It goes to your logs, never into the task, which its owner can read.
+        console.error(
+          `[mcp-toolkit] task ${taskId} failed (workflow run ${context.workflowRunId}):`,
+          failResponse,
+        );
         await endpoints.fail(taskId, {
           code: INTERNAL_ERROR,
           message: `Workflow run failed${failStatus ? ` (status ${failStatus})` : ""}`,
-          data: { response: failResponse, workflowRunId: context.workflowRunId },
+          data: { workflowRunId: context.workflowRunId },
         });
       },
       // Workflow verifies only body and signature; binding the URL refuses a signature that
@@ -114,6 +120,12 @@ export class WorkflowDispatcher implements TaskDispatcher<WorkflowContext<Workfl
   }
 }
 
+/**
+ * The Workflow run that serves a task. `Client.trigger` turns the `workflowRunId` it is given into
+ * `wfr_<id>`, while `Client.cancel` takes the run id as is, so a cancel must add the prefix.
+ */
+export const runIdOf = (taskId: string): string => `wfr_${taskId}`;
+
 /**
  * The workflow's route function. Its first act is always a step: Workflow authorizes every request,
  * the failure callback included, by running the route function until its first step, and refuses

packages/mcp-toolkit/src/tasks/core.test.ts

@@ -543,6 +543,48 @@ describe("createTaskLayer over MCP", () => {
       expect(text(after)).toMatch(/permanent/);
     });
 
+    it("fails a task whose handler returns a tool error, and shows that error like a tool would", async () => {
+      live = await harness(async () => ({
+        isError: true,
+        content: [{ type: "text", text: "Workspace is read-only" }],
+      }));
+      const taskId = await live.start("x");
+      await live.dispatcher.drain();
+      const after = await live.status(taskId);
+      expect(after.isError).toBe(true);
+      expect(after.structuredContent).toMatchObject({
+        status: "failed",
+        // Settled by the layer, not by a retried throw (whose message would be the thrown one).
+        error: { message: "The task returned an error" },
+      });
+      expect(text(after)).toMatch(/Workspace is read-only/);
+    });
+
+    it("keeps a failure's details in the store but out of what the model sees", async () => {
+      live = await harness(steppedHandler(1, 1));
+      const taskId = await live.start("x");
+      await live.dispatcher.drain();
+      const stored = (await live.store.get(taskId))!;
+      await live.store.create({
+        ...stored,
+        status: "failed",
+        error: {
+          code: -32603,
+          message: "Workflow run failed",
+          data: { response: "pg: password auth failed" },
+        },
+      });
+      const after = await live.status(taskId);
+      expect(after.structuredContent?.error).toEqual({
+        code: -32603,
+        message: "Workflow run failed",
+      });
+      expect(JSON.stringify(after)).not.toMatch(/password auth/);
+      expect((await live.store.get(taskId))?.error?.data).toEqual({
+        response: "pg: password auth failed",
+      });
+    });
+
     it("fails the task when it cannot be dispatched, instead of leaving it working", async () => {
       const manual = new ManualDispatcher();
       manual.failNext = new Error("queue unavailable");
@@ -555,6 +597,9 @@ describe("createTaskLayer over MCP", () => {
       };
       const result = await live.call("generate_report", { topic: "x" });
       expect(result.isError).toBe(true);
+      // The transport's error goes to the logs, not to the model.
+      expect(text(result)).not.toMatch(/queue unavailable/);
+      expect(text(result)).toMatch(/could not be queued/);
       expect(await live.store.get(created!)).toMatchObject({
         status: "failed",
         statusMessage: "Could not be queued",

packages/mcp-toolkit/src/tasks/core.ts

@@ -250,7 +250,9 @@ export function createTaskLayer<TContext = unknown>(
           error: { code: INTERNAL_ERROR, message: "The task could not be dispatched" },
         })
         .catch(() => undefined);
-      throw error;
+      // The transport's error (a QStash response, a URL) is for your logs, not the model.
+      console.error(`[mcp-toolkit] could not dispatch task ${task.taskId}:`, error);
+      return errorResult("The task could not be queued. Try again later.");
     }
     return {
       content: [
@@ -294,6 +296,17 @@ export function createTaskLayer<TContext = unknown>(
 
     const result = await definition.handler(task.args, mergeContext(taskContext, context));
     // A cancel that landed meanwhile wins: settle refuses the second terminal write.
+    if (result?.isError === true) {
+      // The handler answered with a tool error, the way a synchronous tool would: the task failed,
+      // and its own content explains why. Not retried: it returned, it didn't throw.
+      await store.settle(taskId, {
+        status: "failed",
+        statusMessage: "Failed",
+        result,
+        error: { code: INTERNAL_ERROR, message: "The task returned an error" },
+      });
+      return;
+    }
     await store.settle(taskId, {
       status: "completed",
       statusMessage: definition.config.completedMessage ?? "Completed",
@@ -336,18 +349,28 @@ function errorResult(text: string): Record<string, unknown> {
   return { isError: true, content: [{ type: "text", text }] };
 }
 
-/** A status line, followed by the task's own result content once it completed. */
+/**
+ * A status line, followed by the task's own result content once it completed, or once it failed by
+ * returning a tool error. That last case also carries `isError`, as the synchronous tool would have.
+ */
 function statusResult(task: Task): Record<string, unknown> {
   const line = `Task ${task.taskId} is ${task.status}${task.statusMessage ? `: ${task.statusMessage}` : "."}`;
+  const handlerError = task.status === "failed" && task.result?.isError === true;
   const extra =
-    task.status === "completed" && Array.isArray(task.result?.content) ? task.result.content : [];
+    (task.status === "completed" || handlerError) && Array.isArray(task.result?.content)
+      ? task.result.content
+      : [];
   const text =
     task.status === "failed"
       ? `${line} Error: ${task.error?.message ?? "unknown"}`
       : task.status === "working"
         ? `${line} Check again in about ${seconds(task.pollIntervalMs)}.`
         : line;
-  return { content: [{ type: "text", text }, ...extra], structuredContent: toWire(task) };
+  return {
+    content: [{ type: "text", text }, ...extra],
+    structuredContent: toWire(task),
+    ...(handlerError ? { isError: true } : {}),
+  };
 }
 
 const seconds = (ms: number) => `${Math.max(1, Math.round(ms / 1000))}s`;
@@ -364,8 +387,12 @@ function mergeContext<TContext>(
   return Object.assign(supplied as object, taskContext) as TaskContext & TContext;
 }
 
-/** Strips the server-only fields, leaving what the model sees in `structuredContent`. */
+/**
+ * Strips the server-only fields, leaving what the model sees in `structuredContent`. That includes
+ * `error.data`: transports put the failing response there (a stack trace, a database error), which
+ * stays in the store for your logs.
+ */
 function toWire(task: Task): WireTask {
-  const { name: _name, args: _args, owner: _owner, ...wire } = task;
-  return wire;
+  const { name: _name, args: _args, owner: _owner, error, ...wire } = task;
+  return error ? { ...wire, error: { code: error.code, message: error.message } } : wire;
 }

packages/mcp-toolkit/src/tasks/types.ts

@@ -27,7 +27,7 @@ export type WireTask = {
   /** Retention window in milliseconds, from creation. */
   ttlMs: number;
   pollIntervalMs: number;
-  /** The tool result, once `completed`. */
+  /** The tool result, once `completed`, or once `failed` because the handler returned `isError`. */
   result?: Record<string, unknown>;
   /** Once `failed`. */
   error?: TaskError;