Test your integration

import { McpServer } from "@modelcontextprotocol/server"
import { z } from "zod"
import { Wakeshift } from "@wakeshift/sdk"

export const wake = new Wakeshift()
export const crawlFinished = wake.defineEvent("crawl.finished", {
  description: "A crawl finished",
  args: z.object({ job: z.string() }),
  payload: z.object({ pages: z.number() }),
  text: (p, a) => `Crawl ${a.job} finished: ${p.pages} pages`,
})

export function makeServer() {
  const server = new McpServer({ name: "crawler", version: "1.0.0" })
  const attached = wake.attach(server)
  attached.registerTool(
    "watch_crawl",
    { description: "Watch a crawl.", inputSchema: z.object({ job: z.string() }), subscribes: [crawlFinished] },
    async ({ job }, _ctx, call) => {
      await call.subscribe(crawlFinished, { job })
      return `Watching crawl ${job}.`
    },
  )
  return server
}

import { test } from "node:test"
import assert from "node:assert/strict"
import { spawn } from "node:child_process"
import { createInterface } from "node:readline"
import { Client, InMemoryTransport } from "@modelcontextprotocol/client"
import { makeServer, crawlFinished } from "./server.js"

async function connectAs(name: string) {
  const [serverSide, clientSide] = InMemoryTransport.createLinkedPair()
  await makeServer().connect(serverSide)
  const client = new Client({ name, version: "1.0.0" })
  await client.connect(clientSide)
  return client
}
const text = (result: any) => result.content.map((c: any) => c.text ?? "").join("\n")

