DX-3022: mcp-toolkit: task authorize + principal, subscription limit, review fixes

- Tasks: `task.principal` (the owner) in TaskContext; optional `authorize(args, { principal, auth, request })` on tasks.define, run before the task is stored or queued
- Tasks: reject non-positive / non-integer `defaults.ttlMs` and `pollIntervalMs`
- Events: `maxSubscriptions` (default 8, Infinity = off) live subscriptions per subscriber, checked before the callback challenge and atomically in the store (Lua over idx:<event> and by:<subscriber>); SubscriptionStore gains count(), put() takes { limit } and returns boolean, delete() takes subscriber
- Events: route and authorize deliveries by the payload's input fields parsed with `input`, so transforms apply on both sides; a payload that doesn't fit `input` makes emit throw
- Drop QStash deduplicationId for tasks and events (Copilot: the sanitized event id was lossy); the event id still reaches the host as webhook-id
- @upstash/redis floor ^1.38.4; aria-pressed on the demo's transport toggle

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gtucop3k8CNVLjgvoDTC8n

Author: Upstash Box

.changeset/mcp-toolkit-initial.md

@@ -11,14 +11,17 @@ lives in a `TaskStore` and the work runs behind a `TaskDispatcher`, and the disp
 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).
+`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, SSRF checks on
-callback URLs, and signing secrets encrypted at rest. `QStashDelivery` signs each attempt with
+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.

CLAUDE.md

@@ -762,11 +762,10 @@ implements TanStack AI's own backend contracts (see its section below) — keep
   `dispatcher.cancel(taskId)` when the returned status is `cancelled` — `cancel` must be
   idempotent. There is no `dispatchId`: `QStashDispatcher.cancel` is a no-op (a redelivery of a
   cancelled task finds it settled and does nothing), Workflow's run id is the task id.
-- **Dispatch dedupe on the task id:** QStash `deduplicationId` and Workflow `workflowRunId` are the
-  task id. That is only safe because ids
-  are random: if keyed (deterministic) ids ever come back, dedupe must move to a per-record key
-  (taskId + createdAt), since QStash keeps dedup ids for 10 minutes and Workflow refuses a reused
-  run id. A failed `dispatch` settles the fresh record `failed` ("Could not be queued") and rethrows.
+- **No QStash `deduplicationId`** (tasks or events): the layer dispatches each task once, and an
+  event's id reaches the host as `webhook-id`, which is where Standard Webhooks dedupes. Workflow's
+  `workflowRunId` is the task id; that is only safe because ids are random (Workflow refuses a
+  reused run id). A failed `dispatch` settles the fresh record `failed` ("Could not be queued") and rethrows.
 - **Workflow context is inferred** from the dispatcher (`TaskDispatcher<TContext>`), so
   `createTaskLayer({ dispatcher: new WorkflowDispatcher(...) })` types `task.run` / `task.sleep` with
   no type argument; a test in `workflow.test.ts` keeps it that way.
@@ -873,8 +872,14 @@ implements TanStack AI's own backend contracts (see its section below) — keep
   **every IP literal** (v4 after the URL parser's normalization, and any `[…]` v6) plus
   localhost/single-label/`.local`/`.internal`; the old private-range/NAT64/6to4 parser is gone.
   It does not resolve DNS. Bad params are `-32602` with a `reason`.
-  **QStash `deduplicationId` cannot contain `:`** (dev server answers 400) — ids are
-  `${eventId}_${subscriptionId}` sanitized; this was caught only by the e2e smoke.
+  **Subscription limit:** `maxSubscriptions` (default 8, `Infinity` = off) live subscriptions per
+  subscriber across all events. The layer checks `store.count` before the challenge (no outbound
+  POST for a caller at the limit), and `put(sub, { limit })` re-checks atomically (Redis: one Lua
+  script over `idx:<event>` and `by:<subscriber>`). Refreshes never count. Routing and delivery-time
+  `authorize` use the payload's input fields *parsed with `input`* (`projection`), so transforms
+  match on both sides; a payload that doesn't fit `input` makes `emit` throw.
+  QStash `deduplicationId` cannot contain `:` (dev server answers 400), one more reason not to
+  derive it from caller-chosen event ids.
 - **Local dev needs the QStash dev server** (`npx @upstash/qstash-cli dev`) — it prints deterministic
   creds. `APP_URL` must be reachable *from QStash*.
 

examples/mcp-toolkit-demo/app/page.tsx

@@ -161,6 +161,7 @@ export default function Page() {
                     key={key}
                     type="button"
                     className={`driver ${server === key ? "on" : ""}`}
+                    aria-pressed={server === key}
                     onClick={() => setServer(key)}
                   >
                     {SERVERS[key].label}

packages/mcp-toolkit/README.md

@@ -121,6 +121,27 @@ progress, then the handler's result) and `task_cancel`. To stop a cancelled task
 `await task.isCancelled()` between steps. Cancelling is cooperative, so running code only stops
 where it checks.
 
+**Check what the arguments point at.** The model fills in tool arguments, so a `workspaceId` in
+them is whatever it was told. Give the task an `authorize`, which runs before the task is stored or
+queued, and use `task.principal` (the user who started the task) inside the handler. Never take a
+user id from the arguments:
+
+```ts
+tasks.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 }] };
+  },
+);
+```
+
 <details>
 <summary><b>What the model sees</b></summary>
 
@@ -166,9 +187,13 @@ export const tasks = createTaskLayer({
 
 tasks.define(
   "migrate_workspace",
-  { description: "Copies a workspace to new storage.", inputSchema: z.object({ workspaceId: z.string() }) },
+  {
+    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));
+    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}`);
@@ -278,8 +303,10 @@ the event id stays the same across retries so the host can drop duplicates.
 <summary><b>Matching and <code>authorize</code></b></summary>
 
 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. A subscription matches
