DX-3022: Add @upstash/mcp-toolkit: long-running MCP tools (tasks) and MCP Events

.changeset/mcp-toolkit-initial.md

@@ -0,0 +1,31 @@
+---
+"@upstash/mcp-toolkit": minor
+---
+
+Add `@upstash/mcp-toolkit`: durable building blocks for MCP servers on the official TypeScript SDK.
+
+`@upstash/mcp-toolkit/tasks` — long-running tools. A task tool answers immediately with a task id;
+the model polls the shared `task_status` tool for progress and the result, and can stop the task
+with `task_cancel`. They are ordinary MCP tools, so they work in every client today. The record
+lives in a `TaskStore` and the work runs behind a `TaskDispatcher`, and the dispatcher owns its
+delivery endpoint (`export const POST = tasks.createExecuteHandler()`) and always verifies QStash's
+signature against the URL it published to. Tasks are declared once at module scope with
+`tasks.define(...)` and attached to each request's server with `tasks.register(server)`. A required
+`principal` scopes tasks to their caller (and subscriptions to theirs); handlers get it as
+`task.principal`, and an optional `authorize` on `tasks.define` checks the arguments before a task
+is stored or queued.
+
+`@upstash/mcp-toolkit/events` — MCP Events with webhook delivery, as ChatGPT ships it. Typed
+`events.define(...)` handles with `emit(payload)`: the payload's values route the event and are what
+the required `authorize` is checked against, at subscribe and again before every delivery.
+`events/list`, `events/subscribe` and `events/unsubscribe` are registered on your server, with a
+signed verification challenge, deterministic subscription ids, expiry and refresh, a configurable
+limit of live subscriptions per subscriber (8 by default), SSRF checks on callback URLs, and signing
+secrets encrypted at rest. `QStashDelivery` signs each attempt with
+Standard Webhooks and lets QStash retry failures (`export const POST = events.createDeliveryHandler()`).
+
+The package uses WebCrypto only, so it runs on Node and edge runtimes.
+
+`@upstash/mcp-toolkit/upstash` holds the Upstash backends for both: `RedisTaskStore`,
+`QStashDispatcher`, `WorkflowDispatcher`, `RedisSubscriptionStore` and `QStashDelivery`.
+`@upstash/redis`, `@upstash/qstash` and `@upstash/workflow` are regular dependencies.

.github/workflows/ci.yml

@@ -78,6 +78,24 @@ jobs:
           fi
           pnpm e2e
 
+      - name: E2E (MCP toolkit demo, local QStash, no model)
+        # Starts the local QStash dev server and the demo built above, and drives it with raw JSON-RPC
+        # as a model and a host would (no model, no QStash account). Against both task servers: run to
+        # completion, cancel, another user's task reads as unknown, a throwing task ends `failed`,
+        # unsigned or forged deliveries are refused. Against Deploy Watch: filtered webhooks with
+        # valid signatures, per-user subscriptions, `authorize` checked at delivery, 410 deletes the
+        # subscription. Skipped without Redis secrets (e.g. fork PRs).
+        working-directory: examples/mcp-toolkit-demo
+        env:
+          UPSTASH_REDIS_REST_URL: ${{ secrets.UPSTASH_REDIS_REST_URL }}
+          UPSTASH_REDIS_REST_TOKEN: ${{ secrets.UPSTASH_REDIS_REST_TOKEN }}
+        run: |
+          if [ -z "$UPSTASH_REDIS_REST_URL" ]; then
+            echo "No Redis secrets available — skipping the MCP toolkit e2e suite."
+            exit 0
+          fi
+          pnpm e2e
+
       - name: E2E eval (eve memory slots, mocked model)
         # Boots the eve demo's agent with a scripted mockModel (no model provider, no
         # OPENAI_API_KEY) and asserts both Upstash Redis memory integrations run at eve's real

README.md