test("Claude Code is woken through its Monitor", async () => {
  const client = await connectAs("claude-code")
  const result = await client.callTool({ name: "watch_crawl", arguments: { job: "j1" } })
  // The result asks Claude Code to run a command in its Monitor tool.
  const command = text(result).match(/with this command: (u='.*?; done) \./s)?.[1]
  assert.ok(command, text(result))
  // Run it as Monitor does: each line it prints is one JSON message.
  const monitor = spawn("sh", ["-c", command], { detached: true, stdio: ["ignore", "pipe", "ignore"] })
  const lines = createInterface({ input: monitor.stdout })
  try {
    await crawlFinished.emit({ job: "j1" }, { pages: 12 })
    for await (const line of lines) {
      const message = JSON.parse(line)
      if (message.type !== "event") continue
      assert.equal(message.subject, "Crawl j1 finished: 12 pages")
      assert.deepEqual(message.data, { pages: 12 })
      break
    }
  } finally {
    process.kill(-monitor.pid!, "SIGKILL") // the command and its curl
    await client.close()
  }
})

test("two calls in one conversation, deduped by event id", async () => {
  const client = await connectAs("claude-code")
  const first = text(await client.callTool({ name: "watch_crawl", arguments: { job: "j4" } }))
  const sessionKey = first.match(/session_key="([^"]+)"/)?.[1]
  assert.ok(sessionKey, first)
  const second = text(await client.callTool({ name: "watch_crawl", arguments: { job: "j5", session_key: sessionKey } }))

  // Run every Monitor command the results gave, as Claude Code would.
  const commands = [first, second].map((t) => t.match(/with this command: (u='.*?; done) \./s)?.[1]).filter((c): c is string => !!c)
  const monitors = commands.map((c) => spawn("sh", ["-c", c], { detached: true, stdio: ["ignore", "pipe", "ignore"] }))
  const seen = new Map<string, string>() // event id -> subject
  const both = new Promise<void>((resolve) => {
    for (const m of monitors) {
      createInterface({ input: m.stdout! }).on("line", (line) => {
        const message = JSON.parse(line)
        if (message.type === "event" && !seen.has(message.id)) seen.set(message.id, message.subject)
        if (seen.size === 2) resolve()
      })
    }
  })
  try {
    await crawlFinished.emit({ job: "j4" }, { pages: 1 })
    await crawlFinished.emit({ job: "j5" }, { pages: 2 })
    await both
    assert.deepEqual([...seen.values()].sort(), ["Crawl j4 finished: 1 pages", "Crawl j5 finished: 2 pages"])
  } finally {
    for (const m of monitors) process.kill(-m.pid!, "SIGKILL")
    await client.close()
  }
})

// A crawl your API reports only by status: the test sets what the status call returns.
export const crawlStatus = new Map<string, number>() // job -> pages, once done
export const crawlPolled = wake.defineEvent("crawl.polled", {
  description: "A crawl finished",
  args: z.object({ job: z.string() }),
  payload: z.object({ pages: z.number() }),
  text: (p, a) => `Crawl ${a.job} finished: ${p.pages} pages`,
  onSubscribe: ({ args }) => watchPolled(args.job),
})

function watchPolled(job: string) {
  crawlPolled.watch(
    { job },
    async () => {
      const pages = crawlStatus.get(job) // your API's status call
      return pages === undefined ? undefined : { pages } // undefined: keep waiting
    },
    { every: "1s", timeout: "2m" },
  )
}

export function makeWatchServer() {
  const server = new McpServer({ name: "crawler", version: "1.0.0" })
  wake.attach(server).registerTool(
    "watch_polled_crawl",
    { description: "Watch a crawl.", inputSchema: z.object({ job: z.string() }), subscribes: [crawlPolled] },
    async ({ job }, _ctx, call) => {
      await call.subscribe(crawlPolled, { job }, { once: true })
      return `Watching crawl ${job}.`
    },
  )
  return server
}

import { randomUUID } from "node:crypto"
import { crawlStatus, makeWatchServer, wake } from "./server.js"

test("a watch wakes Claude Code once the status says done", async () => {
  const [serverSide, clientSide] = InMemoryTransport.createLinkedPair()
  await makeWatchServer().connect(serverSide)
  const client = new Client({ name: "claude-code", version: "1.0.0" })
  await client.connect(clientSide)
  const job = `j-${randomUUID()}` // fresh each run
  const result = await client.callTool({ name: "watch_polled_crawl", arguments: { job } })
  const command = text(result).match(/with this command: (u='.*?; done) \./s)?.[1]
  assert.ok(command, text(result))
  const monitor = spawn("sh", ["-c", command], { detached: true, stdio: ["ignore", "pipe", "ignore"] })
  try {
    const event = new Promise<any>((resolve) =>
      createInterface({ input: monitor.stdout! }).on("line", (line) => {
        const message = JSON.parse(line)
        if (message.type === "event") resolve(message)
      }),
    )
    crawlStatus.set(job, 5) // the crawl is done: the watch's next check emits
    assert.equal((await event).subject, `Crawl ${job} finished: 5 pages`)
  } finally {
    process.kill(-monitor.pid!, "SIGKILL")
    await client.close()
    await wake.close() // stops the watches still running
  }
})

import { test } from "node:test"
import assert from "node:assert/strict"
import { once } from "node:events"
import { createServer } from "node:http"
import type { AddressInfo } from "node:net"
import { createMcpHandler } from "@modelcontextprotocol/server"
import { toNodeHandler } from "@modelcontextprotocol/node"
import { Client, StreamableHTTPClientTransport } from "@modelcontextprotocol/client"
import { makeServer } from "./server.js"

const text = (result: any) => result.content.map((c: any) => c.text ?? "").join("\n")

test("Claude Code over HTTP, on a stateless server", async () => {
  const http = createServer(toNodeHandler(createMcpHandler(makeServer))).listen(0, "127.0.0.1")
  await once(http, "listening")
  const url = new URL(`http://127.0.0.1:${(http.address() as AddressInfo).port}/mcp`)
  const client = new Client({ name: "claude-code", version: "2.1.0" })
  const transport = new StreamableHTTPClientTransport(url, {
    requestInit: { headers: { "User-Agent": "claude-code/2.1.0 (external, cli)" } },
  })
  await client.connect(transport)
  try {
    const result = await client.callTool({ name: "watch_crawl", arguments: { job: "j7" } })
    assert.match(text(result), /with this command: u='/)
  } finally {
    await client.close()
    http.close()
  }
})

import { test } from "node:test"
import assert from "node:assert/strict"
import { Client } from "@modelcontextprotocol/client"
import { StdioClientTransport } from "@modelcontextprotocol/client/stdio"

const text = (result: any) => result.content.map((c: any) => c.text ?? "").join("\n")

test("over stdio, started as Claude Code starts it", async () => {
  const transport = new StdioClientTransport({
    command: "node",
    args: ["dist/stdio.js"], // your server's entry point, as users run it
    env: { ...process.env } as Record<string, string>, // WAKESHIFT_KEY included
    stderr: "inherit",
  })
  const client = new Client({ name: "claude-code", version: "1.0.0" })
  await client.connect(transport)
  try {
    const result = await client.callTool({ name: "watch_crawl", arguments: { job: "j6" } })
    assert.match(text(result), /with this command: u='/)
  } finally {
    await client.close()
  }
})

test("Codex is woken through its connector", async () => {
  const client = await connectAs("codex-mcp-client")
  const thread = "thread-test-1"
  const result = await client.callTool({
    name: "watch_crawl",
    arguments: { job: "j2" },
    _meta: { "x-codex-turn-metadata": { thread_id: thread } }, // the Codex thread
  })
  // The connector's beacon: a signed token whose claims say where to redeem it.
  const beacon = result._meta?.["wakeshift/connector"] as string
  const { redeem, ticket } = JSON.parse(Buffer.from(beacon.split(".")[1], "base64url").toString())
  const redeemed = await fetch(redeem, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ ticket, thread }) })
  const { url } = (await redeemed.json()) as { url: string }
  const socket = new WebSocket(url)
  await new Promise((resolve, reject) => ((socket.onopen = resolve), (socket.onerror = reject)))
  try {
    const frame = new Promise<any>((resolve) => {
      socket.onmessage = (m) => {
        const f = JSON.parse(String(m.data))
        if (f.type !== "event") return
        socket.send(JSON.stringify({ type: "ack", delivery_id: f.delivery_id }))
        resolve(f)
      }
    })
    await crawlFinished.emit({ job: "j2" }, { pages: 3 })
    const f = await frame
    assert.equal(f.event.subject, "Crawl j2 finished: 3 pages")
    assert.deepEqual(f.event.data, { pages: 3 })
  } finally {
    socket.close()
    await client.close()
  }
})

import { test } from "node:test"
import assert from "node:assert/strict"
import { createServer } from "node:http"
import { createMcpHandler } from "@modelcontextprotocol/server"
import { toNodeHandler } from "@modelcontextprotocol/node"

const META = {
  "io.modelcontextprotocol/protocolVersion": "2026-07-28",
  "io.modelcontextprotocol/clientInfo": { name: "native-test", version: "1.0.0" },
  "io.modelcontextprotocol/clientCapabilities": { extensions: { "io.modelcontextprotocol/events": {} } },
}

test("a native client polls an open event, and is refused a tools-only one", async (t) => {
  // A server for each request, as the quickstart serves it; identify reads the test's X-User header.
  const handler = createMcpHandler(() => makeNativeServer({ identify: (ctx) => ({ user: ctx.http?.req?.headers.get("x-user") ?? undefined }) }))
  const http = createServer(toNodeHandler(handler))
  await new Promise<void>((resolve) => http.listen(0, "127.0.0.1", resolve))
  t.after(() => (http.closeAllConnections(), http.close()))
  const url = `http://127.0.0.1:${(http.address() as { port: number }).port}/mcp`
  const rpc = async (method: string, params: Record<string, unknown> = {}) => {
    const res = await fetch(url, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Accept: "application/json, text/event-stream",
        "MCP-Protocol-Version": "2026-07-28",
        "Mcp-Method": method,
        ...(typeof params.name === "string" ? { "Mcp-Name": params.name } : {}),
        "X-User": "ada",
      },
      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params: { ...params, _meta: META } }),
    })
    const body = await res.text()
    return JSON.parse(body.startsWith("{") ? body : body.split("\n").find((l) => l.startsWith("data: "))!.slice(6))
  }

  const poll = { name: "site.updated", arguments: { site: "example.com" } } // a public event
  let { cursor } = (await rpc("events/poll", { ...poll, cursor: null })).result
  await siteUpdated.emit({ site: "example.com" }, { pages: 7 })
  let events: any[] = []
  while (events.length === 0) {
    const r = (await rpc("events/poll", { ...poll, cursor })).result
    ;({ events, cursor } = r)
    if (events.length === 0) await new Promise((resolve) => setTimeout(resolve, 500))
  }
  assert.deepEqual(events[0].data, { pages: 7 })

  const refused = await rpc("events/poll", { name: "crawl.finished", arguments: { job: "j1" }, cursor: null })
  assert.equal(refused.error.code, -32024)
})

const CLAUDE_CODE_META = {
  "io.modelcontextprotocol/protocolVersion": "2026-07-28",
  "io.modelcontextprotocol/clientInfo": { name: "claude-code", version: "2.1.0" },
  "io.modelcontextprotocol/clientCapabilities": {},
}

import { spawn } from "node:child_process"
import { createInterface } from "node:readline"

test("an MCP Events client over stdio lists and polls", async () => {
  const child = spawn("node", ["dist/stdio.js"], { env: process.env, stdio: ["pipe", "pipe", "inherit"] })
  const answers = new Map<number, (message: any) => void>()
  createInterface({ input: child.stdout! }).on("line", (line) => {
    const message = JSON.parse(line)
    answers.get(message.id)?.(message)
  })
  let id = 0
  const rpc = (method: string, params: Record<string, unknown> = {}) =>
    new Promise<any>((resolve) => {
      answers.set(++id, resolve)
      const _meta = {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": { name: "native-test", version: "1.0.0" },
        "io.modelcontextprotocol/clientCapabilities": { extensions: { "io.modelcontextprotocol/events": {} } },
      }
      child.stdin!.write(JSON.stringify({ jsonrpc: "2.0", id, method, params: { ...params, _meta } }) + "\n")
    })
  try {
    const listed = await rpc("events/list")
    assert.ok(listed.result.events.length > 0, JSON.stringify(listed))
    const first = await rpc("events/poll", { name: "site.updated", arguments: { site: "example.com" }, cursor: null })
    assert.ok(first.result.cursor, JSON.stringify(first))
  } finally {
    child.kill()
  }
})

import { test } from "node:test"
import assert from "node:assert/strict"
import { randomBytes } from "node:crypto"
import { createServer } from "node:http"
import { createMcpHandler } from "@modelcontextprotocol/server"
import { toNodeHandler } from "@modelcontextprotocol/node"
import { crawlFinished, makeServer } from "./server.js"

const WEBHOOK_URL = process.env.WEBHOOK_URL // a public HTTPS address that forwards to port 8788 here

// ChatGPT's MCP client is named openai-mcp.
const CHATGPT_META = {
  "io.modelcontextprotocol/protocolVersion": "2026-07-28",
  "io.modelcontextprotocol/clientInfo": { name: "openai-mcp", version: "1.0.0" },
  "io.modelcontextprotocol/clientCapabilities": {},
}

test("ChatGPT subscribes with the voucher its tool call handed it", { skip: !WEBHOOK_URL && "set WEBHOOK_URL" }, async (t) => {
  // The MCP server, built for each request as the quickstart serves it, with no sign-in.
  const http = createServer(toNodeHandler(createMcpHandler(() => makeServer())))
  await new Promise<void>((resolve) => http.listen(0, "127.0.0.1", resolve))
  t.after(() => (http.closeAllConnections(), http.close()))
  const url = `http://127.0.0.1:${(http.address() as { port: number }).port}/mcp`
  const rpc = async (method: string, params: Record<string, unknown> = {}) => {
    const res = await fetch(url, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Accept: "application/json, text/event-stream",
        "MCP-Protocol-Version": "2026-07-28",
        "Mcp-Method": method,
        ...(typeof params.name === "string" ? { "Mcp-Name": params.name } : {}),
      },
      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params: { ...params, _meta: CHATGPT_META } }),
    })
    const body = await res.text()
    return JSON.parse(body.startsWith("{") ? body : body.split("\n").find((l) => l.startsWith("data: "))!.slice(6))
  }

  // ChatGPT's webhook: answers the verification challenge, then takes occurrences.
  let delivered!: (occurrence: any) => void
  const occurrence = new Promise<any>((resolve) => (delivered = resolve))
  const hook = createServer(async (req, res) => {
    let raw = ""
    for await (const chunk of req) raw += chunk
    const body = JSON.parse(raw)
    res.setHeader("Content-Type", "application/json")
    if (body.type === "verification") return void res.end(JSON.stringify({ challenge: body.challenge }))
    res.end("{}")
    delivered(body)
  })
  await new Promise<void>((resolve) => hook.listen(8788, resolve))
  t.after(() => (hook.closeAllConnections(), hook.close()))

  // 1. ChatGPT calls the tool.
  const called = await rpc("tools/call", { name: "watch_crawl", arguments: { job: "j9" } })
  // 2. The voucher is in the result's text.
  const text = called.result.content.map((c: { text?: string }) => c.text ?? "").join("\n")
  const voucher = text.match(/wsv_[a-z2-7]{32}/)?.[0]
  assert.ok(voucher, text)

  // 3. It subscribes with the voucher, as the event argument wakeshift_voucher, and its webhook.
  const delivery = { mode: "webhook", url: WEBHOOK_URL, secret: "whsec_" + randomBytes(32).toString("base64") }
  const subscribed = await rpc("events/subscribe", { name: "crawl.finished", arguments: { job: "j9", wakeshift_voucher: voucher }, delivery })
  assert.ok(subscribed.result?.id, JSON.stringify(subscribed))

  // 4. The emit arrives at the webhook.
  await crawlFinished.emit({ job: "j9" }, { pages: 4 })
  const timeout = new Promise<never>((_, reject) => setTimeout(() => reject(new Error("no delivery in 30s")), 30_000).unref())
  const got = await Promise.race([occurrence, timeout])
  assert.equal(got.name, "crawl.finished")
  assert.deepEqual(got.data, { pages: 4 })

  // Without a voucher, ChatGPT is refused with its forbidden error, -32012.
  const refused = await rpc("events/subscribe", { name: "crawl.finished", arguments: { job: "j10" }, delivery })
  assert.equal(refused.error.code, -32012)
})