-when each argument it gave equals the payload's value. Emitting
+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 apply on both sides and
+`authorize` gets the types it declares. 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:
@@ -294,8 +321,8 @@ when each argument it gave equals the payload's value. Emitting
 
 `() => true` lets every authenticated subscriber hear every matching event.
 
-Pass `{ eventId }` as the second argument to `emit` to deduplicate: emitting the same id twice
-delivers once.
+Pass `{ eventId }` as the second argument to `emit` to give an event a stable id. It is sent as
+`webhook-id`, so a host drops a second emit with the same id as a duplicate.
 
 </details>
 
@@ -318,6 +345,11 @@ delivers once.
   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.
 - **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 (set
+  `maxSubscriptions`; `Infinity` turns it off). Past that, a new subscribe is refused before the
+  callback is challenged, with reason `subscription_limit`; 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.
 - **Host answers.** `410` deletes the subscription, `413` and redirects drop the event, and any
   other error is retried.
 - **Not implemented:** the draft's poll and stream delivery modes (`events/subscribe` refuses
@@ -341,7 +373,9 @@ subscribe yet. The spec draft is
 <details>
 <summary><b>Who can see what</b></summary>
 
-**Tasks.** The server sets the owner from `principal`, and no tool argument can set it.
+**Tasks.** The server sets the 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. For anyone else, the task looks exactly like an unknown id, so they can't even tell it
 exists. Task ids are random UUIDs. No tool lists tasks, and the execute route only accepts signed
@@ -375,8 +409,9 @@ Treat the Redis credentials like any other production secret.
 - A Redis client built with `automaticDeserialization: false` is not supported.
 
 **`mcp-events:sub:<id>`**: the subscription (`event`, `args`, `url`, `encryptedSecret`,
-`subscriber`, `createdAt`, `expiresAt`), expiring with it. **`mcp-events:idx:<event>`**: a sorted
-set of the event's subscription ids, scored by expiry.
+`subscriber`, `createdAt`, `expiresAt`), expiring with it. **`mcp-events:idx:<event>`** and
+**`mcp-events:by:<subscriber>`**: sorted sets of the event's and the subscriber's subscription ids,
+scored by expiry. The second one enforces the limit.
 
 **Not in Redis.** A task message in QStash carries only `{ taskId }`. An event message carries the
 full payload, which stays in QStash (and in its DLQ, if every retry fails) until it is delivered.
@@ -389,13 +424,13 @@ The toolkit never stores the caller's token, the request, the plaintext webhook
 <summary><b>All options</b></summary>
 
 **`createTaskLayer`**: `store`, `dispatcher` and `principal` are required. Optional:
