diff --git a/.changeset/mcp-toolkit-initial.md b/.changeset/mcp-toolkit-initial.md index a9095e5..55b176c 100644 --- a/.changeset/mcp-toolkit-initial.md +++ b/.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`, diff --git a/packages/mcp-toolkit/README.md b/packages/mcp-toolkit/README.md index 0dcf8c4..ca6a0bf 100644 --- a/packages/mcp-toolkit/README.md +++ b/packages/mcp-toolkit/README.md @@ -12,7 +12,7 @@ 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: @@ -20,6 +20,7 @@ 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 { + 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(); ``` diff --git a/packages/mcp-toolkit/src/tasks/backends/workflow.test.ts b/packages/mcp-toolkit/src/tasks/backends/workflow.test.ts index 470f872..3d1d7ac 100644 --- a/packages/mcp-toolkit/src/tasks/backends/workflow.test.ts +++ b/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[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(); + } }); }); diff --git a/packages/mcp-toolkit/src/tasks/backends/workflow.ts b/packages/mcp-toolkit/src/tasks/backends/workflow.ts index 08fdd3e..8d21576 100644 --- a/packages/mcp-toolkit/src/tasks/backends/workflow.ts +++ b/packages/mcp-toolkit/src/tasks/backends/workflow.ts @@ -63,14 +63,14 @@ export class WorkflowDispatcher implements TaskDispatcher { - // 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 { 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`, 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 diff --git a/packages/mcp-toolkit/src/tasks/core.test.ts b/packages/mcp-toolkit/src/tasks/core.test.ts index 6cdb4b7..5f71237 100644 --- a/packages/mcp-toolkit/src/tasks/core.test.ts +++ b/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", diff --git a/packages/mcp-toolkit/src/tasks/core.ts b/packages/mcp-toolkit/src/tasks/core.ts index 6871ea8..b66cb7e 100644 --- a/packages/mcp-toolkit/src/tasks/core.ts +++ b/packages/mcp-toolkit/src/tasks/core.ts @@ -250,7 +250,9 @@ export function createTaskLayer( 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( 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 { 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 { 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( 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; } diff --git a/packages/mcp-toolkit/src/tasks/types.ts b/packages/mcp-toolkit/src/tasks/types.ts index 949fabb..40600ea 100644 --- a/packages/mcp-toolkit/src/tasks/types.ts +++ b/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; /** Once `failed`. */ error?: TaskError;