# server.py
from mcp.server.mcpserver import MCPServer
from pydantic import BaseModel

from wakeshift import ToolCall, Wakeshift

wake = Wakeshift()  # reads WAKESHIFT_KEY


class Job(BaseModel):
    job: str


class Crawl(BaseModel):
    pages: int


crawl_finished = wake.define_event(
    "crawl.finished",
    description="A crawl finished",
    args=Job,
    payload=Crawl,
    text=lambda p, a: f"Crawl {a.job} finished: {p.pages} pages",
)


def make_server(identify=None) -> MCPServer:
    server = MCPServer("crawler")
    attached = wake.attach(server, identify=identify)

    @attached.tool(subscribes=[crawl_finished])
    async def watch_crawl(job: str, call: ToolCall) -> str:
        """Watch a crawl."""
        await call.subscribe(crawl_finished, {"job": job})
        return f"Watching crawl {job}."

    return server

# test_wakes.py
import asyncio
import base64
import contextlib
import json
import os
import re
import signal

import httpx2
import websockets
from mcp import Client
from mcp.types import Implementation

from server import crawl_finished, make_server


def connect_as(name: str) -> Client:
    return Client(make_server(), client_info=Implementation(name=name, version="1.0.0"))


def text(result) -> str:
    return "\n".join(getattr(c, "text", "") for c in result.content)