-`defaults.ttlMs` (1 day) and `defaults.pollIntervalMs` (2s).
+`defaults.ttlMs` (1 day) and `defaults.pollIntervalMs` (2s), in positive whole milliseconds.
 
 **`tasks.define(name, config, handler)`**: `description` and `inputSchema` are required. Optional:
-`title`, `completedMessage`.
+`title`, `completedMessage`, `authorize(args, { principal, auth, request })`.
 
 **`createEventLayer`**: `store`, `delivery` and `principal` are required, and so is `secretKey`
-unless `MCP_EVENTS_SECRET_KEY` is set. Optional: `allowInsecureCallbacks`.
+unless `MCP_EVENTS_SECRET_KEY` is set. Optional: `maxSubscriptions` (8), `allowInsecureCallbacks`.
 
 **`events.define(name, config)`**: `description`, `payload` and `authorize` are required.
 Optional: `title`, `input`.
@@ -436,15 +471,17 @@ interface TaskStore {
 }
 
 interface TaskDispatcher<TContext = unknown> {
-  dispatch(task: Task): Promise<void>; // idempotent per task id
+  dispatch(task: Task): Promise<void>; // called once per task
   cancel(taskId: string): Promise<void>;
   createExecuteHandler(endpoints: TaskEndpoints<TContext>): (request: Request) => Promise<Response>;
 }
 
 interface SubscriptionStore {
-  put(subscription: Subscription): Promise<void>;
+  // false, storing nothing, when a new one would put its subscriber over `limit` (atomically)
+  put(subscription: Subscription, options: { limit: number }): Promise<boolean>;
   get(id: string): Promise<Subscription | null>;
-  delete(subscription: { id: string; event: string }): Promise<void>;
+  count(subscriber: string): Promise<number>; // live subscriptions, across all events
+  delete(subscription: { id: string; event: string; subscriber: string }): Promise<void>;
   find(event: string): Promise<Subscription[]>; // every live subscription to the event
 }
 

packages/mcp-toolkit/package.json

@@ -53,7 +53,7 @@
   ],
   "dependencies": {
     "@upstash/qstash": "^2.11.3",
-    "@upstash/redis": "^1.38.0",
+    "@upstash/redis": "^1.38.4",
     "@upstash/workflow": "^1.3.3",
     "zod": "^4.2.0"
   },

packages/mcp-toolkit/src/events/backends/qstash.test.ts