@@ -15,6 +15,7 @@ are powered by [Upstash Redis Search](https://upstash.com/docs/redis/search/intr
 | [`@upstash/agentkit-eve`](./packages/eve) | Adapter for the Vercel Eve framework. |
 | [`@upstash/agentkit-tanstack-ai`](./packages/tanstack-ai) | Production backends for [TanStack AI](https://tanstack.com/ai) — chat persistence, resumable streams, distributed locks, memory, tool caching, rate limiting and search tools. |
 | [`@upstash/agentkit-eve-extension`](./packages/eve-extension) | The same capabilities as a mountable [Eve extension](https://eve.dev/docs/extensions) — one file in `agent/extensions/` adds memory tools, search tools, and durable chat history the agent can search. |
+| [`@upstash/mcp-toolkit`](./packages/mcp-toolkit) | Durable building blocks for MCP servers on the official TypeScript SDK. `/tasks`: long-running tools that answer with a task id the model polls, with the record in Redis and the work on QStash or Workflow. `/events`: MCP Events, with subscriptions in Redis and signed webhook deliveries retried by QStash. |
 
 ## Core features
 
@@ -33,15 +34,21 @@ are powered by [Upstash Redis Search](https://upstash.com/docs/redis/search/intr
 - **Code sandbox** (Eve only) — an [Upstash Box](https://github.com/upstash/box) sandbox provider for
   Eve's `defineSandbox` (`UpstashSandbox`, eve ≥ 0.65).
 - **Tool-call cache** — memoize deterministic tool results keyed by arguments.
+- **Durable MCP tasks and events** (`@upstash/mcp-toolkit`) — a long-running MCP tool returns a task
+  id instead of blocking, and the model polls `task_status`; the record lives in Redis and the work
+  runs through QStash or Workflow, so it survives the process that accepted the call. MCP Events let
+  a host subscribe and get a signed webhook when something happens, including when a task finishes.
 
 ## Examples
 
 Runnable demos (real Upstash Redis + a mock/real model) live in [`examples/`](./examples):
 [`ai-sdk-demo`](./examples/ai-sdk-demo), [`eve-demo`](./examples/eve-demo),
 [`eve-extension-demo`](./examples/eve-extension-demo) (an eve agent that mounts
-`@upstash/agentkit-eve-extension`), and [`tanstack-ai-demo`](./examples/tanstack-ai-demo) (a
+`@upstash/agentkit-eve-extension`), [`tanstack-ai-demo`](./examples/tanstack-ai-demo) (a
 TanStack AI chat whose persistence, resumable stream and memory work across server instances, with
-a two-instance E2E suite).
+a two-instance E2E suite), and [`mcp-toolkit-demo`](./examples/mcp-toolkit-demo) (an MCP server whose
+long-running tool returns a task id the model polls, plus a separate events server that fires `deploy.finished`, with the client's
+wire log on screen).
 
 ## Development
 

examples/mcp-toolkit-demo/.env.example

@@ -0,0 +1,21 @@
+# Upstash Redis — the durable task record.
+UPSTASH_REDIS_REST_URL=
+UPSTASH_REDIS_REST_TOKEN=
+
+# QStash — the durable execution transport.
+#
+# For local development run `npx @upstash/qstash-cli dev` in another terminal. It prints a URL, a
+# token and both signing keys; paste them here. For a deployed app use the real values from the
+# Upstash console instead and drop QSTASH_URL.
+QSTASH_URL=http://127.0.0.1:8080
+QSTASH_TOKEN=
+QSTASH_CURRENT_SIGNING_KEY=
+QSTASH_NEXT_SIGNING_KEY=
+
+# Encrypts the hosts' webhook signing secrets at rest (Deploy Watch). Required, no default.
+# Generate one with: openssl rand -base64 32
+MCP_EVENTS_SECRET_KEY=
+
+# Where QStash delivers a task. Must be reachable *from QStash*: the local dev server can reach
+# 127.0.0.1, the hosted service cannot — use your deployment URL or a tunnel there.
+APP_URL=http://127.0.0.1:3000

examples/mcp-toolkit-demo/.gitignore

@@ -0,0 +1,4 @@
+node_modules
+.next
+.env*
+!.env.example

examples/mcp-toolkit-demo/README.md

@@ -0,0 +1,134 @@
+# MCP toolkit demo
+
+A Next.js app showing `@upstash/mcp-toolkit` end to end: an MCP tool that answers with a task handle
+instead of blocking, a task record in Upstash Redis, and the work running through QStash (or
+Upstash Workflow) so it survives the process that accepted the call.
+
+The page is the MCP client, and it does exactly what a model would: call `generate_report`, then
+poll `task_status` and maybe call `task_cancel` — three ordinary tools, from a client that declares
+no capabilities at all. It speaks raw stateless JSON-RPC and shows every frame in a wire log.
+
+## What's here
+
+| File | What it does |
+| --- | --- |
+| `app/lib/qstash-server.ts` | Server one: Redis store, QStash dispatcher, and the `generate_report` task tool |
+| `app/lib/workflow-server.ts` | Server two: the same tool on Upstash Workflow, one invocation per step |
+| `app/api/mcp/route.ts`, `app/api/mcp-workflow/route.ts` | The MCP endpoints — the SDK's own `createMcpHandler`, unchanged |
+| `app/api/execute/route.ts`, `app/api/execute-workflow/route.ts` | Where the work is delivered. One line each: the dispatcher owns the endpoint |
+| `app/lib/deploy-watch.ts` | **Deploy Watch**, the events server: the `deploy.finished` event and a `list_recent_deploys` tool |
+| `app/api/deploy-watch/route.ts` | Deploy Watch's MCP endpoint |
+| `app/api/deploy-watch/events/route.ts` | Where QStash delivers each event. One line: the delivery owns the endpoint |
+| `app/api/deploy-watch/deploys/route.ts` | Report a deploy (stands in for your CI), which fires `deploy.finished` |
+| `app/api/receiver/route.ts` | A stand-in for the host's webhook receiver, so events can be seen without a public URL |
+| `app/lib/auth.ts` | Stand-in authentication: `Bearer demo-<name>` is user `demo-<name>`, no header is `demo-user` |
+| `app/page.tsx` | The client: call the tool, poll, cancel, and the wire log |
+| `scripts/smoke.mjs` | The e2e checks, as raw JSON-RPC against a running app |
+| `scripts/e2e.mjs` | Starts the QStash dev server and the built app, and runs the checks against both task servers |
+| `app/lib/e2e.ts` | Test hooks for `scripts/e2e.mjs` only (`MCP_TOOLKIT_E2E=1`) |
+
+## Run it
+
+You need an [Upstash Redis database](https://upstash.com/start-redis). QStash you can run locally,
+fully offline.
+
+```bash
+cp .env.example .env.local     # fill in UPSTASH_REDIS_REST_URL / _TOKEN, and
+                               # MCP_EVENTS_SECRET_KEY=$(openssl rand -base64 32)
+
+pnpm qstash                    # terminal 1 — prints the QStash URL, token and signing keys
+                               #              paste those four into .env.local
+pnpm dev                       # terminal 2
+```
+
+Then open http://localhost:3000, type a topic, and hit **Run tool**.
+
+`APP_URL` is the one setting worth reading twice: it is where QStash delivers the task, so it has to
+be reachable *from QStash*. The local dev server can reach `127.0.0.1`; the hosted service cannot,
+so a deployed app needs its real URL (or a tunnel) there.
+
+## Tests
+
+No model and no QStash account: the checks speak raw JSON-RPC, as a model and a host would.
+
+```bash
+pnpm build && pnpm e2e    # starts the QStash dev server and the app, runs everything, stops them
+```
+
+Against both task servers it checks a task running to completion, cancel, another user's task
+reading as unknown (`Bearer demo-alice` vs `Bearer demo-bob`), a bad token getting 401, a task that
+throws ending `failed`, and unsigned or forged deliveries being refused. Against Deploy Watch: a
+filtered subscription getting exactly one validly signed webhook, another user's unsubscribe
+changing nothing, `authorize` refusing `demo-intern` production deploys at subscribe time and at
+delivery, and a host's 410 deleting the subscription. CI runs it on every PR.
+
+Against an app you already have running (`pnpm dev` + `pnpm qstash`), `pnpm smoke` runs the same
+checks, and `MCP_PATH=/api/mcp-workflow pnpm smoke` the Workflow ones. The failure check is
+skipped there unless the app was started with `MCP_TOOLKIT_E2E=1`.
+
+## Two separate tests
+
+The demo serves two unrelated MCP servers, so tasks and events can be tried on their own:
+
+| | Report Desk (tasks) | Deploy Watch (events) |
+| --- | --- | --- |
+| Endpoint | `/api/mcp` | `/api/deploy-watch` |
+| What it has | `generate_report` + `task_status` / `task_cancel` | the `deploy.finished` event + `list_recent_deploys` |
+| Works in | every MCP client | hosts that support MCP Events (ChatGPT) |
+| Try it | ask the agent for a report; it polls until done | subscribe in a ChatGPT Work chat, then `POST /api/deploy-watch/deploys` |
+
+For ChatGPT, `APP_URL` stays `http://127.0.0.1:3000` (QStash runs next to the app), and the app
+needs a public URL for ChatGPT to reach the MCP endpoint, e.g. a tunnel or an Upstash Box preview.
+
+## Two transports
+
+The demo runs two servers side by side, picked in the UI. They expose the same tools; only
+durability differs:
+
+- **QStash** (`/api/mcp`): one delivery, one invocation. Survives a crash, but the whole handler
+  has to finish inside the route's `maxDuration`.
+- **Workflow** (`/api/mcp-workflow`): each `task.run(...)` becomes its own request, replayed from a
+  journal, so the task can run far longer than any one invocation.
+
+## The three things worth watching
+
+**A tool call returns immediately.** `generate_report` comes back in milliseconds with a
+`taskId` and a `working` status in `structuredContent`, plus a sentence telling the model to call
+`task_status`. The four-step report takes about ten seconds; none of it happens inside that
+request. Once it completes, `task_status` returns the report's own content.
+
+**Cancel is cooperative, in three layers.** Hit **cancel** mid-run and the store flips the status
+to `cancelled`, any redelivery finds the task settled and does nothing (on Workflow the run itself
+is cancelled), and the handler stops at its next step boundary. The last layer is the one you cannot skip: running code only stops where it
+checks. A completion arriving after the cancel is refused — terminal states are final.
+
+**The work is durable, not just the record.** Start a task and kill the dev server mid-run:
+
+```bash
+pnpm dev
+# start a task in the browser, then, a few seconds in:
+kill -9 $(lsof -ti tcp:3000)
+pnpm dev
+```
+
+Keep polling (the page resumes on its own) and the task still reaches `completed`. Redis kept the
+record; QStash's redelivery is what finished the work. Replace the dispatcher with a
+fire-and-forget promise and the same test leaves a permanently `working` task instead.
+
+One caveat this demo learned the hard way: the retry budget has to outlast your restart. QStash
+retries on its configured schedule, and when the attempts run out its failure callback marks the
+task `failed`. With a flat one-second delay every attempt is spent within a few seconds, long
+before a dev server is back up. The dispatcher's defaults spread five attempts over about two
+minutes (1s, 3s, 9s, 27s, 81s) for that reason; five is also the ceiling the local dev server and
+the free tier allow, so raising `retries` needs a plan that permits it. A failed message stays in
+the QStash DLQ.
+
+## Notes
+
+- All three tools are ordinary MCP tools, so this works in every client today — Claude Code,
+  Codex, Cursor, OpenCode, ChatGPT — none of which declare the protocol's Tasks extension yet.
+- The demo has no real login. `app/lib/auth.ts` stands in for one: the MCP routes turn
+  `Authorization: Bearer demo-<name>` into the user `demo-<name>` and pass it to the SDK as
+  `authInfo`, and `principal` reads it back, as a real server would with a verified token. With no
+  header the caller is the shared `demo-user`, so the page and an unauthenticated ChatGPT
+  connection keep working; a real server answers 401 instead.

examples/mcp-toolkit-demo/app/api/deploy-watch/deploys/route.ts

@@ -0,0 +1,30 @@
+/**
+ * Report a deploy, which fires `deploy.finished`. This stands in for your CI telling your server a
+ * deploy finished. POST a partial deploy; anything left out gets a demo value:
+ *
+ *   curl -X POST $APP/api/deploy-watch/deploys -d '{"service":"checkout-api","status":"failed"}'
+ *
+ * Unauthenticated on purpose, so the demo is easy to drive. A real server must authenticate this
+ * route (your CI's token, a signed webhook): anyone who can call it can fire events.
+ */
+import { randomBytes } from "node:crypto";
+import { Deploy, reportDeploy } from "../../../lib/deploy-watch";
+
+export const dynamic = "force-dynamic";
+
+export async function POST(request: Request): Promise<Response> {
+  const input = (await request.json().catch(() => ({}))) as Partial<Deploy>;
+  const deploy = Deploy.parse({
+    id: randomBytes(4).toString("hex"),
+    service: "checkout-api",
+    environment: "production",
+    status: "succeeded",
+    commit: "Retry card payments once on gateway timeouts",
+    author: "arda",
+    durationSeconds: 94,
+    url: "https://example.com/deploys",
+    ...input,
+  });
+  const result = await reportDeploy(deploy);
+  return Response.json({ deploy, ...result });
+}

examples/mcp-toolkit-demo/app/api/deploy-watch/events/route.ts

@@ -0,0 +1,10 @@
+/**
+ * Where QStash delivers each `deploy.finished` event. The handler verifies QStash's signature, signs
+ * the envelope with the subscriber's secret, POSTs it, and answers 500 when the host's callback
+ * failed so QStash retries with backoff.
+ */
+import { events } from "../../../lib/deploy-watch";
+
+export const dynamic = "force-dynamic";
+
+export const POST = events.createDeliveryHandler();

examples/mcp-toolkit-demo/app/api/deploy-watch/route.ts

@@ -0,0 +1,13 @@
+/** The MCP endpoint for **Deploy Watch**, the events demo. */
+import { createMcpHandler } from "@modelcontextprotocol/server";
+import { serveMcp } from "../../lib/auth";
+import { createServer } from "../../lib/deploy-watch";
+
+export const dynamic = "force-dynamic";
+
+const handler = createMcpHandler(() => createServer());
+
+export async function POST(request: Request): Promise<Response> {
+  // Verifies the caller and passes `authInfo` to the SDK, which hands it to `principal`.
+  return serveMcp(handler, request);
+}

examples/mcp-toolkit-demo/app/api/execute-workflow/route.ts

@@ -0,0 +1,12 @@
+/**
+ * Where Upstash Workflow runs each step of a task.
+ *
+ * Also one line — but this endpoint is called once *per step*, so no single invocation has to
+ * cover the whole task and `maxDuration` bounds a step rather than the work.
+ */
+import { tasks } from "../../lib/workflow-server";
+
+export const dynamic = "force-dynamic";
+export const maxDuration = 60;
+
+export const POST = tasks.createExecuteHandler();

examples/mcp-toolkit-demo/app/api/execute/route.ts

@@ -0,0 +1,15 @@
+/**
+ * Where QStash delivers a task — and, once retries are exhausted, its failure callback.
+ *
+ * One line, because everything that has to be right here belongs to the transport: verifying the
+ * signature, reading the task id, telling a delivery from a failure callback, and choosing the
+ * status code that decides whether QStash tries again.
+ */
+import { tasks } from "../../lib/qstash-server";
+
+export const dynamic = "force-dynamic";
+// The demo tool sleeps ~10s and must finish inside this one invocation. That limit is the reason
+// the workflow server exists.
+export const maxDuration = 60;
+
+export const POST = tasks.createExecuteHandler();

examples/mcp-toolkit-demo/app/api/mcp-workflow/route.ts

@@ -0,0 +1,13 @@
+/** The MCP endpoint for the **Workflow** server. Same handler, different task layer. */
+import { createMcpHandler } from "@modelcontextprotocol/server";
+import { serveMcp } from "../../lib/auth";
+import { createServer } from "../../lib/workflow-server";
+
+export const dynamic = "force-dynamic";
+
+const handler = createMcpHandler(() => createServer());
+
+export async function POST(request: Request): Promise<Response> {
+  // Verifies the caller and passes `authInfo` to the SDK, which hands it to `principal`.
+  return serveMcp(handler, request);
+}

examples/mcp-toolkit-demo/app/api/mcp/route.ts

@@ -0,0 +1,18 @@
+/**
+ * The MCP endpoint for the **QStash** server.
+ *
+ * The SDK's own `createMcpHandler`, unchanged: the task layer only registers ordinary tools, so it
+ * needs no custom transport and no extra protocol methods.
+ */
+import { createMcpHandler } from "@modelcontextprotocol/server";
+import { serveMcp } from "../../lib/auth";
+import { createServer } from "../../lib/qstash-server";
+
+export const dynamic = "force-dynamic";
+
+const handler = createMcpHandler(() => createServer());
+
+export async function POST(request: Request): Promise<Response> {
+  // Verifies the caller and passes `authInfo` to the SDK, which hands it to `principal`.
+  return serveMcp(handler, request);
+}