async def test_claude_code_is_woken_through_its_monitor():
    async with connect_as("claude-code") as client:
        result = await client.call_tool("watch_crawl", {"job": "j1"})
    # The result asks Claude Code to run a command in its Monitor tool.
    match = re.search(r"with this command: (u='.*?; done) \.", text(result), re.S)
    assert match, text(result)
    # Run it as Monitor does: each line it prints is one JSON message.
    monitor = await asyncio.create_subprocess_exec(
        "sh", "-c", match[1], stdout=asyncio.subprocess.PIPE, start_new_session=True
    )
    try:
        await crawl_finished.emit({"job": "j1"}, {"pages": 12})
        async with asyncio.timeout(30):
            while (message := json.loads(await monitor.stdout.readline())).get("type") != "event":
                pass
        assert message["subject"] == "Crawl j1 finished: 12 pages"
        assert message["data"] == {"pages": 12}
    finally:
        # The command and its curl. On macOS, killpg raises PermissionError once the
        # group's processes have exited (a once subscription ends the command itself).
        with contextlib.suppress(ProcessLookupError, PermissionError):
            os.killpg(monitor.pid, signal.SIGKILL)
        await monitor.wait()


async def test_codex_is_woken_through_its_connector():
    thread = "thread-test-1"
    async with connect_as("codex-mcp-client") as client:
        result = await client.call_tool(
            "watch_crawl", {"job": "j2"}, meta={"x-codex-turn-metadata": {"thread_id": thread}}
        )
    # The connector's beacon: a signed token whose claims say where to redeem it.
    beacon = result.meta["wakeshift/connector"]
    claims = json.loads(base64.urlsafe_b64decode(beacon.split(".")[1] + "=="))
    async with httpx2.AsyncClient() as http:
        redeemed = await http.post(claims["redeem"], json={"ticket": claims["ticket"], "thread": thread})
    async with websockets.connect(redeemed.json()["url"]) as socket:
        await crawl_finished.emit({"job": "j2"}, {"pages": 3})
        async with asyncio.timeout(30):
            while (frame := json.loads(await socket.recv())).get("type") != "event":
                pass
        await socket.send(json.dumps({"type": "ack", "delivery_id": frame["delivery_id"]}))
        assert frame["event"]["subject"] == "Crawl j2 finished: 3 pages"
        assert frame["event"]["data"] == {"pages": 3}