@@ -35,14 +35,14 @@ describe.skipIf(!hasRedisCreds)("RedisSubscriptionStore (real Redis)", () => {
 
   it("round-trips a subscription, keeping a numeric-looking subscriber a string", async () => {
     const sub = makeSub();
-    await store.put(sub);
+    await store.put(sub, { limit: Infinity });
     expect(await store.get(sub.id)).toEqual(sub);
     expect(await store.get("sub_missing")).toBeNull();
   });
 
   it("expires the record with the subscription", async () => {
     const sub = makeSub({ expiresAt: Date.now() + 30_000 });
-    await store.put(sub);
+    await store.put(sub, { limit: Infinity });
     const pttl = await redis.pttl(subKey(sub.id));
     expect(pttl).toBeGreaterThan(0);
     expect(pttl).toBeLessThanOrEqual(30_000);
@@ -53,7 +53,7 @@ describe.skipIf(!hasRedisCreds)("RedisSubscriptionStore (real Redis)", () => {
     const one = makeSub({ event: "find", args: { repo: "a" } });
     const elsewhere = makeSub({ event: "other", args: {} });
     const expired = makeSub({ event: "find" });
-    for (const sub of [all, one, elsewhere]) await store.put(sub);
+    for (const sub of [all, one, elsewhere]) await store.put(sub, { limit: Infinity });
     // Written live, then left in the index past its expiry.
     await redis.zadd(indexKey("find"), { score: Date.now() - 1, member: expired.id });
 
@@ -68,7 +68,7 @@ describe.skipIf(!hasRedisCreds)("RedisSubscriptionStore (real Redis)", () => {
       makeSub({ event: "popular", args: {}, subscriber: `member-${i}` }),
     );
     for (let i = 0; i < subs.length; i += 100) {
-      await Promise.all(subs.slice(i, i + 100).map((sub) => store.put(sub)));
+      await Promise.all(subs.slice(i, i + 100).map((sub) => store.put(sub, { limit: Infinity })));
     }
     const found = await store.find("popular");
     expect(found).toHaveLength(1100);
@@ -77,21 +77,54 @@ describe.skipIf(!hasRedisCreds)("RedisSubscriptionStore (real Redis)", () => {
 
   it("refreshing replaces the record and keeps one index entry", async () => {
     const sub = makeSub({ event: "refresh", expiresAt: Date.now() + 10_000 });
-    await store.put(sub);
-    await store.put({ ...sub, expiresAt: Date.now() + 50_000 });
+    await store.put(sub, { limit: Infinity });
+    await store.put({ ...sub, expiresAt: Date.now() + 50_000 }, { limit: Infinity });
     expect((await store.get(sub.id))?.expiresAt).toBeGreaterThan(Date.now() + 40_000);
     expect(await store.find("refresh")).toHaveLength(1);
   });
 
   it("prunes expired index entries on the next write", async () => {
     await redis.zadd(indexKey("prune"), { score: Date.now() - 1000, member: "sub_old" });
-    await store.put(makeSub({ event: "prune" }));
+    await store.put(makeSub({ event: "prune" }), { limit: Infinity });
     expect(await redis.zscore(indexKey("prune"), "sub_old")).toBeNull();
   });
 
+  it("refuses a new subscription past the subscriber's limit, atomically, but allows a refresh", async () => {
+    const subscriber = `limited-${Date.now()}`;
+    const [a, b, c] = [1, 2, 3].map((i) => makeSub({ event: `limit-${i}`, subscriber }));
+    expect(await store.put(a!, { limit: 2 })).toBe(true);
+    expect(await store.put(b!, { limit: 2 })).toBe(true);
+    expect(await store.count(subscriber)).toBe(2);
+    expect(await store.put(c!, { limit: 2 })).toBe(false);
+    expect(await store.get(c!.id)).toBeNull();
+    expect(await store.find("limit-3")).toEqual([]);
+    // Replacing one of theirs is not a new subscription.
+    expect(await store.put({ ...a!, expiresAt: Date.now() + 90_000 }, { limit: 2 })).toBe(true);
+    // Concurrent puts can't overshoot: the check and the write are one script.
+    const racers = Array.from({ length: 5 }, (_, i) =>
+      makeSub({ event: `race-${i}`, subscriber: `${subscriber}-race` }),
+    );
+    const results = await Promise.all(racers.map((sub) => store.put(sub, { limit: 2 })));
+    expect(results.filter(Boolean)).toHaveLength(2);
+    expect(await store.count(`${subscriber}-race`)).toBe(2);
+  });
+
+  it("frees a slot on delete and when a subscription expires", async () => {
+    const subscriber = `freed-${Date.now()}`;
+    const a = makeSub({ event: "freed", subscriber });
+    await store.put(a, { limit: 1 });
+    await store.delete(a);
+    expect(await store.count(subscriber)).toBe(0);
+    const b = makeSub({ event: "freed", subscriber, expiresAt: Date.now() + 1_500 });
+    expect(await store.put(b, { limit: 1 })).toBe(true);
+    await new Promise((resolve) => setTimeout(resolve, 1_700));
+    expect(await store.count(subscriber)).toBe(0);
+    expect(await store.put(makeSub({ event: "freed", subscriber }), { limit: 1 })).toBe(true);
+  });
+
   it("deletes the record and its index entry", async () => {
     const sub = makeSub({ event: "delete" });
-    await store.put(sub);
+    await store.put(sub, { limit: Infinity });
     await store.delete(sub);
     expect(await store.get(sub.id)).toBeNull();
     expect(await store.find("delete")).toEqual([]);
@@ -171,7 +204,7 @@ describe("QStashDelivery", () => {
     expect(sent).toEqual([]);
   });
 
-  it("publishes one deduplicated message per subscription, in batches of 100", async () => {
+  it("publishes one message per subscription, in batches of 100", async () => {
     const batches: Record<string, unknown>[][] = [];
     const qstash = {
       batchJSON: async (messages: Record<string, unknown>[]) => {
@@ -187,7 +220,7 @@ describe("QStashDelivery", () => {
       url: URL,
       body: jobs[0],
       retries: 3,
-      deduplicationId: "evt_1_sub_0",
     });
+    expect(batches[0]?.[0]).not.toHaveProperty("deduplicationId");
   });
 });