> ## Documentation index
> Fetch the complete documentation index at: https://upstash.com/docs/llms.txt
> Ask it with GET https://upstash.com/docs/ask?q=<keywords or question>. Short keywords rank best.
> Use these to discover all available pages before exploring further.
> Feature requests: https://upstash.com/docs/common/help/feature-requests.md lists what users asked for and how to vote for a request or add one.
> Page rating: `GET https://upstash.com/docs/feedback?path=/docs/redis/sdks/agentkit/mcp-tasks&helpful=true` (or `false`) records whether this page helped. An optional `comment` (URL-encoded, at most 500 characters) says what was missing or wrong. It is anonymous and needs no key. Comments hold no code, secrets or personal data.

# Long-Running MCP Tools with Redis and QStash

> Long-running tools for MCP servers: start in memory with no setup, then make tasks durable with Upstash Redis and QStash or Workflow. Clients that support MCP Tasks poll natively, and every other client gets a task id the model can poll.

An MCP tool call is a request that waits for its response. When the work takes minutes, something
gives up first: the client's tool timeout, your serverless function's limit, or the model.

`@upstash/mcp-toolkit/tasks` turns a slow tool into a task: the call returns at once and the work
carries on in the background. Clients that support the
[MCP Tasks extension](https://modelcontextprotocol.io/extensions/tasks/overview) (Claude Code) get a
native task, which they poll and show themselves. Every other client gets a task id in an ordinary
tool result, and the model checks on it with the `task_status` and `task_cancel` tools.

Tasks run in memory with nothing else to set up. Add Upstash Redis and
[QStash](/qstash/overall/getstarted) (or [Upstash Workflow](/workflow/getstarted)) to make them
durable. The toolkit sits on top of the official MCP TypeScript SDK and uses WebCrypto only, so it
runs on Node and edge runtimes.

```bash
npm install @upstash/mcp-toolkit @modelcontextprotocol/server zod
```

## An MCP server with a slow tool

```ts
// mcp.ts
import { createMcpHandler, McpServer } from "@modelcontextprotocol/server";
import { McpTasks } from "@upstash/mcp-toolkit/tasks";
import * as z from "zod";

// No store or dispatcher: tasks are kept in memory and run in this process.
const tasks = new McpTasks({ principal: () => "local" });

// Register at module scope; the handler runs after the tool call has answered.
tasks.registerTask(
  "generate_report",
  {
    description: "Writes a report on a topic.",
    inputSchema: z.object({ topic: z.string() }),
  },
  async ({ topic }, task) => {
    for (const step of ["Reading sources", "Drafting", "Polishing"]) {
      if (await task.isCancelled()) return {};
      await task.update(step); // progress, shown to whoever polls
      // the slow part
      await new Promise((resolve) => setTimeout(resolve, 2000));
    }
    return { content: [{ type: "text", text: `A report on ${topic}.` }] };
  },
);

const mcp = createMcpHandler(() => {
  const server = new McpServer({ name: "reports", version: "1.0.0" });
  // generate_report, task_status, task_cancel, tasks/get and tasks/cancel
  tasks.attach(server);
  return server;
});

// Serve POST /mcp with it.
export const handler = (request: Request): Promise<Response> =>
  mcp.fetch(request);
```

`handler` is a Web-standard `(request: Request) => Promise<Response>`, so it mounts in any
framework, or in Node through an adapter such as `@hono/node-server`:

<CodeGroup>
```ts Hono
app.post("/mcp", (c) => handler(c.req.raw));
```

```ts Next.js
// app/mcp/route.ts
export const POST = handler;
```

```ts Bun
Bun.serve({ fetch: handler });
```
</CodeGroup>

That is the whole setup: no database, no queue and no other endpoint. `topic` is typed from
`inputSchema` (Zod, Valibot and ArkType schemas all work). The catch is that the tasks live in this
process. See [Make it durable](#make-it-durable-with-redis-and-qstash) before you deploy to
serverless or run more than one instance.

## What the client sees

`tools/list` has `generate_report`, `task_status` and `task_cancel`, and the server advertises the
Tasks extension (`io.modelcontextprotocol/tasks`, protocol revision 2026-07-28). On every call to
`generate_report`, `McpTasks` checks whether the request declares the extension:

- **A client that declares it** gets a native task (`resultType: "task"`). The client polls
  `tasks/get` itself, shows the progress, and hands the model the handler's result when it is done,
  so the model spends no turns polling. `tasks/cancel` stops it. In Claude Code, the call returns at
  once and Claude keeps working; the footer shows **1 MCP task** while it runs, and the result
  arrives in the conversation when the task completes. See
  [MCP Tasks for server authors](https://code.claude.com/docs/en/mcp#mcp-tasks-for-server-authors).
- **Any other client** gets the task id in an ordinary tool result. The model polls `task_status`
  (progress, then the handler's result) and can stop the task with `task_cancel`.

Both read the same task record.

Which clients declare the extension:

- **Claude Code** CLI v2.1.304 and later, on calls from the main conversation in interactive
  terminal sessions, and in Agent SDK and GitHub Actions runs with `CLAUDE_AUTO_BACKGROUND_TASKS=1`.
  It does not on calls from subagents, in one-shot `claude -p` runs, in the VS Code extension or the
  desktop app, or on third-party providers such as Bedrock; those calls get the tool-based handle.
- **Codex, Cursor, OpenCode and ChatGPT** don't declare it (as of October 2026) and get the
  tool-based handle.

`task_status` and `task_cancel` stay listed alongside native tasks, because they are the fallback
and one client can send both kinds of call. Claude Code shows the model a native task's id, so the
model may call `task_status` on it too; both paths give the same answer.

<Accordion title="Native task responses">
  ```jsonc
  // tools/call generate_report, from a client that declared the extension
  { "resultType": "task", "taskId": "0e30…", "status": "working",
    "statusMessage": "Queued", "createdAt": "…", "lastUpdatedAt": "…",
    "ttlMs": 86400000, "pollIntervalMs": 2000 }

  // tasks/get { taskId }: the same fields, plus `result` (the tool result) once
  // completed, or `error` once failed
  { "resultType": "complete", "taskId": "0e30…", "status": "completed",
    "statusMessage": "Completed", …,
    "result": { "content": [{ "type": "text", "text": "Report on coffee" }] } }

  // tasks/cancel { taskId }
  { "resultType": "complete" }
  ```

  A failed task carries a JSON-RPC `error`. An unknown, expired or malformed task id, another
  user's task, and an unidentified caller all get the same `-32602` error, so ids can't be probed.
</Accordion>

<Accordion title="Tool-based responses">
  ```jsonc
  // generate_report: a handle, right away
  { "content": [{ "type": "text",
                  "text": "Started task 0e30…. Call task_status with …" }],
    "structuredContent": { "taskId": "0e30…", "status": "working",
                           "ttlMs": 86400000, "pollIntervalMs": 2000 } }

  // task_status: progress…
  { "content": [{ "type": "text",
                  "text": "Task 0e30… is working: Reading sources. …" }],
    "structuredContent": { "taskId": "0e30…", "status": "working",
                           "statusMessage": "Reading sources" } }

  // …then the handler's own content, as if the tool had run synchronously
  { "content": [{ "type": "text", "text": "Task 0e30… is completed: …" },
                { "type": "text", "text": "Report on coffee" }],
    "structuredContent": { "taskId": "0e30…", "status": "completed",
                           "result": { "content": [ … ] } } }
  ```

  The states are `working`, `input_required`, `completed`, `failed` and `cancelled`; the last three
  are final. If the model stops polling, nothing is lost: the work finishes anyway, and the result
  can be read until the task expires (1 day by default; set `defaults: { ttlMs }` on `McpTasks`).
</Accordion>

## Make it durable with Redis and QStash

In memory, a task is lost when the process restarts, and it only exists on the instance that took
the call. On serverless, or behind several instances, the next `tasks/get` may reach an instance
that never saw the task, and work left running after the response can be frozen or killed with the
function. Keep the records in Upstash Redis and deliver the work through QStash instead: the record
is shared by every instance, QStash retries a failed run, and each run gets a fresh invocation.

Pass a store and a dispatcher; `registerTask` and the MCP handler stay exactly as they were:

```ts mcp.ts
import { McpTasks } from "@upstash/mcp-toolkit/tasks";
import { QStashDispatcher, RedisTaskStore } from "@upstash/mcp-toolkit/upstash";

const tasks = new McpTasks({
  store: new RedisTaskStore(),
  dispatcher: new QStashDispatcher({
    url: `${process.env.APP_URL}/api/execute`,
  }),
  principal: () => "local",
});

// registerTask(...) and the MCP handler as above, then one more endpoint.
// QStash delivers each task here: serve POST /api/execute with it.
export const execute: (request: Request) => Promise<Response> =
  tasks.createExecuteHandler();
```

The execute endpoint is another `(request: Request) => Promise<Response>`, mounted like the MCP one,
at the URL you gave the dispatcher. It verifies the QStash signature, runs the task's handler and
stores the result. Every instance that serves it must import the module that calls
`tasks.registerTask(...)`, so it knows the handlers.

The Redis store and the QStash dispatcher read their credentials from the environment:

```bash .env
UPSTASH_REDIS_REST_URL=...
UPSTASH_REDIS_REST_TOKEN=...
QSTASH_URL=...                   # your QStash region's URL, from the console
QSTASH_TOKEN=...
# Required: the execute endpoint refuses to run without them.
QSTASH_CURRENT_SIGNING_KEY=...
QSTASH_NEXT_SIGNING_KEY=...
APP_URL=https://your-app.com     # QStash has to be able to reach it
```

For work that can outlast one function invocation, use `WorkflowDispatcher` (same `url`, same
endpoint): each step runs in its own invocation, so a task has no overall time limit. See
[Work longer than one function invocation](#work-longer-than-one-function-invocation).

<Note>
  For local development, run the [QStash dev server](/qstash/howto/local-development) with
  `npx @upstash/qstash-cli dev`, set `QSTASH_URL=http://127.0.0.1:8080` and the credentials it prints,
  and point `APP_URL` at your local server, for example `http://127.0.0.1:3000`.
</Note>

|                                    | in memory (default)         | `QStashDispatcher`         | `WorkflowDispatcher`            |
| ---------------------------------- | --------------------------- | -------------------------- | ------------------------------- |
| Needs                              | nothing                     | Redis, QStash, an endpoint | Redis, QStash, an endpoint      |
| Serverless, several instances      | no                          | yes                        | yes                             |
| Survives the process dying         | no                          | yes (QStash redelivers)    | yes (replayed from the journal) |
| Can run longer than one invocation | n/a: as long as the process | no                         | yes, one invocation per step    |
| When the handler throws            | `failed`, no retry          | the whole task starts over | only the failed step runs again |
| Ask the user (`task.elicit`)       | yes                         | no                         | yes                             |

In production (`NODE_ENV=production`), the in-memory defaults log a one-time warning. If a single
long-lived server is what you want, pass them explicitly to say so:
`new McpTasks({ store: new MemoryTaskStore(), dispatcher: new InProcessDispatcher(), principal })`
(both from `@upstash/mcp-toolkit/tasks`). `await dispatcher.drain()` on shutdown lets running tasks
finish.

## Users

`principal` is a function that returns the id of the user making the call. Each task records who
started it, and `task_status`, `task_cancel`, `tasks/get` and `tasks/cancel` only answer for that
caller; another user's task id reads exactly like an unknown one. `() => "local"` is for a server
with no users of its own. Otherwise, verify the token and hand the result to the SDK, which passes it
to `principal`:

```ts auth.ts
import type { AuthInfo } from "@modelcontextprotocol/server";
import type { Caller } from "@upstash/mcp-toolkit/tasks";

// 1. Your token check. Clerk, WorkOS, Auth0, your own.
export async function verifyToken(
  request: Request,
): Promise<AuthInfo | undefined> {
  const token = request.headers.get("authorization")?.replace(/^Bearer /, "");
  if (!token) return undefined;
  const { sub, client_id } = await verifyJwt(token);
  return { token, clientId: client_id, scopes: [], extra: { userId: sub } };
}

// 2. The toolkit calls this with that AuthInfo. Return the user id or throw.
export function principal({ auth }: Caller): string {
  const userId = auth?.extra?.userId;
  if (typeof userId !== "string") throw new Error("Not authenticated");
  return userId;
}
```

```ts mcp.ts
// new McpTasks({ principal, ... }), and the handler checks the token first
export async function handler(request: Request): Promise<Response> {
  const authInfo = await verifyToken(request);
  if (!authInfo) return new Response("Unauthorized", { status: 401 });
  return mcp.fetch(request, { authInfo });
}
```

[`mcp-handler`](https://www.npmjs.com/package/mcp-handler) is Vercel's adapter that serves an MCP
server as a route handler in Next.js and other Fetch-based frameworks. If you serve the server with
it, its `withMcpAuth` checks the bearer token for you, so you don't write the handler above:

```ts mcp.ts
// served by mcp-handler instead
import { McpTasks } from "@upstash/mcp-toolkit/tasks";
import { createMcpHandler, withMcpAuth } from "mcp-handler";
import { principal, verifyToken } from "./auth";

const tasks = new McpTasks({ principal });
// tasks.registerTask(...) as before

const handler = createMcpHandler((server) => {
  tasks.attach(server);
});

// Runs verifyToken first; no AuthInfo means 401. mcp-handler also passes
// the bearer token, as verifyToken(request, token).
const authHandler = withMcpAuth(handler, verifyToken, { required: true });
export { authHandler as GET, authHandler as POST };
```

<Accordion title="More on principal">
  - **A throw refuses the call.** There is no anonymous mode: if `principal` throws, rejects, or
    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 exactly the `authInfo` your handler passed to `fetch`. Nothing is read from headers on
    its own.
  - `principal` also receives `request`, for cookie or session apps (`Caller` is `{ auth, request }`),
    and may be async. `request` is unverified, so check the session yourself, and never trust a
    header like `x-user-id`.
</Accordion>

### Check what the arguments point at

Owning the task is not the same as being allowed to touch what it works on. The model fills in the
tool arguments, so a `workspaceId` in them is whatever it was told. Give the task an `authorize`,
which runs when the tool is called, before anything is stored or queued, and use `task.principal`
(the user who started the task) inside the handler. Never take a user id from the arguments:

```ts
tasks.registerTask(
  "export_workspace",
  {
    description: "Exports a workspace to CSV.",
    inputSchema: z.object({ workspaceId: z.string() }),
    // May this user export this workspace? `false` refuses the call.
    authorize: ({ workspaceId }, { principal }) =>
      canExport(principal, workspaceId),
  },
  async ({ workspaceId }, task) => {
    const csv = await exportWorkspace(workspaceId, { as: task.principal });
    return { content: [{ type: "text", text: csv }] };
  },
);
```

`authorize(args, { principal, auth, request })` gets the validated arguments, the caller's id and
the same `auth` and `request` as `principal`. It is the only place a task sees the caller's token:
the handler runs after the tool call has answered (with QStash, in a delivery request that carries
no user auth), so `task.principal` is how it knows who it is working for.

**Errors and limits.** When `authorize` or a store throws, the caller sees only `Internal error. Try
again later.` (a tool error, or `-32603` on `tasks/*` and `events/*`) and the real error goes to
`console.error`, so a database host or credential in it never reaches the model. Each user may have
at most 100 tasks working or waiting for input (`maxActiveTasks`), and a task's arguments may be at
most 64 KiB of JSON (`maxArgsBytes`); past either, the call is refused before anything is stored.
A task waiting for an answer counts as active until it is answered, cancelled or expires.

## Cancellation

`task_cancel` and `tasks/cancel` mark the task `cancelled`. With Workflow the run is cancelled too;
with QStash a later redelivery finds the task settled and does nothing. Cancelling is cooperative,
so running code only stops where it checks `await task.isCancelled()`, as the loop in the
[first example](#an-mcp-server-with-a-slow-tool) does between steps. `isCancelled()` also
returns `true` once the task has expired.

<Warning>
  Arguments are validated when the tool is called, then stored as JSON and handed to the handler
  later. Values that don't survive JSON, such as a `Date` produced by `z.coerce.date()`, arrive as
  strings. Keep task schemas to plain values, or parse them again inside the handler.
</Warning>

## Ask the user from a task

A task can stop partway, ask the user a question, and carry on with the answer. MCP calls this
[elicitation](https://modelcontextprotocol.io/specification/2026-07-28/client/elicitation): the
client shows the user a form built from a schema and sends back what they entered.

```ts
tasks.registerTask(
  "publish_post",
  {
    description: "Publishes a post once the user approves it.",
    inputSchema: z.object({ title: z.string() }),
  },
  async ({ title }, task) => {
    const answer = await task.elicit({
      message: `Publish "${title}" now?`,
      requestedSchema: z.object({ publish: z.boolean() }),
    });
    // `content` is typed and validated against requestedSchema
    if (answer.action === "accept" && answer.content.publish) {
      return { content: [{ type: "text", text: "Published." }] };
    }
    return { content: [{ type: "text", text: "Not published." }] };
  },
);
```

`answer.action` is `"accept"` (with `content`), `"decline"` or `"cancel"`. While it waits, the task
is `input_required`; the client answers with `tasks/update` and the handler carries on.

- **Backends.** Works in memory and with `WorkflowDispatcher` (which waits without holding a
  function open; call `elicit` between steps, not inside `task.run`). Not with `QStashDispatcher`:
  there, `task` has no `elicit` in the types (a plain-JS call fails the task).
- **Claude Code** shows the form in interactive terminal sessions. In headless sessions it can't, so
  it cancels the task.
- **Clients without native tasks** get the question from `task_status` if they support
  elicitation on a 2026-07-28 connection. Otherwise, including on 2025-era connections (the default
  in many clients, the official SDK client 2.3.1 among them), the model is told the task is waiting
  and that only the user can answer. There is no tool for the model to answer in their place.
- If the task is cancelled or expires while waiting, `elicit` throws; don't catch that error. A
  task asks at most 16 questions, one at a time.

## Tool-based handle for every client

To answer every client with the tool-based handle, set `mode: "tools"`. The server then serves no
`tasks/*` methods and doesn't advertise the extension:

```ts
const tasks = new McpTasks({ principal, mode: "tools" }); // default: "native"
```

Use it when something depends on the task id appearing in the tool result, such as a client
integration or test that reads `structuredContent.taskId`, or when you want every client to see the
same tools and responses.

## Work longer than one function invocation

With `QStashDispatcher`, the whole handler runs in one serverless invocation. If it goes past your
platform's time limit, it is killed, and the retry starts the handler from the beginning.

`WorkflowDispatcher` runs each step in its own invocation and replays finished steps from a journal,
so a task has no overall time limit. Only the dispatcher changes; the endpoints stay the same. The
handler's `task` then also has `run`, `sleep` and `call`:

```ts
import {
  RedisTaskStore,
  WorkflowDispatcher,
} from "@upstash/mcp-toolkit/upstash";

const tasks = new McpTasks({
  store: new RedisTaskStore(),
  dispatcher: new WorkflowDispatcher({
    url: `${process.env.APP_URL}/api/execute`,
  }),
  principal,
});

tasks.registerTask(
  "migrate_workspace",
  {
    description: "Copies a workspace to new storage.",
    inputSchema: z.object({ workspaceId: z.string() }),
    authorize: ({ workspaceId }, { principal }) =>
      isOwner(principal, workspaceId),
  },
  async ({ workspaceId }, task) => {
    const batches = await task.run("plan", () =>
      listBatches(workspaceId, task.principal),
    );
    for (const [i, batch] of batches.entries()) {
      if (await task.isCancelled()) return {};
      await task.update(`Copying batch ${i + 1}/${batches.length}`);
      await task.run(`copy-${i}`, () => copyBatch(batch));
    }
    return { content: [{ type: "text", text: "Done" }] };
  },
);
```

<Accordion title="Rules for Workflow handlers">
  The handler runs again from the top on every step, and finished steps are replayed from the
  journal:

  - Put the work inside `task.run`. That makes it run once and survive a crash.
  - `task.update(...)` does not need wrapping.
  - Keep `task.isCancelled()` outside steps. It has to run again each time, or a later cancel is
    never seen.
  - Don't nest `task.run` calls.
  - Each step must still fit within your function's time limit.

  The task's TTL starts when the task is created and is never extended. When it runs out, the record
  is deleted and `isCancelled()` returns true. The default is 1 day; set `ttlMs` higher for work that
  can take longer.

  With `WorkflowDispatcher`, cancelling a task also cancels its run; with QStash, a running task stops
  at its next `isCancelled()`.
</Accordion>

## Retries and failures

<Accordion title="How tasks fail, and how deliveries are verified and retried">
  - **In memory** (`InProcessDispatcher`) there are no retries: when your handler throws, the task is
    `failed` at once, with the message "The task handler threw an error". What it threw is logged and
    kept in the record's `error.data`, never shown to the model.
  - The execute endpoint answers **200** when the task ran (or had already finished), **500** when
    your handler threw, so QStash tries again, and **489** with `Upstash-NonRetryable-Error` when the
    signature or body is bad.
  - Every delivery's QStash signature is checked against the URL you gave the dispatcher, not
    `request.url`, so it works behind a proxy and a signature issued for another endpoint is refused.
    Without signing keys (or a `receiver` you pass), the route throws on its first request instead of
    running anything unverified.
  - With QStash or Workflow, 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 and a dispatch error are never returned. QStash and Workflow only log them; the
    in-memory dispatcher also keeps the message in `error.data`.
  - By default QStash retries 5 times (6 attempts in total) 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.
</Accordion>

## Security and storage

<Accordion title="Who can see what">
  The server sets a task's owner from `principal`, and no tool argument can set it. Whether a caller
  may start a task with given arguments is your `authorize`; the handler gets the owner as
  `task.principal`. `task_status`, `task_cancel`, `tasks/get` and `tasks/cancel` only accept UUID task
  ids, and compare the stored owner with the caller. For anyone else, a task looks exactly like an
  unknown id (a tool error for the tools, `-32602` for `tasks/*`). Task ids are random UUIDs. Nothing
  lists tasks, and the execute endpoint only accepts signed QStash deliveries and never returns a
  task.

  Anyone with write access to your Redis can change an owner, so treat the Redis credentials like any
  other production secret.
</Accordion>

<Accordion title="What is stored in Redis">
  **`mcp:task:<taskId>`** is a hash with one JSON-encoded field per property: `taskId`, `name`,
  `args`, `owner`, `status`, `statusMessage`, `result` / `error`, `createdAt`, `lastUpdatedAt`,
  `ttlMs`, `pollIntervalMs`, and, for a task that asks the user, `inputRequests` (the open question)
  and one `response:<key>` field per answer.

  - `args`, `result` and the user's answers are **plain JSON**. Keep secrets out of tool arguments,
    results and forms, or use a short `ttlMs`.
  - The task is written with its TTL before the tool replies, because the model's next poll may reach
    another instance. The TTL counts from creation and is never extended.
  - Status changes go through one guarded Lua script, and the first final status wins, so a completion
    can't overwrite a cancel. Each property is its own field, so a progress update and a cancel never
    overwrite each other.
  - A Redis client built with `automaticDeserialization: false` is not supported.

  **`mcp:task:active:<owner>`** is a sorted set of the owner's task ids, scored by expiry, for
  `maxActiveTasks`. Counting drops the expired and finished ones.

  A task message in QStash carries only `{ taskId }`. The toolkit never stores the caller's token or
  the request.
</Accordion>

## Options

<Accordion title="All options">
  **`new McpTasks(options)`**: `principal` is required. Optional: `store` (a `MemoryTaskStore` by default)
  and `dispatcher` (an `InProcessDispatcher` by default; a dispatcher needs a store with it), `mode`
  (`"native"`, or `"tools"` for the tool-based handle only), `maxActiveTasks` (100 per user working
  or `input_required`; `Infinity` turns it off), `maxArgsBytes` (64 KiB of JSON), `defaults.ttlMs`
  (1 day) and
  `defaults.pollIntervalMs` (2 seconds), in positive whole milliseconds.

  **`tasks.registerTask(name, config, handler)`**: `description` and `inputSchema` are required.
  Optional: `title`, `completedMessage` (defaults to `"Completed"`),
  `authorize(args, { principal, auth, request })`. Call it at module scope. `tasks.attach(server)`
  adds the task tools to a server, and `tasks.createExecuteHandler()` is the dispatcher's endpoint.

  **`RedisTaskStore`**: `redis` (defaults to one from env), `prefix` (`mcp:task:`), `enableTelemetry`.

  **`QStashDispatcher`**: `url` is required. Optional: `qstash`, `receiver`, `retries` (5),
  `retryDelay`, `enableTelemetry`.

  **`WorkflowDispatcher`**: `url` is required. Optional: `client`, `qstash`, `receiver`, `retries`,
  `enableTelemetry`.

  `receiver` defaults to one built from the `QSTASH_*_SIGNING_KEY` variables. The Redis and QStash
  clients get `@upstash/mcp-toolkit@<version>` added to their `Upstash-Telemetry-Sdk` header; turn it
  off with `enableTelemetry: false` or `UPSTASH_DISABLE_TELEMETRY`.
</Accordion>

<Accordion title="Custom backends">
  `@upstash/mcp-toolkit/tasks` doesn't import anything from Upstash. A backend implements one of these
  interfaces (types exported from the same entry point). `MemoryTaskStore` and `InProcessDispatcher`
  are small examples:

  ```ts
  interface TaskStore {
    create(task: Task): Promise<void>; // with its TTL
    get(taskId: string): Promise<Task | null>;
    // ignored once the task is final
    update(taskId: string, patch: TaskPatch): Promise<void>;
    settle(
      taskId: string,
      patch: TaskPatch & { status: TerminalTaskStatus },
    ): Promise<Task | null>; // first final status wins
    // records answers while input_required; resolves the keys it resumed
    respond(
      taskId: string,
      responses: Record<string, TaskInputResponse>,
      statusMessage: string,
    ): Promise<string[]>;
    // working + input_required, for maxActiveTasks
    countActive(owner: string): Promise<number>;
  }

  // CanAsk: whether handlers get task.elicit. TContext: what the transport
  // adds to `task` (WorkflowDispatcher: TaskDispatcher<true, WorkflowContext>).
  // McpTasks reads both off the dispatcher you pass.
  type TaskDispatcher<CanAsk extends boolean = false, TContext = unknown> = {
    dispatch(task: Task): Promise<void>; // called once per task
    cancel(taskId: string): Promise<void>;
    createExecuteHandler(
      endpoints: TaskEndpoints<TContext>,
    ): (request: Request) => Promise<Response>;
    // called once, by new McpTasks(...)
    connect?(endpoints: TaskEndpoints<TContext>): void;
  } & (CanAsk extends true
    ? {
        // required with CanAsk = true; ruled out otherwise
        waitForInput(wait: TaskInputWait, context: TContext): Promise<void>;
        resume(taskId: string, keys: string[]): Promise<void>;
      }
    : { waitForInput?: never; resume?: never });
  ```
</Accordion>

## Not supported yet

- Asking the user with `QStashDispatcher` (use `WorkflowDispatcher`), and input requests other
  than form elicitation, such as URL elicitation.
- Listing tasks: `tasks/list` and `notifications/tasks`.
- `tasks/result`. A completed task's result comes with `tasks/get`.

## Example

The [MCP toolkit demo](https://github.com/upstash/agentkit/tree/main/examples/mcp-toolkit-demo) is a
Next.js app that runs the same tool on QStash and on Workflow, with a live log of every MCP request.