# test_chatgpt.py
import asyncio
import base64
import json
import os
import queue
import re
import secrets
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import httpx2
import pytest

from server import crawl_finished, make_server

WEBHOOK_URL = os.environ.get("WEBHOOK_URL")  # a public HTTPS address that forwards to port 8788 here
# ChatGPT's MCP client is named openai-mcp.
CHATGPT_META = {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientInfo": {"name": "openai-mcp", "version": "1.0.0"},
    "io.modelcontextprotocol/clientCapabilities": {},
}
occurrences: queue.Queue = queue.Queue()


class Webhook(BaseHTTPRequestHandler):
    """ChatGPT's webhook: answers the verification challenge, then takes occurrences."""

    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        answer = {"challenge": body["challenge"]} if body.get("type") == "verification" else {}
        out = json.dumps(answer).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(out)))
        self.end_headers()
        self.wfile.write(out)
        if not answer:
            occurrences.put(body)


@pytest.mark.skipif(not WEBHOOK_URL, reason="set WEBHOOK_URL")
async def test_chatgpt_subscribes_with_the_voucher():
    hook = ThreadingHTTPServer(("127.0.0.1", 8788), Webhook)
    threading.Thread(target=hook.serve_forever, daemon=True).start()
    app = make_server().streamable_http_app(stateless_http=True, json_response=True)  # no sign-in
    try:
        async with app.router.lifespan_context(app):
            async with httpx2.AsyncClient(
                transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1:8000"
            ) as http:

                async def rpc(method: str, params: dict) -> dict:
                    headers = {
                        "Accept": "application/json, text/event-stream",
                        "MCP-Protocol-Version": "2026-07-28",
                        "Mcp-Method": method,
                        "Mcp-Name": params["name"],
                    }
                    body = {"jsonrpc": "2.0", "id": 1, "method": method, "params": {**params, "_meta": CHATGPT_META}}
                    return (await http.post("/mcp", json=body, headers=headers)).json()

                # 1. ChatGPT calls the tool.
                called = await rpc("tools/call", {"name": "watch_crawl", "arguments": {"job": "j9"}})
                # 2. The voucher is in the result's text.
                text = "\n".join(c.get("text", "") for c in called["result"]["content"])
                voucher = re.search(r"wsv_[a-z2-7]{32}", text)
                assert voucher, text

                # 3. It subscribes with the voucher, as the event argument wakeshift_voucher, and its webhook.
                secret = "whsec_" + base64.b64encode(secrets.token_bytes(32)).decode()
                delivery = {"mode": "webhook", "url": WEBHOOK_URL, "secret": secret}
                subscribed = await rpc(
                    "events/subscribe",
                    {"name": "crawl.finished", "arguments": {"job": "j9", "wakeshift_voucher": voucher[0]}, "delivery": delivery},
                )
                assert subscribed.get("result", {}).get("id"), subscribed

                # 4. The emit arrives at the webhook.
                await crawl_finished.emit({"job": "j9"}, {"pages": 4})
                got = await asyncio.to_thread(occurrences.get, timeout=30)
                assert got["name"] == "crawl.finished"
                assert got["data"] == {"pages": 4}

                # Without a voucher, ChatGPT is refused with its forbidden error, -32012.
                refused = await rpc("events/subscribe", {"name": "crawl.finished", "arguments": {"job": "j10"}, "delivery": delivery})
                assert refused["error"]["code"] == -32012
    finally:
        hook.shutdown()
        hook.server_close()

