diff --git a/docs.json b/docs.json index 98369b6d5..134daf489 100644 --- a/docs.json +++ b/docs.json @@ -748,7 +748,9 @@ "pages": [ "redis/sdks/agentkit/ai-sdk", "redis/sdks/agentkit/eve", - "redis/sdks/agentkit/tanstack-ai" + "redis/sdks/agentkit/tanstack-ai", + "redis/sdks/agentkit/mcp-tasks", + "redis/sdks/agentkit/mcp-events" ] }, "redis/sdks/mcp", @@ -894,7 +896,9 @@ "pages": [ "redis/sdks/agentkit/ai-sdk", "redis/sdks/agentkit/eve", - "redis/sdks/agentkit/tanstack-ai" + "redis/sdks/agentkit/tanstack-ai", + "redis/sdks/agentkit/mcp-tasks", + "redis/sdks/agentkit/mcp-events" ] } ] @@ -1687,6 +1691,8 @@ "pages": [ "qstash/sdks/ts/overview", "qstash/sdks/ts/gettingstarted", + "qstash/sdks/ts/mcp-tasks", + "qstash/sdks/ts/mcp-events", { "group": "Examples", "pages": [ @@ -2870,6 +2876,14 @@ { "source": "/qstash/api/url-groups/remove", "destination": "/qstash/api-reference/url-groups/delete-a-url-group" + }, + { + "source": "/qstash/sdks/ts/mcp-tasks", + "destination": "/redis/sdks/agentkit/mcp-tasks" + }, + { + "source": "/qstash/sdks/ts/mcp-events", + "destination": "/redis/sdks/agentkit/mcp-events" } ] } diff --git a/llms-full.txt b/llms-full.txt index 55526efc8..621b5e728 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -15986,7 +15986,7 @@ To view a more detailed Next.js quick start guide for setting up QStash, refer t It's also possible to schedule a background job to run at a later time using [schedules](/docs/qstash/features/schedules). -If you'd like to invoke another endpoint when the background job is complete, you can use [callbacks](/docs/qstash/features/callbacks). +If you'd like to invoke another endpoint when the background job is complete, you can use [callbacks](/docs/qstash/features/callbacks). # Batching Source: https://upstash.com/docs/qstash/features/batch @@ -16285,7 +16285,7 @@ client.message.batch_json( "messageId": "msg_..." } ] -``` +``` # Callbacks Source: https://upstash.com/docs/qstash/features/callbacks @@ -17340,7 +17340,7 @@ from qstash import QStash client = QStash("") client.queue.get("my-queue") ``` - + # Retry Source: https://upstash.com/docs/qstash/features/retry @@ -18039,7 +18039,7 @@ via the [REST API](/docs/qstash/api-reference/dlq/retry-a-dlq-message). ## Retrying Early -A message that is waiting out its retry delay can be retried immediately from the console. See [Run a Delayed Message Now](/docs/qstash/howto/retry-now). +A message that is waiting out its retry delay can be retried immediately from the console. See [Run a Delayed Message Now](/docs/qstash/howto/retry-now). # Local Development Source: https://upstash.com/docs/qstash/howto/local-development @@ -21747,6 +21747,20 @@ const client = new Client({ }); ``` +# MCP Events with QStash +Source: https://upstash.com/docs/qstash/sdks/ts/mcp-events + + + MCP Events for your server: subscriptions in Upstash Redis, signed webhook deliveries retried by QStash. + + +# Long-Running MCP Tools with QStash +Source: https://upstash.com/docs/qstash/sdks/ts/mcp-tasks + + + Durable long-running MCP tools on QStash or Upstash Workflow, with the task record in Upstash Redis. + + # Overview Source: https://upstash.com/docs/qstash/sdks/ts/overview @@ -87166,6 +87180,1452 @@ Two complete `eve` apps live in the AgentKit repo: +# MCP Events with Redis and QStash +Source: https://upstash.com/docs/redis/sdks/agentkit/mcp-events + +Most agents only act when someone messages them. [MCP Events](https://github.com/modelcontextprotocol/experimental-ext-triggers-events) +lets an agent wait for something instead: the host subscribes to an event on your MCP server, and +when it happens your server POSTs a signed webhook that wakes the agent up. + +`@upstash/mcp-toolkit/events` implements the server side. You define events with typed payloads and +call `emit` when something happens. The toolkit handles subscriptions, callback verification, +signing and retries: subscriptions live in Upstash Redis, and every delivery goes through +[QStash](/docs/qstash/overall/getstarted), which retries failed webhooks with backoff. + + + MCP Events is a draft. As of October 2026, ChatGPT is the only widely used host that subscribes to + it (webhook delivery, in Work chats), following [OpenAI's MCP Events guide](https://developers.openai.com/plugins/build/mcp-events). + Codex supports events only for OpenAI's own connectors. Claude Code, Cursor and OpenCode don't + subscribe yet. + + +```bash +npm install @upstash/mcp-toolkit @modelcontextprotocol/server mcp-handler zod +``` + +## Quickstart + +Four files in a Next.js app. + +### 1. Identify the caller + +Every subscription belongs to the user who subscribed. `verifyToken` turns the request's bearer token +into the SDK's `AuthInfo`, with your user id in `extra`, and `principal` reads that id back for the +toolkit. If you also use [MCP Tasks](/docs/redis/sdks/agentkit/mcp-tasks), both layers share this file: + +```ts lib/auth.ts +import type { AuthInfo } from "@modelcontextprotocol/server"; + +// 1. Runs on every MCP request (wired up with withMcpAuth in step 3). +// 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"); + return userId; +} +``` + +### 2. Define the event layer and your events + +```ts lib/events.ts +import { createEventLayer } from "@upstash/mcp-toolkit/events"; +import { QStashDelivery, RedisSubscriptionStore } from "@upstash/mcp-toolkit/upstash"; +import * as z from "zod"; +import { principal } from "./auth"; + +export const events = createEventLayer({ + store: new RedisSubscriptionStore(), + delivery: new QStashDelivery({ url: `${process.env.APP_URL}/api/events` }), + principal, +}); + +export const commentCreated = events.define("comment.created", { + description: "A new comment was added to a document.", + input: z.object({ documentId: z.string().optional() }), // what a subscriber may filter on + payload: z.object({ documentId: z.string(), text: z.string() }), + // May this user see comments on this document? + authorize: (args, { principal }) => canRead(principal, args.documentId), +}); +``` + +`authorize` is required. See [Matching and authorize](#matching-and-authorize) for when it runs. + +### 3. Serve the MCP endpoint + +```ts app/api/mcp/route.ts +import { createMcpHandler, withMcpAuth } from "mcp-handler"; +import { verifyToken } from "../../../lib/auth"; +import { events } from "../../../lib/events"; + +const handler = createMcpHandler((server) => { + events.register(server); // adds events/list, events/subscribe and events/unsubscribe +}); + +// Runs verifyToken before any request is handled; no AuthInfo means 401. +const authHandler = withMcpAuth(handler, verifyToken, { required: true }); +export { authHandler as GET, authHandler as POST }; +``` + +`withMcpAuth` hands the `AuthInfo` from `verifyToken` to the MCP server, and the toolkit passes it +to `principal` on every subscribe. With [MCP Tasks](/docs/redis/sdks/agentkit/mcp-tasks) too, call +`tasks.register(server)` in the same callback. + +### 4. Add the delivery endpoint + +```ts app/api/events/route.ts +import { events } from "../../../lib/events"; + +export const POST = events.createDeliveryHandler(); +``` + +QStash delivers each event here. The handler verifies the QStash signature, signs the envelope with +the subscriber's secret and POSTs it to the host. + +### Emit + +Call `emit` wherever the change happens: a route, a webhook from your own app, a background job. + +```ts +await commentCreated.emit({ documentId: "doc_123", text: "Ship it?" }); +``` + +`emit` is typed by the `payload` schema and validates against it. Every subscription that matches +the payload, and that `authorize` still allows for `documentId: "doc_123"`, gets a signed webhook. +It returns `{ eventId }`. + +### Environment variables + +```bash .env +UPSTASH_REDIS_REST_URL=... +UPSTASH_REDIS_REST_TOKEN=... +QSTASH_TOKEN=... +# Required: the delivery endpoint refuses to run without them. +QSTASH_CURRENT_SIGNING_KEY=... +QSTASH_NEXT_SIGNING_KEY=... +# Encrypts the hosts' signing secrets at rest. Generate with: openssl rand -base64 32 +MCP_EVENTS_SECRET_KEY=... +# Where QStash delivers events. Must be publicly reachable. +APP_URL=https://your-app.vercel.app +``` + +## What happens on the wire + +1. The host calls `events/subscribe` with the event name, arguments, its callback URL and a + `whsec_` signing secret. +2. For a new callback or secret, your server POSTs a signed verification challenge and requires the + host to echo it back, then stores the subscription and answers with its `id` and a `refreshBefore` time. +3. When you `emit`, every matching subscription that `authorize` still allows gets a POST with the + event envelope and [Standard Webhooks](https://www.standardwebhooks.com/) headers: + +```json +{ + "eventId": "evt_7c1…", + "name": "comment.created", + "timestamp": "2026-10-07T12:05:00Z", + "data": { "documentId": "doc_123", "text": "Ship it?" }, + "cursor": null +} +``` + +QStash retries failed deliveries with backoff. The event id stays the same across retries, so the +host can drop duplicates, and the signature is computed fresh on every attempt. + +## Matching and authorize + +Every `input` field must also be a `payload` field: the payload is the one place an event's values +come from, and the same values are used to route it and to authorize it. The payload's values for the +input fields are parsed with the `input` schema, so its transforms (`.toLowerCase()`, say) apply on +both sides and `authorize` gets the types it declares; a payload whose values don't fit `input` makes +`emit` throw. A subscription matches when each argument it gave equals the payload's value. Emitting `{ documentId: "doc_123", text }` reaches +subscribers of `{ documentId: "doc_123" }` and of `{}`. + +`authorize(args, caller)` runs twice: + +* **On every subscribe and refresh**, with the subscription's arguments and + `{ principal, phase: "subscribe", auth, request }`, before the callback is challenged or anything + is stored. A refusal is an error to the host. +* **Before every delivery**, with the _event's_ values for the input fields and + `{ principal, phase: "deliver" }`. So a subscriber who filtered on nothing is still checked against + each event's `documentId`, and revoked access stops the events. A refusal drops that delivery; a + throw (your database is down, say) makes QStash retry it. + +The token is never stored, so the delivery-time check gets no `auth` or `request`. +`authorize: () => true` lets every authenticated subscriber hear every matching event. + + + Pass `{ eventId }` as the second argument to `emit`. It is sent as `webhook-id`, so the host drops a + second emit with the same id as a duplicate: + + ```ts + await commentCreated.emit({ documentId, text }, { eventId: comment.id }); + ``` + + + + **The host routes to its user.** Every subscription carries a callback URL and signing secret that + the host generated for it. ChatGPT, for example, sends a unique + `https://connectors.api.openai.com/webhook/mcp-events/` per monitor. Posting to that URL + reaches the right user's agent, so your server never needs to know who the host user is. Two users + subscribing to the same event produce two subscriptions and two deliveries. + + **Your server knows who subscribed.** `events/subscribe` arrives with the same auth as any other + MCP request, and `principal` turns it into the subscriber's id. `events/subscribe` and + `events/unsubscribe` resolve `principal` before anything else, so an unidentified caller learns + nothing about your events and is refused with reason `not_authenticated`. A subscription's id is a + hash of subscriber, callback URL, event and arguments, so a user can only refresh or remove their + own. A server with no users of its own passes `principal: () => "local"`. + + Every recipient of an emit gets the same payload, so don't put data in it that only some matching + subscribers may see. + + +## Security + + + * **The callback URL.** It must be `https` on a public host name. Refused: every IP literal, + `localhost`, single-label, `.local` and `.internal` names, and credentials in the URL. Redirects + are never followed. Before a subscription is stored, the server POSTs a signed challenge and + requires the host to echo it back. Every failure returns the same `-32015` error, so a subscriber + can't use it to probe your network (the details go to your logs). DNS is not resolved, so use + egress filtering in production. + * **The signing secret.** It must be `whsec_` followed by 24 to 64 base64 bytes, and it is stored + encrypted with AES-256-GCM under `MCP_EVENTS_SECRET_KEY`, which must be base64 of at least 32 + random bytes. A refresh with the same secret skips the challenge. If you rotate the key, stored + subscriptions stop receiving events until the host refreshes them. + * **The delivery endpoint.** It verifies QStash's signature with the `QSTASH_*_SIGNING_KEY` + variables (or a `receiver` you pass). Without them it throws on the first request instead of + sending unverified deliveries. A bad signature or body gets `489` with + `Upstash-NonRetryable-Error: true`. + * **Lifetime.** A subscription lasts 7 days by default and 30 days at most. + * **How many.** A subscriber holds at most 8 live subscriptions across all events. Past that, a new + subscribe is refused with reason `subscription_limit`, before the callback is challenged; + refreshing an existing one still works. Each subscription is a webhook per matching emit, so this + bounds what one user can make your server send. Set `maxSubscriptions` to change it, or + `Infinity` to turn it off. + + + + `allowInsecureCallbacks: true` turns the callback URL checks off. Use it only for local + development, never in production. + + + + | Callback answers | What happens | + | --- | --- | + | `2xx` | Delivered. | + | `410 Gone` | The subscription is deleted. | + | `413`, or a redirect | The event is dropped, not retried. | + | Anything else, or no answer | QStash retries with backoff. | + + + + **`mcp-events:sub:`** is the subscription (`event`, `args`, `url`, `encryptedSecret`, + `subscriber`, `createdAt`, `expiresAt`), expiring with it. **`mcp-events:idx:`** and + **`mcp-events:by:`** are sorted sets of the event's and the subscriber's subscription + ids, scored by expiry. The second one enforces the per-subscriber limit. + + An event message carries the full payload, which stays in QStash (and in its DLQ, if every retry + fails) until it is delivered. The toolkit never stores the caller's token, the request, the + plaintext webhook secret or `MCP_EVENTS_SECRET_KEY`. Anyone with write access to your Redis can + change a callback URL, so treat the Redis credentials like any other production secret. + + +## Options + + + **`createEventLayer`**: `store`, `delivery` and `principal` are required, and so is `secretKey` + unless `MCP_EVENTS_SECRET_KEY` is set. Optional: `maxSubscriptions` (8 per subscriber), + `allowInsecureCallbacks`. + + **`events.define(name, config)`**: `description`, `payload` and `authorize` are required. + Optional: `title`, `input`. + + **`RedisSubscriptionStore`**: `redis` (defaults to one from env), `prefix` (`mcp-events:`), + `enableTelemetry`. + + **`QStashDelivery`**: `url` is required. Optional: `qstash`, `receiver`, `enableTelemetry`. + Deliveries are retried 3 times with QStash's backoff. + + + + `@upstash/mcp-toolkit/events` doesn't import anything from Upstash. A backend implements one of + these interfaces (types exported from the same entry point): + + ```ts + interface SubscriptionStore { + // false, storing nothing, when a new one would put its subscriber over `limit` (atomically) + put(subscription: Subscription, options: { limit: number }): Promise; + get(id: string): Promise; + count(subscriber: string): Promise; // live subscriptions, across all events + delete(subscription: { id: string; event: string; subscriber: string }): Promise; + find(event: string): Promise; // every live subscription to the event + } + + interface EventDelivery { + enqueue(jobs: DeliveryJob[]): Promise; + createDeliveryHandler(send: SendJob): (request: Request) => Promise; + } + ``` + + The entry point also exports `verifyWebhook` (Standard Webhooks) for writing a receiver. + + +Not implemented: the draft's poll and stream delivery modes (`events/subscribe` refuses them) and +event replay (`cursor` is always `null`). + +## Example + +The [MCP toolkit demo](https://github.com/upstash/agentkit/tree/main/examples/mcp-toolkit-demo) +includes Deploy Watch, an events-only server with a `deploy.finished` event filtered by +environment, plus a local receiver that verifies each signed delivery. We tested it end to end with +ChatGPT monitors. It needs `MCP_EVENTS_SECRET_KEY` set, and its deploy-report route is +unauthenticated so the demo is easy to drive; a real server must authenticate the code that emits. + +# MCP Events with Redis and QStash +Source: https://upstash.com/docs/redis/sdks/agentkit/mcp-events + +Most agents only act when someone messages them. [MCP Events](https://github.com/modelcontextprotocol/experimental-ext-triggers-events) +lets an agent wait for something instead: the host subscribes to an event on your MCP server, and +when it happens your server POSTs a signed webhook that wakes the agent up. + +`@upstash/mcp-toolkit/events` implements the server side. You define events with typed payloads and +call `emit` when something happens. The toolkit handles subscriptions, callback verification, +signing and retries: subscriptions live in Upstash Redis, and every delivery goes through +[QStash](/docs/qstash/overall/getstarted), which retries failed webhooks with backoff. + + + MCP Events is a draft. As of October 2026, ChatGPT is the only widely used host that subscribes to + it (webhook delivery, in Work chats), following [OpenAI's MCP Events guide](https://developers.openai.com/plugins/build/mcp-events). + Codex supports events only for OpenAI's own connectors. Claude Code, Cursor and OpenCode don't + subscribe yet. + + +```bash +npm install @upstash/mcp-toolkit @modelcontextprotocol/server mcp-handler zod +``` + +## Quickstart + +Four files in a Next.js app. + +### 1. Identify the caller + +Every subscription belongs to the user who subscribed. `verifyToken` turns the request's bearer token +into the SDK's `AuthInfo`, with your user id in `extra`, and `principal` reads that id back for the +toolkit. If you also use [MCP Tasks](/docs/redis/sdks/agentkit/mcp-tasks), both layers share this file: + +```ts lib/auth.ts +import type { AuthInfo } from "@modelcontextprotocol/server"; + +// 1. Runs on every MCP request (wired up with withMcpAuth in step 3). +// 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"); + return userId; +} +``` + +### 2. Define the event layer and your events + +```ts lib/events.ts +import { createEventLayer } from "@upstash/mcp-toolkit/events"; +import { QStashDelivery, RedisSubscriptionStore } from "@upstash/mcp-toolkit/upstash"; +import * as z from "zod"; +import { principal } from "./auth"; + +export const events = createEventLayer({ + store: new RedisSubscriptionStore(), + delivery: new QStashDelivery({ url: `${process.env.APP_URL}/api/events` }), + principal, +}); + +export const commentCreated = events.define("comment.created", { + description: "A new comment was added to a document.", + input: z.object({ documentId: z.string().optional() }), // what a subscriber may filter on + payload: z.object({ documentId: z.string(), text: z.string() }), + // May this user see comments on this document? + authorize: (args, { principal }) => canRead(principal, args.documentId), +}); +``` + +`authorize` is required. See [Matching and authorize](#matching-and-authorize) for when it runs. + +### 3. Serve the MCP endpoint + +```ts app/api/mcp/route.ts +import { createMcpHandler, withMcpAuth } from "mcp-handler"; +import { verifyToken } from "../../../lib/auth"; +import { events } from "../../../lib/events"; + +const handler = createMcpHandler((server) => { + events.register(server); // adds events/list, events/subscribe and events/unsubscribe +}); + +// Runs verifyToken before any request is handled; no AuthInfo means 401. +const authHandler = withMcpAuth(handler, verifyToken, { required: true }); +export { authHandler as GET, authHandler as POST }; +``` + +`withMcpAuth` hands the `AuthInfo` from `verifyToken` to the MCP server, and the toolkit passes it +to `principal` on every subscribe. With [MCP Tasks](/docs/redis/sdks/agentkit/mcp-tasks) too, call +`tasks.register(server)` in the same callback. + +### 4. Add the delivery endpoint + +```ts app/api/events/route.ts +import { events } from "../../../lib/events"; + +export const POST = events.createDeliveryHandler(); +``` + +QStash delivers each event here. The handler verifies the QStash signature, signs the envelope with +the subscriber's secret and POSTs it to the host. + +### Emit + +Call `emit` wherever the change happens: a route, a webhook from your own app, a background job. + +```ts +await commentCreated.emit({ documentId: "doc_123", text: "Ship it?" }); +``` + +`emit` is typed by the `payload` schema and validates against it. Every subscription that matches +the payload, and that `authorize` still allows for `documentId: "doc_123"`, gets a signed webhook. +It returns `{ eventId }`. + +### Environment variables + +```bash .env +UPSTASH_REDIS_REST_URL=... +UPSTASH_REDIS_REST_TOKEN=... +QSTASH_TOKEN=... +# Required: the delivery endpoint refuses to run without them. +QSTASH_CURRENT_SIGNING_KEY=... +QSTASH_NEXT_SIGNING_KEY=... +# Encrypts the hosts' signing secrets at rest. Generate with: openssl rand -base64 32 +MCP_EVENTS_SECRET_KEY=... +# Where QStash delivers events. Must be publicly reachable. +APP_URL=https://your-app.vercel.app +``` + +## What happens on the wire + +1. The host calls `events/subscribe` with the event name, arguments, its callback URL and a + `whsec_` signing secret. +2. For a new callback or secret, your server POSTs a signed verification challenge and requires the + host to echo it back, then stores the subscription and answers with its `id` and a `refreshBefore` time. +3. When you `emit`, every matching subscription that `authorize` still allows gets a POST with the + event envelope and [Standard Webhooks](https://www.standardwebhooks.com/) headers: + +```json +{ + "eventId": "evt_7c1…", + "name": "comment.created", + "timestamp": "2026-10-07T12:05:00Z", + "data": { "documentId": "doc_123", "text": "Ship it?" }, + "cursor": null +} +``` + +QStash retries failed deliveries with backoff. The event id stays the same across retries, so the +host can drop duplicates, and the signature is computed fresh on every attempt. + +## Matching and authorize + +Every `input` field must also be a `payload` field: the payload is the one place an event's values +come from, and the same values are used to route it and to authorize it. The payload's values for the +input fields are parsed with the `input` schema, so its transforms (`.toLowerCase()`, say) apply on +both sides and `authorize` gets the types it declares; a payload whose values don't fit `input` makes +`emit` throw. A subscription matches when each argument it gave equals the payload's value. Emitting `{ documentId: "doc_123", text }` reaches +subscribers of `{ documentId: "doc_123" }` and of `{}`. + +`authorize(args, caller)` runs twice: + +* **On every subscribe and refresh**, with the subscription's arguments and + `{ principal, phase: "subscribe", auth, request }`, before the callback is challenged or anything + is stored. A refusal is an error to the host. +* **Before every delivery**, with the _event's_ values for the input fields and + `{ principal, phase: "deliver" }`. So a subscriber who filtered on nothing is still checked against + each event's `documentId`, and revoked access stops the events. A refusal drops that delivery; a + throw (your database is down, say) makes QStash retry it. + +The token is never stored, so the delivery-time check gets no `auth` or `request`. +`authorize: () => true` lets every authenticated subscriber hear every matching event. + + + Pass `{ eventId }` as the second argument to `emit`. It is sent as `webhook-id`, so the host drops a + second emit with the same id as a duplicate: + + ```ts + await commentCreated.emit({ documentId, text }, { eventId: comment.id }); + ``` + + + + **The host routes to its user.** Every subscription carries a callback URL and signing secret that + the host generated for it. ChatGPT, for example, sends a unique + `https://connectors.api.openai.com/webhook/mcp-events/` per monitor. Posting to that URL + reaches the right user's agent, so your server never needs to know who the host user is. Two users + subscribing to the same event produce two subscriptions and two deliveries. + + **Your server knows who subscribed.** `events/subscribe` arrives with the same auth as any other + MCP request, and `principal` turns it into the subscriber's id. `events/subscribe` and + `events/unsubscribe` resolve `principal` before anything else, so an unidentified caller learns + nothing about your events and is refused with reason `not_authenticated`. A subscription's id is a + hash of subscriber, callback URL, event and arguments, so a user can only refresh or remove their + own. A server with no users of its own passes `principal: () => "local"`. + + Every recipient of an emit gets the same payload, so don't put data in it that only some matching + subscribers may see. + + +## Security + + + * **The callback URL.** It must be `https` on a public host name. Refused: every IP literal, + `localhost`, single-label, `.local` and `.internal` names, and credentials in the URL. Redirects + are never followed. Before a subscription is stored, the server POSTs a signed challenge and + requires the host to echo it back. Every failure returns the same `-32015` error, so a subscriber + can't use it to probe your network (the details go to your logs). DNS is not resolved, so use + egress filtering in production. + * **The signing secret.** It must be `whsec_` followed by 24 to 64 base64 bytes, and it is stored + encrypted with AES-256-GCM under `MCP_EVENTS_SECRET_KEY`, which must be base64 of at least 32 + random bytes. A refresh with the same secret skips the challenge. If you rotate the key, stored + subscriptions stop receiving events until the host refreshes them. + * **The delivery endpoint.** It verifies QStash's signature with the `QSTASH_*_SIGNING_KEY` + variables (or a `receiver` you pass). Without them it throws on the first request instead of + sending unverified deliveries. A bad signature or body gets `489` with + `Upstash-NonRetryable-Error: true`. + * **Lifetime.** A subscription lasts 7 days by default and 30 days at most. + * **How many.** A subscriber holds at most 8 live subscriptions across all events. Past that, a new + subscribe is refused with reason `subscription_limit`, before the callback is challenged; + refreshing an existing one still works. Each subscription is a webhook per matching emit, so this + bounds what one user can make your server send. Set `maxSubscriptions` to change it, or + `Infinity` to turn it off. + + + + `allowInsecureCallbacks: true` turns the callback URL checks off. Use it only for local + development, never in production. + + + + | Callback answers | What happens | + | --- | --- | + | `2xx` | Delivered. | + | `410 Gone` | The subscription is deleted. | + | `413`, or a redirect | The event is dropped, not retried. | + | Anything else, or no answer | QStash retries with backoff. | + + + + **`mcp-events:sub:`** is the subscription (`event`, `args`, `url`, `encryptedSecret`, + `subscriber`, `createdAt`, `expiresAt`), expiring with it. **`mcp-events:idx:`** and + **`mcp-events:by:`** are sorted sets of the event's and the subscriber's subscription + ids, scored by expiry. The second one enforces the per-subscriber limit. + + An event message carries the full payload, which stays in QStash (and in its DLQ, if every retry + fails) until it is delivered. The toolkit never stores the caller's token, the request, the + plaintext webhook secret or `MCP_EVENTS_SECRET_KEY`. Anyone with write access to your Redis can + change a callback URL, so treat the Redis credentials like any other production secret. + + +## Options + + + **`createEventLayer`**: `store`, `delivery` and `principal` are required, and so is `secretKey` + unless `MCP_EVENTS_SECRET_KEY` is set. Optional: `maxSubscriptions` (8 per subscriber), + `allowInsecureCallbacks`. + + **`events.define(name, config)`**: `description`, `payload` and `authorize` are required. + Optional: `title`, `input`. + + **`RedisSubscriptionStore`**: `redis` (defaults to one from env), `prefix` (`mcp-events:`), + `enableTelemetry`. + + **`QStashDelivery`**: `url` is required. Optional: `qstash`, `receiver`, `enableTelemetry`. + Deliveries are retried 3 times with QStash's backoff. + + + + `@upstash/mcp-toolkit/events` doesn't import anything from Upstash. A backend implements one of + these interfaces (types exported from the same entry point): + + ```ts + interface SubscriptionStore { + // false, storing nothing, when a new one would put its subscriber over `limit` (atomically) + put(subscription: Subscription, options: { limit: number }): Promise; + get(id: string): Promise; + count(subscriber: string): Promise; // live subscriptions, across all events + delete(subscription: { id: string; event: string; subscriber: string }): Promise; + find(event: string): Promise; // every live subscription to the event + } + + interface EventDelivery { + enqueue(jobs: DeliveryJob[]): Promise; + createDeliveryHandler(send: SendJob): (request: Request) => Promise; + } + ``` + + The entry point also exports `verifyWebhook` (Standard Webhooks) for writing a receiver. + + +Not implemented: the draft's poll and stream delivery modes (`events/subscribe` refuses them) and +event replay (`cursor` is always `null`). + +## Example + +The [MCP toolkit demo](https://github.com/upstash/agentkit/tree/main/examples/mcp-toolkit-demo) +includes Deploy Watch, an events-only server with a `deploy.finished` event filtered by +environment, plus a local receiver that verifies each signed delivery. We tested it end to end with +ChatGPT monitors. It needs `MCP_EVENTS_SECRET_KEY` set, and its deploy-report route is +unauthenticated so the demo is easy to drive; a real server must authenticate the code that emits. + +# Long-Running MCP Tools with Redis and QStash +Source: https://upstash.com/docs/redis/sdks/agentkit/mcp-tasks + +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 tool answers immediately with a task +id, the work runs on [QStash](/docs/qstash/overall/getstarted) or [Upstash Workflow](/docs/workflow/getstarted) +so it survives restarts and the client's tool timeout, and the task record lives in Upstash Redis. +The model checks on it with a shared `task_status` tool and can stop it with `task_cancel`. + +These are ordinary MCP tools, so they work in every client today: Claude Code, Codex, Cursor, +OpenCode and ChatGPT. No client capability is required. 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 mcp-handler zod +``` + +## Quickstart + +Four files in a Next.js app. + +### 1. Identify the caller + +Every task belongs to the user who started it. Two functions, kept in one file so the path is easy +to follow: + +* `verifyToken` runs on every MCP request. It verifies the bearer token and returns the SDK's + `AuthInfo`, with your user id in `extra`. Step 3 wires it up with `withMcpAuth`. +* `principal` is what the toolkit calls. It gets that `AuthInfo` and returns the user id, or throws. + +```ts lib/auth.ts +import type { AuthInfo } from "@modelcontextprotocol/server"; + +// 1. Runs on every MCP request. 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"); + return userId; +} +``` + +### 2. Define the task layer and the task + +```ts lib/tasks.ts +import { createTaskLayer } from "@upstash/mcp-toolkit/tasks"; +import { QStashDispatcher, RedisTaskStore } from "@upstash/mcp-toolkit/upstash"; +import * as z from "zod"; +import { principal } from "./auth"; + +export const tasks = createTaskLayer({ + store: new RedisTaskStore(), + dispatcher: new QStashDispatcher({ url: `${process.env.APP_URL}/api/execute` }), + principal, +}); + +// Define at module scope, so the /api/execute instance knows the handler too. +tasks.define( + "generate_report", + { description: "Generates a report on a topic.", inputSchema: z.object({ topic: z.string() }) }, + async ({ topic }, task) => { + await task.update("Reading sources"); // the model sees this when it polls + return { content: [{ type: "text", text: await writeReport(topic) }] }; + }, +); +``` + +The handler runs later, in `/api/execute`, when QStash delivers the task, not inside the tool call. +`topic` is typed from `inputSchema`: Zod, Valibot and ArkType schemas all work. + +### 3. Serve the MCP endpoint + +```ts app/api/mcp/route.ts +import { createMcpHandler, withMcpAuth } from "mcp-handler"; +import { verifyToken } from "../../../lib/auth"; +import { tasks } from "../../../lib/tasks"; + +const handler = createMcpHandler((server) => { + tasks.register(server); // adds generate_report, task_status and task_cancel +}); + +// 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 }; +``` + +`withMcpAuth` hands the `AuthInfo` from `verifyToken` to the MCP server, and the toolkit passes it +to `principal`. The task layer only registers tools, so nothing else changes in your route. + +### 4. Add the execute endpoint + +```ts app/api/execute/route.ts +import { tasks } from "../../../lib/tasks"; + +export const POST = tasks.createExecuteHandler(); +``` + +This is where QStash delivers each task. The handler verifies the QStash signature, runs the task's +handler and stores the result. + +### Environment variables + +```bash .env +UPSTASH_REDIS_REST_URL=... +UPSTASH_REDIS_REST_TOKEN=... +QSTASH_TOKEN=... +# Required: the execute endpoint refuses to run without them. +QSTASH_CURRENT_SIGNING_KEY=... +QSTASH_NEXT_SIGNING_KEY=... +# Where QStash delivers tasks. Must be publicly reachable. +APP_URL=https://your-app.vercel.app +``` + + + For local development, run the [QStash dev server](/docs/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 use `APP_URL=http://127.0.0.1:3000`. + + +## What the model sees + +`tools/list` now has `generate_report` (starts the task and returns its id), `task_status` (shows +progress, then the handler's result) and `task_cancel`. + + + ```jsonc + // generate_report: a handle, right away + { "content": [{ "type": "text", "text": "Started task 0e30…. Call task_status with taskId \"0e30…\" in about 2s to check on it." }], + "structuredContent": { "taskId": "0e30…", "status": "working", "ttlMs": 86400000, "pollIntervalMs": 2000 } } + + // task_status: progress… + { "content": [{ "type": "text", "text": "Task 0e30… is working: Reading sources. Check again in about 2s." }], + "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: Completed" }, + { "type": "text", "text": "Report on coffee" }], + "structuredContent": { "taskId": "0e30…", "status": "completed", "result": { "content": [ … ] } } } + ``` + + The states are `working`, `completed`, `failed` and `cancelled`; the last three are final. The + task object has the same shape as the one in the MCP Tasks extension. + + 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 the layer). + + +## Cancellation + +`task_cancel` marks the task `cancelled` and stops any pending delivery. Cancelling is cooperative, +so running code only stops where it checks `task.isCancelled()`: + +```ts +async ({ sources }, task) => { + for (const source of sources) { + if (await task.isCancelled()) return {}; + await task.update(`Reading ${source}`); + await read(source); + } + return { content: [{ type: "text", text: "Done" }] }; +}; +``` + +`isCancelled()` also returns `true` once the task has expired. + + + 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. + + +## Users + +Each task records who started it, and `task_status` / `task_cancel` only answer for that caller. +Another user's task id reads exactly like an unknown one, so nobody can tell it exists. + +### 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. If Alice starts a task with Bob's +`workspaceId`, the task is hers, she can read its result, and the handler acted on Bob's workspace. + +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 lib/tasks.ts +tasks.define( + "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 later, in a delivery request that carries no user auth, so `task.principal` is how it +knows who it is working for. + +### Where principal comes from + +`principal` receives `{ auth, request }` and may be async. `auth` is exactly what `verifyToken` +returned in [step 1](#1-identify-the-caller): `withMcpAuth` runs it, and the toolkit hands the result +to `principal`. Nothing is read from request headers on its own. Without `mcp-handler`, pass the +`AuthInfo` yourself with the SDK's `createMcpHandler`: `handler.fetch(request, { authInfo })`. + + + * **A throw refuses the call.** There is no anonymous mode: if `principal` throws, rejects, or + returns anything but a non-empty string, the tool 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. + * `principal` also receives `request`, for cookie or session apps. It is unverified, so check the + session yourself, and never trust a header like `x-user-id`: + + ```ts + import type { Caller } from "@upstash/mcp-toolkit/tasks"; + + export async function principal({ request }: Caller): Promise { + const session = await getSession(request); + if (!session) throw new Error("Not authenticated"); + return session.userId; + } + ``` + + * A server with no users of its own passes `principal: () => "local"`. + + +## 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 routes stay the same. The +handler's `task` then also has `run`, `sleep` and `call`, inferred from the dispatcher: + +```ts lib/tasks.ts +import { RedisTaskStore, WorkflowDispatcher } from "@upstash/mcp-toolkit/upstash"; + +export const tasks = createTaskLayer({ + store: new RedisTaskStore(), + dispatcher: new WorkflowDispatcher({ url: `${process.env.APP_URL}/api/execute` }), + principal, +}); + +tasks.define( + "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" }] }; + }, +); +``` + + + 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. + + | | `QStashDispatcher` | `WorkflowDispatcher` | + | --- | --- | --- | + | Survives the process dying | yes (QStash redelivers) | yes (replayed from the journal) | + | Can run longer than one invocation | no | yes, one invocation per step | + | On a retry | the whole task starts over | only the failed step runs again | + | Cancel stops a running task | at its next `isCancelled` | the run itself is cancelled | + + +## Retries and failures + + + * The execute route 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: true` 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. + This holds for `WorkflowDispatcher` too. Without signing keys (or a `receiver` you pass), the + route throws on its first request 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. + * 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. + + +## Security and storage + + + 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` and + `task_cancel` only accept UUID task ids, and compare the stored owner with the caller. Task ids are + random UUIDs. No tool lists tasks, and the execute route 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. + + + + **`mcp:task:`** is a hash with one JSON-encoded field per property: `taskId`, `name`, + `args`, `owner`, `status`, `statusMessage`, `result` / `error`, `createdAt`, `lastUpdatedAt`, + `ttlMs`, `pollIntervalMs`. + + * `args` and `result` are **plain JSON**. Keep secrets out of tool arguments and results, 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. + + A task message in QStash carries only `{ taskId }`. The toolkit never stores the caller's token or + the request. + + +## Options + + + **`createTaskLayer`**: `store`, `dispatcher` and `principal` are required. Optional: + `defaults.ttlMs` (1 day) and `defaults.pollIntervalMs` (2 seconds), in positive whole milliseconds. + + **`tasks.define(name, config, handler)`**: `description` and `inputSchema` are required. Optional: + `title`, `completedMessage` (defaults to `"Completed"`), `authorize(args, { principal, auth, request })`. + + **`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@` added to their `Upstash-Telemetry-Sdk` header; turn it + off with `enableTelemetry: false` or `UPSTASH_DISABLE_TELEMETRY`. + + + + `@upstash/mcp-toolkit/tasks` doesn't import anything from Upstash. A backend implements one of these + interfaces (types exported from the same entry point): + + ```ts + interface TaskStore { + create(task: Task): Promise; // with its TTL + get(taskId: string): Promise; + update(taskId: string, patch: TaskPatch): Promise; // ignored once the task is final + settle(taskId: string, patch: TaskPatch & { status: TerminalTaskStatus }): Promise; // first final status wins + } + + interface TaskDispatcher { + dispatch(task: Task): Promise; // called once per task + cancel(taskId: string): Promise; + createExecuteHandler(endpoints: TaskEndpoints): (request: Request) => Promise; + } + ``` + + +## Why tools, not the Tasks extension? + +The 2026-07-28 MCP specification defines a [Tasks extension](https://modelcontextprotocol.io/extensions/tasks/overview) +where the *client* polls, so the model spends no turns waiting. But a server may only return a task +to a client that declared the extension, and as of October 2026 Claude Code, Codex, Cursor and +OpenCode don't. Plain tools cost the model a few polling calls but work everywhere today. + +The store and dispatcher don't depend on the tools, so an adapter for the extension can serve the +same records once clients support it. Also not implemented yet: `input_required` (a handler asking +the user something partway through a task) and listing tasks. + +## 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. + +# Long-Running MCP Tools with Redis and QStash +Source: https://upstash.com/docs/redis/sdks/agentkit/mcp-tasks + +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 tool answers immediately with a task +id, the work runs on [QStash](/docs/qstash/overall/getstarted) or [Upstash Workflow](/docs/workflow/getstarted) +so it survives restarts and the client's tool timeout, and the task record lives in Upstash Redis. +The model checks on it with a shared `task_status` tool and can stop it with `task_cancel`. + +These are ordinary MCP tools, so they work in every client today: Claude Code, Codex, Cursor, +OpenCode and ChatGPT. No client capability is required. 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 mcp-handler zod +``` + +## Quickstart + +Four files in a Next.js app. + +### 1. Identify the caller + +Every task belongs to the user who started it. Two functions, kept in one file so the path is easy +to follow: + +* `verifyToken` runs on every MCP request. It verifies the bearer token and returns the SDK's + `AuthInfo`, with your user id in `extra`. Step 3 wires it up with `withMcpAuth`. +* `principal` is what the toolkit calls. It gets that `AuthInfo` and returns the user id, or throws. + +```ts lib/auth.ts +import type { AuthInfo } from "@modelcontextprotocol/server"; + +// 1. Runs on every MCP request. 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"); + return userId; +} +``` + +### 2. Define the task layer and the task + +```ts lib/tasks.ts +import { createTaskLayer } from "@upstash/mcp-toolkit/tasks"; +import { QStashDispatcher, RedisTaskStore } from "@upstash/mcp-toolkit/upstash"; +import * as z from "zod"; +import { principal } from "./auth"; + +export const tasks = createTaskLayer({ + store: new RedisTaskStore(), + dispatcher: new QStashDispatcher({ url: `${process.env.APP_URL}/api/execute` }), + principal, +}); + +// Define at module scope, so the /api/execute instance knows the handler too. +tasks.define( + "generate_report", + { description: "Generates a report on a topic.", inputSchema: z.object({ topic: z.string() }) }, + async ({ topic }, task) => { + await task.update("Reading sources"); // the model sees this when it polls + return { content: [{ type: "text", text: await writeReport(topic) }] }; + }, +); +``` + +The handler runs later, in `/api/execute`, when QStash delivers the task, not inside the tool call. +`topic` is typed from `inputSchema`: Zod, Valibot and ArkType schemas all work. + +### 3. Serve the MCP endpoint + +```ts app/api/mcp/route.ts +import { createMcpHandler, withMcpAuth } from "mcp-handler"; +import { verifyToken } from "../../../lib/auth"; +import { tasks } from "../../../lib/tasks"; + +const handler = createMcpHandler((server) => { + tasks.register(server); // adds generate_report, task_status and task_cancel +}); + +// 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 }; +``` + +`withMcpAuth` hands the `AuthInfo` from `verifyToken` to the MCP server, and the toolkit passes it +to `principal`. The task layer only registers tools, so nothing else changes in your route. + +### 4. Add the execute endpoint + +```ts app/api/execute/route.ts +import { tasks } from "../../../lib/tasks"; + +export const POST = tasks.createExecuteHandler(); +``` + +This is where QStash delivers each task. The handler verifies the QStash signature, runs the task's +handler and stores the result. + +### Environment variables + +```bash .env +UPSTASH_REDIS_REST_URL=... +UPSTASH_REDIS_REST_TOKEN=... +QSTASH_TOKEN=... +# Required: the execute endpoint refuses to run without them. +QSTASH_CURRENT_SIGNING_KEY=... +QSTASH_NEXT_SIGNING_KEY=... +# Where QStash delivers tasks. Must be publicly reachable. +APP_URL=https://your-app.vercel.app +``` + + + For local development, run the [QStash dev server](/docs/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 use `APP_URL=http://127.0.0.1:3000`. + + +## What the model sees + +`tools/list` now has `generate_report` (starts the task and returns its id), `task_status` (shows +progress, then the handler's result) and `task_cancel`. + + + ```jsonc + // generate_report: a handle, right away + { "content": [{ "type": "text", "text": "Started task 0e30…. Call task_status with taskId \"0e30…\" in about 2s to check on it." }], + "structuredContent": { "taskId": "0e30…", "status": "working", "ttlMs": 86400000, "pollIntervalMs": 2000 } } + + // task_status: progress… + { "content": [{ "type": "text", "text": "Task 0e30… is working: Reading sources. Check again in about 2s." }], + "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: Completed" }, + { "type": "text", "text": "Report on coffee" }], + "structuredContent": { "taskId": "0e30…", "status": "completed", "result": { "content": [ … ] } } } + ``` + + The states are `working`, `completed`, `failed` and `cancelled`; the last three are final. The + task object has the same shape as the one in the MCP Tasks extension. + + 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 the layer). + + +## Cancellation + +`task_cancel` marks the task `cancelled` and stops any pending delivery. Cancelling is cooperative, +so running code only stops where it checks `task.isCancelled()`: + +```ts +async ({ sources }, task) => { + for (const source of sources) { + if (await task.isCancelled()) return {}; + await task.update(`Reading ${source}`); + await read(source); + } + return { content: [{ type: "text", text: "Done" }] }; +}; +``` + +`isCancelled()` also returns `true` once the task has expired. + + + 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. + + +## Users + +Each task records who started it, and `task_status` / `task_cancel` only answer for that caller. +Another user's task id reads exactly like an unknown one, so nobody can tell it exists. + +### 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. If Alice starts a task with Bob's +`workspaceId`, the task is hers, she can read its result, and the handler acted on Bob's workspace. + +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 lib/tasks.ts +tasks.define( + "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 later, in a delivery request that carries no user auth, so `task.principal` is how it +knows who it is working for. + +### Where principal comes from + +`principal` receives `{ auth, request }` and may be async. `auth` is exactly what `verifyToken` +returned in [step 1](#1-identify-the-caller): `withMcpAuth` runs it, and the toolkit hands the result +to `principal`. Nothing is read from request headers on its own. Without `mcp-handler`, pass the +`AuthInfo` yourself with the SDK's `createMcpHandler`: `handler.fetch(request, { authInfo })`. + + + * **A throw refuses the call.** There is no anonymous mode: if `principal` throws, rejects, or + returns anything but a non-empty string, the tool 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. + * `principal` also receives `request`, for cookie or session apps. It is unverified, so check the + session yourself, and never trust a header like `x-user-id`: + + ```ts + import type { Caller } from "@upstash/mcp-toolkit/tasks"; + + export async function principal({ request }: Caller): Promise { + const session = await getSession(request); + if (!session) throw new Error("Not authenticated"); + return session.userId; + } + ``` + + * A server with no users of its own passes `principal: () => "local"`. + + +## 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 routes stay the same. The +handler's `task` then also has `run`, `sleep` and `call`, inferred from the dispatcher: + +```ts lib/tasks.ts +import { RedisTaskStore, WorkflowDispatcher } from "@upstash/mcp-toolkit/upstash"; + +export const tasks = createTaskLayer({ + store: new RedisTaskStore(), + dispatcher: new WorkflowDispatcher({ url: `${process.env.APP_URL}/api/execute` }), + principal, +}); + +tasks.define( + "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" }] }; + }, +); +``` + + + 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. + + | | `QStashDispatcher` | `WorkflowDispatcher` | + | --- | --- | --- | + | Survives the process dying | yes (QStash redelivers) | yes (replayed from the journal) | + | Can run longer than one invocation | no | yes, one invocation per step | + | On a retry | the whole task starts over | only the failed step runs again | + | Cancel stops a running task | at its next `isCancelled` | the run itself is cancelled | + + +## Retries and failures + + + * The execute route 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: true` 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. + This holds for `WorkflowDispatcher` too. Without signing keys (or a `receiver` you pass), the + route throws on its first request 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. + * 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. + + +## Security and storage + + + 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` and + `task_cancel` only accept UUID task ids, and compare the stored owner with the caller. Task ids are + random UUIDs. No tool lists tasks, and the execute route 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. + + + + **`mcp:task:`** is a hash with one JSON-encoded field per property: `taskId`, `name`, + `args`, `owner`, `status`, `statusMessage`, `result` / `error`, `createdAt`, `lastUpdatedAt`, + `ttlMs`, `pollIntervalMs`. + + * `args` and `result` are **plain JSON**. Keep secrets out of tool arguments and results, 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. + + A task message in QStash carries only `{ taskId }`. The toolkit never stores the caller's token or + the request. + + +## Options + + + **`createTaskLayer`**: `store`, `dispatcher` and `principal` are required. Optional: + `defaults.ttlMs` (1 day) and `defaults.pollIntervalMs` (2 seconds), in positive whole milliseconds. + + **`tasks.define(name, config, handler)`**: `description` and `inputSchema` are required. Optional: + `title`, `completedMessage` (defaults to `"Completed"`), `authorize(args, { principal, auth, request })`. + + **`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@` added to their `Upstash-Telemetry-Sdk` header; turn it + off with `enableTelemetry: false` or `UPSTASH_DISABLE_TELEMETRY`. + + + + `@upstash/mcp-toolkit/tasks` doesn't import anything from Upstash. A backend implements one of these + interfaces (types exported from the same entry point): + + ```ts + interface TaskStore { + create(task: Task): Promise; // with its TTL + get(taskId: string): Promise; + update(taskId: string, patch: TaskPatch): Promise; // ignored once the task is final + settle(taskId: string, patch: TaskPatch & { status: TerminalTaskStatus }): Promise; // first final status wins + } + + interface TaskDispatcher { + dispatch(task: Task): Promise; // called once per task + cancel(taskId: string): Promise; + createExecuteHandler(endpoints: TaskEndpoints): (request: Request) => Promise; + } + ``` + + +## Why tools, not the Tasks extension? + +The 2026-07-28 MCP specification defines a [Tasks extension](https://modelcontextprotocol.io/extensions/tasks/overview) +where the *client* polls, so the model spends no turns waiting. But a server may only return a task +to a client that declared the extension, and as of October 2026 Claude Code, Codex, Cursor and +OpenCode don't. Plain tools cost the model a few polling calls but work everywhere today. + +The store and dispatcher don't depend on the tools, so an adapter for the extension can serve the +same records once clients support it. Also not implemented yet: `input_required` (a handler asking +the user something partway through a task) and listing tasks. + +## 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. + # TanStack AI Persistence, Resumable Streams & Memory with Redis Source: https://upstash.com/docs/redis/sdks/agentkit/tanstack-ai diff --git a/llms.txt b/llms.txt index 445da8970..32a8c92d2 100644 --- a/llms.txt +++ b/llms.txt @@ -314,6 +314,8 @@ Search these docs: GET https://upstash.com/docs/search?q= returns JSON { - [Schedules](https://upstash.com/docs/qstash/sdks/ts/examples/schedules.md) - [URL Groups](https://upstash.com/docs/qstash/sdks/ts/examples/url-groups.md) - [Getting Started](https://upstash.com/docs/qstash/sdks/ts/gettingstarted.md) +- [MCP Events with QStash](https://upstash.com/docs/qstash/sdks/ts/mcp-events.md): Deliver MCP Events webhooks durably through QStash with @upstash/mcp-toolkit. +- [Long-Running MCP Tools with QStash](https://upstash.com/docs/qstash/sdks/ts/mcp-tasks.md): Run long MCP tool calls durably on QStash with @upstash/mcp-toolkit. - [Overview](https://upstash.com/docs/qstash/sdks/ts/overview.md) - [Channels](https://upstash.com/docs/realtime/features/channels.md) - [Client-Side Usage](https://upstash.com/docs/realtime/features/client-side.md) @@ -741,6 +743,10 @@ Search these docs: GET https://upstash.com/docs/search?q= returns JSON { - [Vercel AI SDK Memory, RAG & Chat History with Redis](https://upstash.com/docs/redis/sdks/agentkit/ai-sdk.md): Add long-term memory, RAG, and chat history to the Vercel AI SDK with Upstash Redis — drop-in tools for generateText and streamText, no separate vector database. - [Memory, Chat History, RAG, Rate Limiting & Sandboxes for the Vercel Eve Agent Framework](https://upstash.com/docs/redis/sdks/agentkit/eve.md): Add long-term memory, searchable chat history, RAG, rate limiting, tool caching, and sandboxes to Vercel's Eve agent framework with Upstash Redis — no separate vector database. - [Memory, Chat History, RAG, Rate Limiting & Sandboxes for the Vercel Eve Agent Framework](https://upstash.com/docs/redis/sdks/agentkit/eve.md): Add long-term memory, searchable chat history, RAG, rate limiting, tool caching, and sandboxes to Vercel's Eve agent framework with Upstash Redis — no separate vector database. +- [MCP Events with Redis and QStash](https://upstash.com/docs/redis/sdks/agentkit/mcp-events.md): Let agents subscribe to events on your MCP server and get a signed webhook when one happens. Subscriptions live in Upstash Redis and deliveries are retried by QStash. +- [MCP Events with Redis and QStash](https://upstash.com/docs/redis/sdks/agentkit/mcp-events.md): Let agents subscribe to events on your MCP server and get a signed webhook when one happens. Subscriptions live in Upstash Redis and deliveries are retried by QStash. +- [Long-Running MCP Tools with Redis and QStash](https://upstash.com/docs/redis/sdks/agentkit/mcp-tasks.md): Durable long-running tools for MCP servers: a tool answers with a task id at once, the work runs on QStash or Upstash Workflow, and the model polls for the result. Works in every MCP client. +- [Long-Running MCP Tools with Redis and QStash](https://upstash.com/docs/redis/sdks/agentkit/mcp-tasks.md): Durable long-running tools for MCP servers: a tool answers with a task id at once, the work runs on QStash or Upstash Workflow, and the model polls for the result. Works in every MCP client. - [TanStack AI Persistence, Resumable Streams & Memory with Redis](https://upstash.com/docs/redis/sdks/agentkit/tanstack-ai.md): Production backends for TanStack AI on Upstash Redis — chat persistence, resumable streams, distributed locks, long-term memory, tool caching, rate limiting, and RAG tools. - [TanStack AI Persistence, Resumable Streams & Memory with Redis](https://upstash.com/docs/redis/sdks/agentkit/tanstack-ai.md): Production backends for TanStack AI on Upstash Redis — chat persistence, resumable streams, distributed locks, long-term memory, tool caching, rate limiting, and RAG tools. - [Upstash Redis MCP](https://upstash.com/docs/redis/sdks/mcp.md) diff --git a/qstash/sdks/ts/mcp-events.mdx b/qstash/sdks/ts/mcp-events.mdx new file mode 100644 index 000000000..16d512510 --- /dev/null +++ b/qstash/sdks/ts/mcp-events.mdx @@ -0,0 +1,9 @@ +--- +title: "MCP Events with QStash" +sidebarTitle: "MCP Events" +description: "Deliver MCP Events webhooks durably through QStash with @upstash/mcp-toolkit." +--- + + + MCP Events for your server: subscriptions in Upstash Redis, signed webhook deliveries retried by QStash. + diff --git a/qstash/sdks/ts/mcp-tasks.mdx b/qstash/sdks/ts/mcp-tasks.mdx new file mode 100644 index 000000000..4c73baea1 --- /dev/null +++ b/qstash/sdks/ts/mcp-tasks.mdx @@ -0,0 +1,9 @@ +--- +title: "Long-Running MCP Tools with QStash" +sidebarTitle: "MCP Tasks" +description: "Run long MCP tool calls durably on QStash with @upstash/mcp-toolkit." +--- + + + Durable long-running MCP tools on QStash or Upstash Workflow, with the task record in Upstash Redis. + diff --git a/redis/sdks/agentkit/mcp-events.mdx b/redis/sdks/agentkit/mcp-events.mdx new file mode 100644 index 000000000..080c2b8ad --- /dev/null +++ b/redis/sdks/agentkit/mcp-events.mdx @@ -0,0 +1,313 @@ +--- +title: "MCP Events with Redis and QStash" +sidebarTitle: "MCP Events" +description: "Let agents subscribe to events on your MCP server and get a signed webhook when one happens. Subscriptions live in Upstash Redis and deliveries are retried by QStash." +--- + +Most agents only act when someone messages them. [MCP Events](https://github.com/modelcontextprotocol/experimental-ext-triggers-events) +lets an agent wait for something instead: the host subscribes to an event on your MCP server, and +when it happens your server POSTs a signed webhook that wakes the agent up. + +`@upstash/mcp-toolkit/events` implements the server side. You define events with typed payloads and +call `emit` when something happens. The toolkit handles subscriptions, callback verification, +signing and retries: subscriptions live in Upstash Redis, and every delivery goes through +[QStash](/qstash/overall/getstarted), which retries failed webhooks with backoff. + + + MCP Events is a draft. As of October 2026, ChatGPT is the only widely used host that subscribes to + it (webhook delivery, in Work chats), following [OpenAI's MCP Events guide](https://developers.openai.com/plugins/build/mcp-events). + Codex supports events only for OpenAI's own connectors. Claude Code, Cursor and OpenCode don't + subscribe yet. + + +```bash +npm install @upstash/mcp-toolkit @modelcontextprotocol/server mcp-handler zod +``` + +## Quickstart + +Four files in a Next.js app. + +### 1. Identify the caller + +Every subscription belongs to the user who subscribed. `verifyToken` turns the request's bearer token +into the SDK's `AuthInfo`, with your user id in `extra`, and `principal` reads that id back for the +toolkit. If you also use [MCP Tasks](/redis/sdks/agentkit/mcp-tasks), both layers share this file: + +```ts lib/auth.ts +import type { AuthInfo } from "@modelcontextprotocol/server"; + +// 1. Runs on every MCP request (wired up with withMcpAuth in step 3). +// 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"); + return userId; +} +``` + +### 2. Define the event layer and your events + +```ts lib/events.ts +import { createEventLayer } from "@upstash/mcp-toolkit/events"; +import { QStashDelivery, RedisSubscriptionStore } from "@upstash/mcp-toolkit/upstash"; +import * as z from "zod"; +import { principal } from "./auth"; + +export const events = createEventLayer({ + store: new RedisSubscriptionStore(), + delivery: new QStashDelivery({ url: `${process.env.APP_URL}/api/events` }), + principal, +}); + +export const commentCreated = events.define("comment.created", { + description: "A new comment was added to a document.", + input: z.object({ documentId: z.string().optional() }), // what a subscriber may filter on + payload: z.object({ documentId: z.string(), text: z.string() }), + // May this user see comments on this document? + authorize: (args, { principal }) => canRead(principal, args.documentId), +}); +``` + +`authorize` is required. See [Matching and authorize](#matching-and-authorize) for when it runs. + +### 3. Serve the MCP endpoint + +```ts app/api/mcp/route.ts +import { createMcpHandler, withMcpAuth } from "mcp-handler"; +import { verifyToken } from "../../../lib/auth"; +import { events } from "../../../lib/events"; + +const handler = createMcpHandler((server) => { + events.register(server); // adds events/list, events/subscribe and events/unsubscribe +}); + +// Runs verifyToken before any request is handled; no AuthInfo means 401. +const authHandler = withMcpAuth(handler, verifyToken, { required: true }); +export { authHandler as GET, authHandler as POST }; +``` + +`withMcpAuth` hands the `AuthInfo` from `verifyToken` to the MCP server, and the toolkit passes it +to `principal` on every subscribe. With [MCP Tasks](/redis/sdks/agentkit/mcp-tasks) too, call +`tasks.register(server)` in the same callback. + +### 4. Add the delivery endpoint + +```ts app/api/events/route.ts +import { events } from "../../../lib/events"; + +export const POST = events.createDeliveryHandler(); +``` + +QStash delivers each event here. The handler verifies the QStash signature, signs the envelope with +the subscriber's secret and POSTs it to the host. + +### Emit + +Call `emit` wherever the change happens: a route, a webhook from your own app, a background job. + +```ts +await commentCreated.emit({ documentId: "doc_123", text: "Ship it?" }); +``` + +`emit` is typed by the `payload` schema and validates against it. Every subscription that matches +the payload, and that `authorize` still allows for `documentId: "doc_123"`, gets a signed webhook. +It returns `{ eventId }`. + +### Environment variables + +```bash .env +UPSTASH_REDIS_REST_URL=... +UPSTASH_REDIS_REST_TOKEN=... +QSTASH_TOKEN=... +# Required: the delivery endpoint refuses to run without them. +QSTASH_CURRENT_SIGNING_KEY=... +QSTASH_NEXT_SIGNING_KEY=... +# Encrypts the hosts' signing secrets at rest. Generate with: openssl rand -base64 32 +MCP_EVENTS_SECRET_KEY=... +# Where QStash delivers events. Must be publicly reachable. +APP_URL=https://your-app.vercel.app +``` + +## What happens on the wire + +1. The host calls `events/subscribe` with the event name, arguments, its callback URL and a + `whsec_` signing secret. +2. For a new callback or secret, your server POSTs a signed verification challenge and requires the + host to echo it back, then stores the subscription and answers with its `id` and a `refreshBefore` time. +3. When you `emit`, every matching subscription that `authorize` still allows gets a POST with the + event envelope and [Standard Webhooks](https://www.standardwebhooks.com/) headers: + +```json +{ + "eventId": "evt_7c1…", + "name": "comment.created", + "timestamp": "2026-10-07T12:05:00Z", + "data": { "documentId": "doc_123", "text": "Ship it?" }, + "cursor": null +} +``` + +QStash retries failed deliveries with backoff. The event id stays the same across retries, so the +host can drop duplicates, and the signature is computed fresh on every attempt. + +## Matching and authorize + +Every `input` field must also be a `payload` field: the payload is the one place an event's values +come from, and the same values are used to route it and to authorize it. The payload's values for the +input fields are parsed with the `input` schema, so its transforms (`.toLowerCase()`, say) apply on +both sides and `authorize` gets the types it declares; a payload whose values don't fit `input` makes +`emit` throw. A subscription matches when each argument it gave equals the payload's value. Emitting `{ documentId: "doc_123", text }` reaches +subscribers of `{ documentId: "doc_123" }` and of `{}`. + +`authorize(args, caller)` runs twice: + +- **On every subscribe and refresh**, with the subscription's arguments and + `{ principal, phase: "subscribe", auth, request }`, before the callback is challenged or anything + is stored. A refusal is an error to the host. +- **Before every delivery**, with the _event's_ values for the input fields and + `{ principal, phase: "deliver" }`. So a subscriber who filtered on nothing is still checked against + each event's `documentId`, and revoked access stops the events. A refusal drops that delivery; a + throw (your database is down, say) makes QStash retry it. + +The token is never stored, so the delivery-time check gets no `auth` or `request`. +`authorize: () => true` lets every authenticated subscriber hear every matching event. + + + Pass `{ eventId }` as the second argument to `emit`. It is sent as `webhook-id`, so the host drops a + second emit with the same id as a duplicate: + + ```ts + await commentCreated.emit({ documentId, text }, { eventId: comment.id }); + ``` + + + + **The host routes to its user.** Every subscription carries a callback URL and signing secret that + the host generated for it. ChatGPT, for example, sends a unique + `https://connectors.api.openai.com/webhook/mcp-events/` per monitor. Posting to that URL + reaches the right user's agent, so your server never needs to know who the host user is. Two users + subscribing to the same event produce two subscriptions and two deliveries. + + **Your server knows who subscribed.** `events/subscribe` arrives with the same auth as any other + MCP request, and `principal` turns it into the subscriber's id. `events/subscribe` and + `events/unsubscribe` resolve `principal` before anything else, so an unidentified caller learns + nothing about your events and is refused with reason `not_authenticated`. A subscription's id is a + hash of subscriber, callback URL, event and arguments, so a user can only refresh or remove their + own. A server with no users of its own passes `principal: () => "local"`. + + Every recipient of an emit gets the same payload, so don't put data in it that only some matching + subscribers may see. + + +## Security + + + - **The callback URL.** It must be `https` on a public host name. Refused: every IP literal, + `localhost`, single-label, `.local` and `.internal` names, and credentials in the URL. Redirects + are never followed. Before a subscription is stored, the server POSTs a signed challenge and + requires the host to echo it back. Every failure returns the same `-32015` error, so a subscriber + can't use it to probe your network (the details go to your logs). DNS is not resolved, so use + egress filtering in production. + - **The signing secret.** It must be `whsec_` followed by 24 to 64 base64 bytes, and it is stored + encrypted with AES-256-GCM under `MCP_EVENTS_SECRET_KEY`, which must be base64 of at least 32 + random bytes. A refresh with the same secret skips the challenge. If you rotate the key, stored + subscriptions stop receiving events until the host refreshes them. + - **The delivery endpoint.** It verifies QStash's signature with the `QSTASH_*_SIGNING_KEY` + variables (or a `receiver` you pass). Without them it throws on the first request instead of + sending unverified deliveries. A bad signature or body gets `489` with + `Upstash-NonRetryable-Error: true`. + - **Lifetime.** A subscription lasts 7 days by default and 30 days at most. + - **How many.** A subscriber holds at most 8 live subscriptions across all events. Past that, a new + subscribe is refused with reason `subscription_limit`, before the callback is challenged; + refreshing an existing one still works. Each subscription is a webhook per matching emit, so this + bounds what one user can make your server send. Set `maxSubscriptions` to change it, or + `Infinity` to turn it off. + + + + `allowInsecureCallbacks: true` turns the callback URL checks off. Use it only for local + development, never in production. + + + + | Callback answers | What happens | + | --- | --- | + | `2xx` | Delivered. | + | `410 Gone` | The subscription is deleted. | + | `413`, or a redirect | The event is dropped, not retried. | + | Anything else, or no answer | QStash retries with backoff. | + + + + **`mcp-events:sub:`** is the subscription (`event`, `args`, `url`, `encryptedSecret`, + `subscriber`, `createdAt`, `expiresAt`), expiring with it. **`mcp-events:idx:`** and + **`mcp-events:by:`** are sorted sets of the event's and the subscriber's subscription + ids, scored by expiry. The second one enforces the per-subscriber limit. + + An event message carries the full payload, which stays in QStash (and in its DLQ, if every retry + fails) until it is delivered. The toolkit never stores the caller's token, the request, the + plaintext webhook secret or `MCP_EVENTS_SECRET_KEY`. Anyone with write access to your Redis can + change a callback URL, so treat the Redis credentials like any other production secret. + + +## Options + + + **`createEventLayer`**: `store`, `delivery` and `principal` are required, and so is `secretKey` + unless `MCP_EVENTS_SECRET_KEY` is set. Optional: `maxSubscriptions` (8 per subscriber), + `allowInsecureCallbacks`. + + **`events.define(name, config)`**: `description`, `payload` and `authorize` are required. + Optional: `title`, `input`. + + **`RedisSubscriptionStore`**: `redis` (defaults to one from env), `prefix` (`mcp-events:`), + `enableTelemetry`. + + **`QStashDelivery`**: `url` is required. Optional: `qstash`, `receiver`, `enableTelemetry`. + Deliveries are retried 3 times with QStash's backoff. + + + + `@upstash/mcp-toolkit/events` doesn't import anything from Upstash. A backend implements one of + these interfaces (types exported from the same entry point): + + ```ts + interface SubscriptionStore { + // false, storing nothing, when a new one would put its subscriber over `limit` (atomically) + put(subscription: Subscription, options: { limit: number }): Promise; + get(id: string): Promise; + count(subscriber: string): Promise; // live subscriptions, across all events + delete(subscription: { id: string; event: string; subscriber: string }): Promise; + find(event: string): Promise; // every live subscription to the event + } + + interface EventDelivery { + enqueue(jobs: DeliveryJob[]): Promise; + createDeliveryHandler(send: SendJob): (request: Request) => Promise; + } + ``` + + The entry point also exports `verifyWebhook` (Standard Webhooks) for writing a receiver. + + +Not implemented: the draft's poll and stream delivery modes (`events/subscribe` refuses them) and +event replay (`cursor` is always `null`). + +## Example + +The [MCP toolkit demo](https://github.com/upstash/agentkit/tree/main/examples/mcp-toolkit-demo) +includes Deploy Watch, an events-only server with a `deploy.finished` event filtered by +environment, plus a local receiver that verifies each signed delivery. We tested it end to end with +ChatGPT monitors. It needs `MCP_EVENTS_SECRET_KEY` set, and its deploy-report route is +unauthenticated so the demo is easy to drive; a real server must authenticate the code that emits. diff --git a/redis/sdks/agentkit/mcp-tasks.mdx b/redis/sdks/agentkit/mcp-tasks.mdx new file mode 100644 index 000000000..ce5f7acc8 --- /dev/null +++ b/redis/sdks/agentkit/mcp-tasks.mdx @@ -0,0 +1,414 @@ +--- +title: "Long-Running MCP Tools with Redis and QStash" +sidebarTitle: "MCP Tasks" +description: "Durable long-running tools for MCP servers: a tool answers with a task id at once, the work runs on QStash or Upstash Workflow, and the model polls for the result. Works in every MCP client." +--- + +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 tool answers immediately with a task +id, the work runs on [QStash](/qstash/overall/getstarted) or [Upstash Workflow](/workflow/getstarted) +so it survives restarts and the client's tool timeout, and the task record lives in Upstash Redis. +The model checks on it with a shared `task_status` tool and can stop it with `task_cancel`. + +These are ordinary MCP tools, so they work in every client today: Claude Code, Codex, Cursor, +OpenCode and ChatGPT. No client capability is required. 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 mcp-handler zod +``` + +## Quickstart + +Four files in a Next.js app. + +### 1. Identify the caller + +Every task belongs to the user who started it. Two functions, kept in one file so the path is easy +to follow: + +- `verifyToken` runs on every MCP request. It verifies the bearer token and returns the SDK's + `AuthInfo`, with your user id in `extra`. Step 3 wires it up with `withMcpAuth`. +- `principal` is what the toolkit calls. It gets that `AuthInfo` and returns the user id, or throws. + +```ts lib/auth.ts +import type { AuthInfo } from "@modelcontextprotocol/server"; + +// 1. Runs on every MCP request. 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"); + return userId; +} +``` + +### 2. Define the task layer and the task + +```ts lib/tasks.ts +import { createTaskLayer } from "@upstash/mcp-toolkit/tasks"; +import { QStashDispatcher, RedisTaskStore } from "@upstash/mcp-toolkit/upstash"; +import * as z from "zod"; +import { principal } from "./auth"; + +export const tasks = createTaskLayer({ + store: new RedisTaskStore(), + dispatcher: new QStashDispatcher({ url: `${process.env.APP_URL}/api/execute` }), + principal, +}); + +// Define at module scope, so the /api/execute instance knows the handler too. +tasks.define( + "generate_report", + { description: "Generates a report on a topic.", inputSchema: z.object({ topic: z.string() }) }, + async ({ topic }, task) => { + await task.update("Reading sources"); // the model sees this when it polls + return { content: [{ type: "text", text: await writeReport(topic) }] }; + }, +); +``` + +The handler runs later, in `/api/execute`, when QStash delivers the task, not inside the tool call. +`topic` is typed from `inputSchema`: Zod, Valibot and ArkType schemas all work. + +### 3. Serve the MCP endpoint + +```ts app/api/mcp/route.ts +import { createMcpHandler, withMcpAuth } from "mcp-handler"; +import { verifyToken } from "../../../lib/auth"; +import { tasks } from "../../../lib/tasks"; + +const handler = createMcpHandler((server) => { + tasks.register(server); // adds generate_report, task_status and task_cancel +}); + +// 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 }; +``` + +`withMcpAuth` hands the `AuthInfo` from `verifyToken` to the MCP server, and the toolkit passes it +to `principal`. The task layer only registers tools, so nothing else changes in your route. + +### 4. Add the execute endpoint + +```ts app/api/execute/route.ts +import { tasks } from "../../../lib/tasks"; + +export const POST = tasks.createExecuteHandler(); +``` + +This is where QStash delivers each task. The handler verifies the QStash signature, runs the task's +handler and stores the result. + +### Environment variables + +```bash .env +UPSTASH_REDIS_REST_URL=... +UPSTASH_REDIS_REST_TOKEN=... +QSTASH_TOKEN=... +# Required: the execute endpoint refuses to run without them. +QSTASH_CURRENT_SIGNING_KEY=... +QSTASH_NEXT_SIGNING_KEY=... +# Where QStash delivers tasks. Must be publicly reachable. +APP_URL=https://your-app.vercel.app +``` + + + 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 use `APP_URL=http://127.0.0.1:3000`. + + +## What the model sees + +`tools/list` now has `generate_report` (starts the task and returns its id), `task_status` (shows +progress, then the handler's result) and `task_cancel`. + + + ```jsonc + // generate_report: a handle, right away + { "content": [{ "type": "text", "text": "Started task 0e30…. Call task_status with taskId \"0e30…\" in about 2s to check on it." }], + "structuredContent": { "taskId": "0e30…", "status": "working", "ttlMs": 86400000, "pollIntervalMs": 2000 } } + + // task_status: progress… + { "content": [{ "type": "text", "text": "Task 0e30… is working: Reading sources. Check again in about 2s." }], + "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: Completed" }, + { "type": "text", "text": "Report on coffee" }], + "structuredContent": { "taskId": "0e30…", "status": "completed", "result": { "content": [ … ] } } } + ``` + + The states are `working`, `completed`, `failed` and `cancelled`; the last three are final. The + task object has the same shape as the one in the MCP Tasks extension. + + 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 the layer). + + +## Cancellation + +`task_cancel` marks the task `cancelled` and stops any pending delivery. Cancelling is cooperative, +so running code only stops where it checks `task.isCancelled()`: + +```ts +async ({ sources }, task) => { + for (const source of sources) { + if (await task.isCancelled()) return {}; + await task.update(`Reading ${source}`); + await read(source); + } + return { content: [{ type: "text", text: "Done" }] }; +}; +``` + +`isCancelled()` also returns `true` once the task has expired. + + + 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. + + +## Users + +Each task records who started it, and `task_status` / `task_cancel` only answer for that caller. +Another user's task id reads exactly like an unknown one, so nobody can tell it exists. + +### 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. If Alice starts a task with Bob's +`workspaceId`, the task is hers, she can read its result, and the handler acted on Bob's workspace. + +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 lib/tasks.ts +tasks.define( + "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 later, in a delivery request that carries no user auth, so `task.principal` is how it +knows who it is working for. + +### Where principal comes from + +`principal` receives `{ auth, request }` and may be async. `auth` is exactly what `verifyToken` +returned in [step 1](#1-identify-the-caller): `withMcpAuth` runs it, and the toolkit hands the result +to `principal`. Nothing is read from request headers on its own. Without `mcp-handler`, pass the +`AuthInfo` yourself with the SDK's `createMcpHandler`: `handler.fetch(request, { authInfo })`. + + + - **A throw refuses the call.** There is no anonymous mode: if `principal` throws, rejects, or + returns anything but a non-empty string, the tool 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. + - `principal` also receives `request`, for cookie or session apps. It is unverified, so check the + session yourself, and never trust a header like `x-user-id`: + + ```ts + import type { Caller } from "@upstash/mcp-toolkit/tasks"; + + export async function principal({ request }: Caller): Promise { + const session = await getSession(request); + if (!session) throw new Error("Not authenticated"); + return session.userId; + } + ``` + + - A server with no users of its own passes `principal: () => "local"`. + + +## 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 routes stay the same. The +handler's `task` then also has `run`, `sleep` and `call`, inferred from the dispatcher: + +```ts lib/tasks.ts +import { RedisTaskStore, WorkflowDispatcher } from "@upstash/mcp-toolkit/upstash"; + +export const tasks = createTaskLayer({ + store: new RedisTaskStore(), + dispatcher: new WorkflowDispatcher({ url: `${process.env.APP_URL}/api/execute` }), + principal, +}); + +tasks.define( + "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" }] }; + }, +); +``` + + + 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. + + | | `QStashDispatcher` | `WorkflowDispatcher` | + | --- | --- | --- | + | Survives the process dying | yes (QStash redelivers) | yes (replayed from the journal) | + | Can run longer than one invocation | no | yes, one invocation per step | + | On a retry | the whole task starts over | only the failed step runs again | + | Cancel stops a running task | at its next `isCancelled` | the run itself is cancelled | + + +## Retries and failures + + + - The execute route 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: true` 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. + This holds for `WorkflowDispatcher` too. Without signing keys (or a `receiver` you pass), the + route throws on its first request 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. + - 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. + + +## Security and storage + + + 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` and + `task_cancel` only accept UUID task ids, and compare the stored owner with the caller. Task ids are + random UUIDs. No tool lists tasks, and the execute route 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. + + + + **`mcp:task:`** is a hash with one JSON-encoded field per property: `taskId`, `name`, + `args`, `owner`, `status`, `statusMessage`, `result` / `error`, `createdAt`, `lastUpdatedAt`, + `ttlMs`, `pollIntervalMs`. + + - `args` and `result` are **plain JSON**. Keep secrets out of tool arguments and results, 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. + + A task message in QStash carries only `{ taskId }`. The toolkit never stores the caller's token or + the request. + + +## Options + + + **`createTaskLayer`**: `store`, `dispatcher` and `principal` are required. Optional: + `defaults.ttlMs` (1 day) and `defaults.pollIntervalMs` (2 seconds), in positive whole milliseconds. + + **`tasks.define(name, config, handler)`**: `description` and `inputSchema` are required. Optional: + `title`, `completedMessage` (defaults to `"Completed"`), `authorize(args, { principal, auth, request })`. + + **`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@` added to their `Upstash-Telemetry-Sdk` header; turn it + off with `enableTelemetry: false` or `UPSTASH_DISABLE_TELEMETRY`. + + + + `@upstash/mcp-toolkit/tasks` doesn't import anything from Upstash. A backend implements one of these + interfaces (types exported from the same entry point): + + ```ts + interface TaskStore { + create(task: Task): Promise; // with its TTL + get(taskId: string): Promise; + update(taskId: string, patch: TaskPatch): Promise; // ignored once the task is final + settle(taskId: string, patch: TaskPatch & { status: TerminalTaskStatus }): Promise; // first final status wins + } + + interface TaskDispatcher { + dispatch(task: Task): Promise; // called once per task + cancel(taskId: string): Promise; + createExecuteHandler(endpoints: TaskEndpoints): (request: Request) => Promise; + } + ``` + + +## Why tools, not the Tasks extension? + +The 2026-07-28 MCP specification defines a [Tasks extension](https://modelcontextprotocol.io/extensions/tasks/overview) +where the *client* polls, so the model spends no turns waiting. But a server may only return a task +to a client that declared the extension, and as of October 2026 Claude Code, Codex, Cursor and +OpenCode don't. Plain tools cost the model a few polling calls but work everywhere today. + +The store and dispatcher don't depend on the tools, so an adapter for the extension can serve the +same records once clients support it. Also not implemented yet: `input_required` (a handler asking +the user something partway through a task) and listing tasks. + +## 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.