# test_fastmcp_wakes.py
import mcp_types  # FastMCP 4; on FastMCP 2 and 3: from mcp import types as mcp_types
from fastmcp import Client

from server import crawl_finished, mcp  # your FastMCP server, attached


def connect_as(name: str) -> Client:
    return Client(mcp, client_info=mcp_types.Implementation(name=name, version="1.0.0"))


async def test_claude_code_on_fastmcp():
    async with connect_as("claude-code") as client:
        result = await client.call_tool("watch_crawl", {"job": "j1"})
    text = "\n".join(getattr(c, "text", "") for c in result.content)
    assert "with this command: u='" in text  # then run it and emit, as above


async def test_codex_on_fastmcp():
    async with connect_as("codex-mcp-client") as client:
        result = await client.call_tool(
            "watch_crawl", {"job": "j2"}, meta={"x-codex-turn-metadata": {"thread_id": "thread-test-1"}}
        )
    assert result.meta["wakeshift/connector"]  # the beacon: redeem it, as above

// server.go
package crawler

import (
	"context"
	"fmt"
	"log"

	"github.com/modelcontextprotocol/go-sdk/mcp"
	"github.com/wakeshift/wakeshift-go"
)

type crawlArgs struct {
	Job string `json:"job"`
}

type crawl struct {
	Pages int `json:"pages"`
}

var (
	wake          = must(wakeshift.New()) // reads WAKESHIFT_KEY
	crawlFinished = must(wakeshift.DefineEvent(wake, "crawl.finished", wakeshift.EventDef[crawlArgs, crawl]{
		Description: "A crawl finished",
		Text:        func(p crawl, a crawlArgs) string { return fmt.Sprintf("Crawl %s finished: %d pages", a.Job, p.Pages) },
	}))
)

func newServer() *mcp.Server {
	srv := mcp.NewServer(&mcp.Implementation{Name: "crawler", Version: "1.0.0"}, nil)
	attached := wake.Attach(srv, wakeshift.AttachOptions{})
	wakeshift.AddTool(attached, &mcp.Tool{Name: "watch_crawl", Description: "Watch a crawl."}, watchCrawl, wakeshift.Subscribes(crawlFinished))
	return srv
}

func watchCrawl(ctx context.Context, call *wakeshift.ToolCall, in crawlArgs) (*mcp.CallToolResult, any, error) {
	if _, err := crawlFinished.Subscribe(ctx, call, crawlArgs{Job: in.Job}); err != nil {
		return nil, nil, err
	}
	return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: "Watching crawl " + in.Job + "."}}}, nil, nil
}

func must[T any](v T, err error) T {
	if err != nil {
		log.Fatal(err)
	}
	return v
}

// wakes_test.go
package crawler

import (
	"bufio"
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"net/http"
	"os/exec"
	"regexp"
	"strings"
	"syscall"
	"testing"
	"time"

	"github.com/coder/websocket"
	"github.com/coder/websocket/wsjson"
	"github.com/modelcontextprotocol/go-sdk/mcp"
)

// connectAs connects a client with the given name to a fresh server, in memory.
func connectAs(t *testing.T, name string) *mcp.ClientSession {
	t.Helper()
	serverSide, clientSide := mcp.NewInMemoryTransports()
	if _, err := newServer().Connect(t.Context(), serverSide, nil); err != nil {
		t.Fatal(err)
	}
	client := mcp.NewClient(&mcp.Implementation{Name: name, Version: "1.0.0"}, nil)
	session, err := client.Connect(t.Context(), clientSide, nil)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { session.Close() })
	return session
}

func text(res *mcp.CallToolResult) string {
	var parts []string
	for _, c := range res.Content {
		if tc, ok := c.(*mcp.TextContent); ok {
			parts = append(parts, tc.Text)
		}
	}
	return strings.Join(parts, "\n")
}

func TestClaudeCodeWokenThroughMonitor(t *testing.T) {
	session := connectAs(t, "claude-code")
	res, err := session.CallTool(t.Context(), &mcp.CallToolParams{Name: "watch_crawl", Arguments: map[string]any{"job": "j1"}})
	if err != nil {
		t.Fatal(err)
	}
	// The result asks Claude Code to run a command in its Monitor tool.
	m := regexp.MustCompile(`(?s)with this command: (u='.*?; done) \.`).FindStringSubmatch(text(res))
	if m == nil {
		t.Fatalf("no Monitor command in: %s", text(res))
	}
	// Run it as Monitor does: each line it prints is one JSON message.
	monitor := exec.Command("sh", "-c", m[1])
	monitor.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	stdout, _ := monitor.StdoutPipe()
	if err := monitor.Start(); err != nil {
		t.Fatal(err)
	}
	defer syscall.Kill(-monitor.Process.Pid, syscall.SIGKILL) // the command and its curl

	if _, err := crawlFinished.Emit(t.Context(), crawlArgs{Job: "j1"}, crawl{Pages: 12}); err != nil {
		t.Fatal(err)
	}
	lines := bufio.NewScanner(stdout)
	for lines.Scan() {
		var msg struct {
			Type    string `json:"type"`
			Subject string `json:"subject"`
			Data    crawl  `json:"data"`
		}
		if json.Unmarshal(lines.Bytes(), &msg) != nil || msg.Type != "event" {
			continue
		}
		if msg.Subject != "Crawl j1 finished: 12 pages" || msg.Data.Pages != 12 {
			t.Fatalf("event: %+v", msg)
		}
		return
	}
	t.Fatal("the Monitor command ended without the event")
}

func TestCodexWokenThroughConnector(t *testing.T) {
	session := connectAs(t, "codex-mcp-client")
	const thread = "thread-test-1"
	res, err := session.CallTool(t.Context(), &mcp.CallToolParams{
		Name:      "watch_crawl",
		Arguments: map[string]any{"job": "j2"},
		Meta:      mcp.Meta{"x-codex-turn-metadata": map[string]any{"thread_id": thread}}, // the Codex thread
	})
	if err != nil {
		t.Fatal(err)
	}
	// The connector's beacon: a signed token whose claims say where to redeem it.
	beacon, _ := res.Meta["wakeshift/connector"].(string)
	parts := strings.Split(beacon, ".")
	if len(parts) != 3 {
		t.Fatalf("no beacon in _meta: %v", res.Meta)
	}
	claims, _ := base64.RawURLEncoding.DecodeString(parts[1])
	var c struct{ Redeem, Ticket string }
	if err := json.Unmarshal(claims, &c); err != nil {
		t.Fatal(err)
	}
	body, _ := json.Marshal(map[string]string{"ticket": c.Ticket, "thread": thread})
	resp, err := http.Post(c.Redeem, "application/json", bytes.NewReader(body))
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	var redeemed struct{ URL string }
	if err := json.NewDecoder(resp.Body).Decode(&redeemed); err != nil || redeemed.URL == "" {
		t.Fatalf("redeem: %d %v", resp.StatusCode, err)
	}

	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(ctx, redeemed.URL, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.CloseNow()

	if _, err := crawlFinished.Emit(ctx, crawlArgs{Job: "j2"}, crawl{Pages: 3}); err != nil {
		t.Fatal(err)
	}
	for {
		var frame struct {
			Type       string `json:"type"`
			DeliveryID string `json:"delivery_id"`
			Event      struct {
				Subject string `json:"subject"`
				Data    crawl  `json:"data"`
			} `json:"event"`
		}
		if err := wsjson.Read(ctx, conn, &frame); err != nil {
			t.Fatal(err)
		}
		if frame.Type != "event" {
			continue
		}
		wsjson.Write(ctx, conn, map[string]string{"type": "ack", "delivery_id": frame.DeliveryID})
		if frame.Event.Subject != "Crawl j2 finished: 3 pages" || frame.Event.Data.Pages != 3 {
			t.Fatalf("event: %+v", frame.Event)
		}
		return
	}
}

// chatgpt_test.go
package crawler

import (
	"bytes"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/modelcontextprotocol/go-sdk/mcp"
	"github.com/wakeshift/wakeshift-go"
)

// chatGPTMeta is ChatGPT's: its MCP client is named openai-mcp.
var chatGPTMeta = map[string]any{
	"io.modelcontextprotocol/protocolVersion":    "2026-07-28",
	"io.modelcontextprotocol/clientInfo":         map[string]any{"name": "openai-mcp", "version": "1.0.0"},
	"io.modelcontextprotocol/clientCapabilities": map[string]any{},
}

type rpcReply struct {
	Result struct {
		ID      string `json:"id"`
		Content []struct {
			Text string `json:"text"`
		} `json:"content"`
	} `json:"result"`
	Error *struct {
		Code int `json:"code"`
	} `json:"error"`
}

func TestChatGPTSubscribesWithVoucher(t *testing.T) {
	webhookURL := os.Getenv("WEBHOOK_URL") // a public HTTPS address that forwards to port 8788 here
	if webhookURL == "" {
		t.Skip("set WEBHOOK_URL")
	}
	// The MCP server, built for each request, with no sign-in.
	opts := wakeshift.AttachOptions{}
	h := mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return newServer() }, &mcp.StreamableHTTPOptions{Stateless: true, JSONResponse: true})
	srv := httptest.NewServer(wake.Handler(h, opts))
	t.Cleanup(srv.Close)
	rpc := func(method, name string, params map[string]any) rpcReply {
		t.Helper()
		params["name"], params["_meta"] = name, chatGPTMeta
		body, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": 1, "method": method, "params": params})
		req, _ := http.NewRequest("POST", srv.URL, bytes.NewReader(body))
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Accept", "application/json, text/event-stream")
		req.Header.Set("MCP-Protocol-Version", "2026-07-28")
		req.Header.Set("Mcp-Method", method)
		req.Header.Set("Mcp-Name", name)
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		defer resp.Body.Close()
		var out rpcReply
		if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
			t.Fatal(err)
		}
		return out
	}

	// ChatGPT's webhook: answers the verification challenge, then takes occurrences.
	occurrences := make(chan map[string]any, 1)
	ln, err := net.Listen("tcp", "127.0.0.1:8788")
	if err != nil {
		t.Fatal(err)
	}
	hook := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var body map[string]any
		json.NewDecoder(r.Body).Decode(&body)
		w.Header().Set("Content-Type", "application/json")
		if body["type"] == "verification" {
			json.NewEncoder(w).Encode(map[string]any{"challenge": body["challenge"]})
			return
		}
		w.Write([]byte("{}"))
		select {
		case occurrences <- body:
		default:
		}
	})}
	go hook.Serve(ln)
	t.Cleanup(func() { hook.Close() })

	// 1. ChatGPT calls the tool.
	called := rpc("tools/call", "watch_crawl", map[string]any{"arguments": map[string]any{"job": "j9"}})
	// 2. The voucher is in the result's text.
	var parts []string
	for _, c := range called.Result.Content {
		parts = append(parts, c.Text)
	}
	text := strings.Join(parts, "\n")
	voucher := regexp.MustCompile(`wsv_[a-z2-7]{32}`).FindString(text)
	if voucher == "" {
		t.Fatalf("no voucher in: %s", text)
	}

	// 3. It subscribes with the voucher, as the event argument wakeshift_voucher, and its webhook.
	key := make([]byte, 32)
	rand.Read(key)
	delivery := map[string]any{"mode": "webhook", "url": webhookURL, "secret": "whsec_" + base64.StdEncoding.EncodeToString(key)}
	subscribed := rpc("events/subscribe", "crawl.finished", map[string]any{
		"arguments": map[string]any{"job": "j9", "wakeshift_voucher": voucher},
		"delivery":  delivery,
	})
	if subscribed.Error != nil || subscribed.Result.ID == "" {
		t.Fatalf("subscribe: %+v", subscribed)
	}

	// 4. The emit arrives at the webhook.
	if _, err := crawlFinished.Emit(t.Context(), crawlArgs{Job: "j9"}, crawl{Pages: 4}); err != nil {
		t.Fatal(err)
	}
	select {
	case got := <-occurrences:
		data, _ := got["data"].(map[string]any)
		if got["name"] != "crawl.finished" || data["pages"] != float64(4) {
			t.Fatalf("occurrence: %v", got)
		}
	case <-time.After(30 * time.Second):
		t.Fatal("no delivery in 30s")
	}

	// Without a voucher, ChatGPT is refused with its forbidden error, -32012.
	refused := rpc("events/subscribe", "crawl.finished", map[string]any{"arguments": map[string]any{"job": "j10"}, "delivery": delivery})
	if refused.Error == nil || refused.Error.Code != -32012 {
		t.Fatalf("without a voucher: %+v", refused)
	}
}