[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10},"actionData","errors","root",{"_159":160,"_161":162},"routes/_app.x.$handle.$name",{"_99":98,"_32":45,"_107":89,"_27":108,"_13":14,"_109":-5,"_11":110,"_111":112,"_113":114,"_115":116,"_117":118,"_20":-5,"_119":-5,"_120":-5,"_121":25,"_122":123,"_124":58,"_125":-5,"_126":-5,"_127":128},"routes/_app.x.$handle.$name.branch.$branch.version.$version.diff",{"_11":12,"_13":14,"_15":16,"_17":-5,"_18":-5},"project",{"_43":44,"_32":45,"_38":46,"_47":-5,"_48":25,"_49":50,"_51":50,"_52":53,"_54":-5,"_18":55,"_56":31,"_57":58,"_59":-5,"_60":-5,"_61":62,"_63":64,"_65":66,"_67":68,"_69":70,"_71":-5,"_72":73,"_74":73,"_75":76,"_77":76,"_78":76,"_79":73,"_80":73,"_81":82,"_83":84,"_85":86,"_87":25},"version",9,"changes",[19],"aiChatInfo","userId",{"_20":21,"_22":23,"_24":25,"_26":25,"_27":28,"_29":28},"file",{"_30":31,"_32":28,"_33":34,"_35":37,"_38":39,"_40":14,"_41":42,"_24":25},"previous",{"_30":31,"_32":28,"_33":34,"_35":36},"deleted",false,"deletedHere","path","main.ts","latestPath","fileId","ee1b71f8-8045-11f1-b556-1607ee4eb77e","name","fileType","http","content","//https://mcp-bridge.val.run/\n// bridge.ts — Val Town HTTP val. Public MCP relay over a SQLite job queue.\n//\n// Neither side can accept an inbound connection: the local machine is\n// behind NAT with no forwarded ports, and Val Town has no persistent\n// process to hold a socket open. So the queue is the protocol. MCP clients\n// POST to /mcp/:channel and block until a local agent — which only ever\n// makes outbound calls — claims the row and posts a result back.\n//\n// No tool registry lives here. Every method except the initialized\n// notification is forwarded verbatim; the local agent's backend is the\n// only source of truth for what tools exist.\n//\n// OAUTH LAYER (added): claude.ai's custom-connector flow requires the\n// target to behave as an OAuth 2.1 authorization server with PKCE and\n// Dynamic Client Registration — a bare static bearer token is not an\n// accepted auth mode for connectors added by URL. Since this bridge has\n// exactly one real credential (CLIENT_TOKEN) and exactly one consumer\n// per channel, the AS below is deliberately single-tenant: /register\n// always succeeds (client_id isn't a security boundary here — PKCE is),\n// /authorize is a one-field form gated by CLIENT_TOKEN itself (so the\n// interactive consent step doesn't introduce a second secret to manage),\n// and /token exchanges a PKCE-verified code for CLIENT_TOKEN itself as\n// the access_token. This means /mcp's existing bearer check needs no\n// changes — an OAuth-issued token IS the token it already accepts.\n\nimport { Hono } from \"npm:hono\";\nimport { sqlite } from \"https://esm.town/v/std/sqlite/main.ts\";\n\nconst CLIENT_TOKEN = requireEnv(\"BRIDGE_CLIENT_TOKEN\");\nconst WORKER_TOKEN = requireEnv(\"BRIDGE_WORKER_TOKEN\");\n\nconst DEFAULT_CHANNEL = \"default\";\nconst REQUEST_TTL_MS = 120_000;\nconst OAUTH_CODE_TTL_MS = 60_000;\n\n// Val Town execution timeout: 60s (Free) / 600s (Pro) per single val run.\n// Held under the shorter ceiling so requests always resolve cleanly.\nconst CLIENT_MAX_WAIT_MS = 55_000;\nconst CLIENT_POLL_INTERVAL_MS = 250;\n\nconst WORKER_POLL_WINDOW_MS = 25_000;\nconst WORKER_POLL_INTERVAL_MS = 500;\n\n// Streaming push is self-bounded well under the execution ceiling because\n// it's undocumented whether that ceiling applies to a chunked response the\n// same way it applies to a normal one — this is safe either way.\nconst STREAM_MAX_MS = 50_000;\nconst STREAM_CHECK_INTERVAL_MS = 200;\nconst STREAM_HEARTBEAT_INTERVAL_MS = 10_000;\n\nfunction requireEnv(key: string): string {\n const v = Deno.env.get(key);\n if (!v) throw new Error(`Missing required environment variable: ${key}`);\n return v;\n}\n\nawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS mcp_bridge_requests (\n id TEXT PRIMARY KEY,\n channel TEXT NOT NULL DEFAULT 'default',\n method TEXT NOT NULL,\n params TEXT NOT NULL,\n status TEXT NOT NULL DEFAULT 'pending', -- pending | claimed | done\n result TEXT,\n error TEXT,\n created_at INTEGER NOT NULL\n )\n`);\nawait sqlite.execute(\n `CREATE INDEX IF NOT EXISTS idx_bridge_channel_status ON mcp_bridge_requests(channel, status, created_at)`,\n);\n\nawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS oauth_codes (\n code TEXT PRIMARY KEY,\n redirect_uri TEXT NOT NULL,\n code_challenge TEXT NOT NULL,\n created_at INTEGER NOT NULL\n )\n`);\n\nconst sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));\nconst rpcResult = (id: unknown, result: unknown) => ({\n jsonrpc: \"2.0\",\n id,\n result,\n});\nconst rpcError = (id: unknown, code: number, message: string) => ({\n jsonrpc: \"2.0\",\n id,\n error: { code, message },\n});\nconst bearer = (req: Request) =>\n (req.headers.get(\"authorization\") ?? \"\").match(/^Bearer (.+)$/)?.[1] ?? null;\n\nfunction escapeHtml(s: string): string {\n return s.replace(\n /[&<>\"']/g,\n (\n ch,\n ) => ({\n \"&\": \"&\",\n \"<\": \"<\",\n \">\": \">\",\n '\"': \""\",\n \"'\": \"'\",\n }[ch]!),\n );\n}\n\nasync function sha256Base64Url(input: string): Promise {\n const digest = await crypto.subtle.digest(\n \"SHA-256\",\n new TextEncoder().encode(input),\n );\n return btoa(String.fromCharCode(...new Uint8Array(digest)))\n .replace(/\\+/g, \"-\")\n .replace(/\\//g, \"_\")\n .replace(/=+$/, \"\");\n}\n\nasync function purgeExpired() {\n await sqlite.execute({\n sql:\n `DELETE FROM mcp_bridge_requests WHERE status != 'done' AND created_at < ?`,\n args: [Date.now() - REQUEST_TTL_MS],\n });\n await sqlite.execute({\n sql: `DELETE FROM oauth_codes WHERE created_at < ?`,\n args: [Date.now() - OAUTH_CODE_TTL_MS],\n });\n}\n\n// Atomic claim: the UPDATE...WHERE...RETURNING is one SQLite statement, so\n// concurrent pollers/streamers can never claim the same row.\nasync function claimOne(\n channel: string,\n): Promise<\n { id: string; method: string; params: Record } | null\n> {\n const { rows } = await sqlite.execute({\n sql: `UPDATE mcp_bridge_requests SET status = 'claimed'\n WHERE id = (SELECT id FROM mcp_bridge_requests WHERE channel = ? AND status = 'pending' ORDER BY created_at LIMIT 1)\n RETURNING id, method, params`,\n args: [channel],\n });\n if (rows.length === 0) return null;\n const row = rows[0] as any as { id: string; method: string; params: string };\n return { id: row.id, method: row.method, params: JSON.parse(row.params) };\n}\n\nasync function enqueue(\n channel: string,\n method: string,\n params: Record,\n): Promise {\n const id = crypto.randomUUID();\n await sqlite.execute({\n sql:\n `INSERT INTO mcp_bridge_requests (id, channel, method, params, status, created_at) VALUES (?, ?, ?, ?, 'pending', ?)`,\n args: [id, channel, method, JSON.stringify(params), Date.now()],\n });\n return id;\n}\n\n// Blocks until the row lands in 'done' or CLIENT_MAX_WAIT_MS elapses.\nasync function dispatch(\n channel: string,\n method: string,\n params: Record,\n) {\n const id = await enqueue(channel, method, params);\n const deadline = Date.now() + CLIENT_MAX_WAIT_MS;\n\n while (Date.now() < deadline) {\n const { rows } = await sqlite.execute({\n sql:\n `SELECT result, error FROM mcp_bridge_requests WHERE id = ? AND status = 'done'`,\n args: [id],\n });\n if (rows.length > 0) {\n const row = rows[0] as any as {\n result: string | null;\n error: string | null;\n };\n await sqlite.execute({\n sql: `DELETE FROM mcp_bridge_requests WHERE id = ?`,\n args: [id],\n });\n return {\n result: row.result != null ? JSON.parse(row.result) : undefined,\n error: row.error ?? undefined,\n };\n }\n await sleep(CLIENT_POLL_INTERVAL_MS);\n }\n await sqlite.execute({\n sql: `DELETE FROM mcp_bridge_requests WHERE id = ?`,\n args: [id],\n });\n return null;\n}\n\ninterface JsonRpcRequest {\n jsonrpc: \"2.0\";\n id?: string | number | null;\n method: string;\n params?: Record;\n}\n\nasync function handleRpc(channel: string, req: JsonRpcRequest) {\n const { id, method, params = {} } = req;\n\n if (method === \"notifications/initialized\") {\n await enqueue(channel, method, params); // fire-and-forget: no response expected\n return null;\n }\n\n const outcome = await dispatch(channel, method, params);\n if (outcome === null) {\n return rpcError(\n id,\n -32001,\n `Timed out waiting for the local agent on channel '${channel}' to answer '${method}'.`,\n );\n }\n return outcome.error\n ? rpcError(id, -32000, outcome.error)\n : rpcResult(id, outcome.result);\n}\n\nconst app = new Hono();\n\nasync function handleMcpPost(c: any, channel: string) {\n if (bearer(c.req.raw) !== CLIENT_TOKEN) {\n const origin = new URL(c.req.url).origin;\n c.header(\n \"WWW-Authenticate\",\n `Bearer resource_metadata=\"${origin}/.well-known/oauth-protected-resource\"`,\n );\n return c.json(rpcError(null, -32000, \"Unauthorized\"), 401);\n }\n await purgeExpired();\n\n let body: JsonRpcRequest | JsonRpcRequest[];\n try {\n body = await c.req.json();\n } catch {\n return c.json(rpcError(null, -32700, \"Parse error\"), 400);\n }\n\n const batch = Array.isArray(body) ? body : [body];\n const responses = (await Promise.all(batch.map((r) => handleRpc(channel, r))))\n .filter((r) => r !== null);\n\n if (responses.length === 0) return c.body(null, 202);\n return c.json(Array.isArray(body) ? responses : responses[0]);\n}\n\napp.post(\"/mcp\", (c) => handleMcpPost(c, DEFAULT_CHANNEL));\napp.post(\"/mcp/:channel\", (c) => handleMcpPost(c, c.req.param(\"channel\")));\napp.get(\n \"/mcp\",\n (c) => c.json({ error: \"POST only — no SSE client transport.\" }, 405),\n);\n\nasync function handlePoll(c: any, channel: string) {\n if (bearer(c.req.raw) !== WORKER_TOKEN) return c.body(null, 401);\n const deadline = Date.now() + WORKER_POLL_WINDOW_MS;\n while (Date.now() < deadline) {\n const job = await claimOne(channel);\n if (job) return c.json(job);\n await sleep(WORKER_POLL_INTERVAL_MS);\n }\n return c.body(null, 204);\n}\napp.get(\"/poll\", (c) => handlePoll(c, DEFAULT_CHANNEL));\napp.get(\"/poll/:channel\", (c) => handlePoll(c, c.req.param(\"channel\")));\n\nfunction handleStream(c: any, channel: string) {\n if (bearer(c.req.raw) !== WORKER_TOKEN) return c.body(null, 401);\n\n const stream = new ReadableStream({\n async start(controller) {\n const encoder = new TextEncoder();\n const deadline = Date.now() + STREAM_MAX_MS;\n let lastHeartbeat = Date.now();\n try {\n while (Date.now() < deadline) {\n const job = await claimOne(channel);\n if (job) {\n controller.enqueue(encoder.encode(JSON.stringify(job) + \"\\n\"));\n continue;\n }\n if (Date.now() - lastHeartbeat > STREAM_HEARTBEAT_INTERVAL_MS) {\n controller.enqueue(encoder.encode(\"\\n\")); // keeps intermediaries from buffering the whole stream\n lastHeartbeat = Date.now();\n }\n await sleep(STREAM_CHECK_INTERVAL_MS);\n }\n } catch {\n // client disconnected — no job was left half-claimed, nothing to clean up\n } finally {\n controller.close();\n }\n },\n });\n\n return new Response(stream, {\n headers: {\n \"content-type\": \"application/x-ndjson\",\n \"cache-control\": \"no-store\",\n },\n });\n}\napp.get(\"/stream\", (c) => handleStream(c, DEFAULT_CHANNEL));\napp.get(\"/stream/:channel\", (c) => handleStream(c, c.req.param(\"channel\")));\n\napp.post(\"/respond\", async (c) => {\n if (bearer(c.req.raw) !== WORKER_TOKEN) return c.body(null, 401);\n const { id, result, error } = await c.req.json() as {\n id: string;\n result?: unknown;\n error?: string;\n };\n await sqlite.execute({\n sql:\n `UPDATE mcp_bridge_requests SET status = 'done', result = ?, error = ? WHERE id = ?`,\n args: [\n result !== undefined ? JSON.stringify(result) : null,\n error ?? null,\n id,\n ],\n });\n return c.body(null, 204);\n});\n\napp.get(\"/health\", async (c) => {\n await purgeExpired();\n const { rows } = await sqlite.execute(\n `SELECT channel, COUNT(*) as stuck FROM mcp_bridge_requests WHERE status != 'done' GROUP BY channel`,\n );\n return c.json({ ok: rows.length === 0, stuck: rows });\n});\n\n// ── OAuth 2.1 authorization server (single-tenant) ─────────────────────\n\napp.get(\"/.well-known/oauth-authorization-server\", (c) => {\n const origin = new URL(c.req.url).origin;\n return c.json({\n issuer: origin,\n authorization_endpoint: `${origin}/authorize`,\n token_endpoint: `${origin}/token`,\n registration_endpoint: `${origin}/register`,\n response_types_supported: [\"code\"],\n grant_types_supported: [\"authorization_code\", \"refresh_token\"],\n code_challenge_methods_supported: [\"S256\"],\n token_endpoint_auth_methods_supported: [\"none\"],\n });\n});\n\napp.get(\"/.well-known/oauth-protected-resource\", (c) => {\n const origin = new URL(c.req.url).origin;\n return c.json({\n resource: origin,\n authorization_servers: [origin],\n bearer_methods_supported: [\"header\"],\n });\n});\n\n// DCR: client_id isn't a security boundary for this AS (PKCE is), so\n// registration always succeeds and nothing is persisted.\napp.post(\"/register\", async (c) => {\n let body: Record = {};\n try {\n body = await c.req.json();\n } catch {\n // empty registration request is fine\n }\n return c.json({\n client_id: crypto.randomUUID(),\n client_id_issued_at: Math.floor(Date.now() / 1000),\n redirect_uris: body.redirect_uris ?? [],\n token_endpoint_auth_method: \"none\",\n grant_types: [\"authorization_code\", \"refresh_token\"],\n response_types: [\"code\"],\n }, 201);\n});\n\nconst AUTHORIZE_FIELDS = [\n \"response_type\",\n \"client_id\",\n \"redirect_uri\",\n \"code_challenge\",\n \"code_challenge_method\",\n \"state\",\n \"resource\",\n] as const;\n\nfunction renderAuthorizeForm(\n q: Record,\n error: string | null,\n): string {\n const hidden = AUTHORIZE_FIELDS\n .map((k) =>\n q[k] !== undefined\n ? ``\n : \"\"\n )\n .join(\"\\n\");\n return `\n

valtown-mcp-bridge authorization

\n${error ? `

${escapeHtml(error)}

` : \"\"}\n\n`;\n}\n\napp.get(\"/authorize\", (c) => {\n const q = c.req.query();\n return c.html(renderAuthorizeForm(q, null));\n});\n\napp.post(\"/authorize\", async (c) => {\n const body = await c.req.parseBody() as Record;\n\n if (body.key !== CLIENT_TOKEN) {\n return c.html(renderAuthorizeForm(body, \"Incorrect key.\"), 401);\n }\n if (body.response_type !== \"code\") {\n return c.html(\n renderAuthorizeForm(body, \"Only response_type=code is supported.\"),\n 400,\n );\n }\n if (body.code_challenge_method !== \"S256\") {\n return c.html(\n renderAuthorizeForm(body, \"Only PKCE S256 is supported.\"),\n 400,\n );\n }\n const redirectUri = body.redirect_uri ?? \"\";\n if (\n !/^https:\\/\\//.test(redirectUri) &&\n !/^http:\\/\\/(localhost|127\\.0\\.0\\.1)(:\\d+)?\\//.test(redirectUri)\n ) {\n return c.html(renderAuthorizeForm(body, \"Unrecognized redirect_uri.\"), 400);\n }\n if (!body.code_challenge) {\n return c.html(renderAuthorizeForm(body, \"Missing code_challenge.\"), 400);\n }\n\n const code = crypto.randomUUID();\n await sqlite.execute({\n sql:\n `INSERT INTO oauth_codes (code, redirect_uri, code_challenge, created_at) VALUES (?, ?, ?, ?)`,\n args: [code, redirectUri, body.code_challenge, Date.now()],\n });\n\n const dest = new URL(redirectUri);\n dest.searchParams.set(\"code\", code);\n if (body.state) dest.searchParams.set(\"state\", body.state);\n return c.redirect(dest.toString(), 302);\n});\n\napp.post(\"/token\", async (c) => {\n let params: Record = {};\n try {\n params = await c.req.parseBody() as Record;\n } catch {\n try {\n params = await c.req.json();\n } catch {\n return c.json({ error: \"invalid_request\" }, 400);\n }\n }\n\n if (params.grant_type === \"authorization_code\") {\n const { rows } = await sqlite.execute({\n sql:\n `SELECT redirect_uri, code_challenge, created_at FROM oauth_codes WHERE code = ?`,\n args: [params.code ?? \"\"],\n });\n if (rows.length === 0) {\n return c.json({\n error: \"invalid_grant\",\n error_description: \"unknown or already-used code\",\n }, 400);\n }\n const row = rows[0] as any as {\n redirect_uri: string;\n code_challenge: string;\n created_at: number;\n };\n await sqlite.execute({\n sql: `DELETE FROM oauth_codes WHERE code = ?`,\n args: [params.code],\n }); // one-time use\n\n if (Date.now() - row.created_at > OAUTH_CODE_TTL_MS) {\n return c.json({\n error: \"invalid_grant\",\n error_description: \"code expired\",\n }, 400);\n }\n if (params.redirect_uri && params.redirect_uri !== row.redirect_uri) {\n return c.json({\n error: \"invalid_grant\",\n error_description: \"redirect_uri mismatch\",\n }, 400);\n }\n const computed = await sha256Base64Url(params.code_verifier ?? \"\");\n if (computed !== row.code_challenge) {\n return c.json({\n error: \"invalid_grant\",\n error_description: \"PKCE verification failed\",\n }, 400);\n }\n\n return c.json({\n access_token: CLIENT_TOKEN,\n token_type: \"bearer\",\n expires_in: 3600,\n refresh_token: CLIENT_TOKEN,\n });\n }\n\n if (params.grant_type === \"refresh_token\") {\n if (params.refresh_token !== CLIENT_TOKEN) {\n return c.json({ error: \"invalid_grant\" }, 400);\n }\n return c.json({\n access_token: CLIENT_TOKEN,\n token_type: \"bearer\",\n expires_in: 3600,\n refresh_token: CLIENT_TOKEN,\n });\n }\n\n return c.json({ error: \"unsupported_grant_type\" }, 400);\n});\n\napp.get(\"/\", (c) =>\n c.json({\n name: \"valtown-mcp-bridge\",\n endpoints: {\n client: [\"POST /mcp[/:channel]\"],\n agent: [\n \"GET /poll[/:channel]\",\n \"GET /stream[/:channel]\",\n \"POST /respond\",\n ],\n oauth: [\n \"GET /.well-known/oauth-authorization-server\",\n \"GET /.well-known/oauth-protected-resource\",\n \"POST /register\",\n \"GET|POST /authorize\",\n \"POST /token\",\n ],\n },\n }));\n\napp.onError((err) => Promise.reject(err));\n\nexport default app.fetch;","//https://mcp-bridge.val.run/\n// bridge.ts — Val Town HTTP val. Public MCP relay over a SQLite job queue.\n//\n// Neither side can accept an inbound connection: the local machine is\n// behind NAT with no forwarded ports, and Val Town has no persistent\n// process to hold a socket open. So the queue is the protocol. MCP clients\n// POST to /mcp/:channel and block until a local agent — which only ever\n// makes outbound calls — claims the row and posts a result back.\n//\n// No tool registry lives here. Every method except the initialized\n// notification is forwarded verbatim; the local agent's backend is the\n// only source of truth for what tools exist.\n//\n// OAUTH LAYER (added): claude.ai's custom-connector flow requires the\n// target to behave as an OAuth 2.1 authorization server with PKCE and\n// Dynamic Client Registration — a bare static bearer token is not an\n// accepted auth mode for connectors added by URL. Since this bridge has\n// exactly one real credential (CLIENT_TOKEN) and exactly one consumer\n// per channel, the AS below is deliberately single-tenant: /register\n// always succeeds (client_id isn't a security boundary here — PKCE is),\n// /authorize is a one-field form gated by CLIENT_TOKEN itself (so the\n// interactive consent step doesn't introduce a second secret to manage),\n// and /token exchanges a PKCE-verified code for CLIENT_TOKEN itself as\n// the access_token. This means /mcp's existing bearer check needs no\n// changes — an OAuth-issued token IS the token it already accepts.\n\nimport { Hono } from \"npm:hono\";\nimport { sqlite } from \"https://esm.town/v/std/sqlite/main.ts\";\n\nconst CLIENT_TOKEN = requireEnv(\"BRIDGE_CLIENT_TOKEN\");\nconst WORKER_TOKEN = requireEnv(\"BRIDGE_WORKER_TOKEN\");\n\nconst DEFAULT_CHANNEL = \"default\";\nconst REQUEST_TTL_MS = 120_000;\nconst OAUTH_CODE_TTL_MS = 60_000;\n\n// Val Town execution timeout: 60s (Free) / 600s (Pro) per single val run.\n// Held under the shorter ceiling so requests always resolve cleanly.\nconst CLIENT_MAX_WAIT_MS = 55_000;\nconst CLIENT_POLL_INTERVAL_MS = 250;\n\nconst WORKER_POLL_WINDOW_MS = 25_000;\nconst WORKER_POLL_INTERVAL_MS = 500;\n\n// Streaming push is self-bounded well under the execution ceiling because\n// it's undocumented whether that ceiling applies to a chunked response the\n// same way it applies to a normal one — this is safe either way.\nconst STREAM_MAX_MS = 50_000;\nconst STREAM_CHECK_INTERVAL_MS = 200;\nconst STREAM_HEARTBEAT_INTERVAL_MS = 10_000;\n\nfunction requireEnv(key: string): string {\n const v = Deno.env.get(key);\n if (!v) throw new Error(`Missing required environment variable: ${key}`);\n return v;\n}\n\nawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS mcp_bridge_requests (\n id TEXT PRIMARY KEY,\n channel TEXT NOT NULL DEFAULT 'default',\n method TEXT NOT NULL,\n params TEXT NOT NULL,\n status TEXT NOT NULL DEFAULT 'pending', -- pending | claimed | done\n result TEXT,\n error TEXT,\n created_at INTEGER NOT NULL\n )\n`);\nawait sqlite.execute(\n `CREATE INDEX IF NOT EXISTS idx_bridge_channel_status ON mcp_bridge_requests(channel, status, created_at)`,\n);\n\nawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS oauth_codes (\n code TEXT PRIMARY KEY,\n redirect_uri TEXT NOT NULL,\n code_challenge TEXT NOT NULL,\n created_at INTEGER NOT NULL\n )\n`);\n\nconst sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));\nconst rpcResult = (id: unknown, result: unknown) => ({\n jsonrpc: \"2.0\",\n id,\n result,\n});\nconst rpcError = (id: unknown, code: number, message: string) => ({\n jsonrpc: \"2.0\",\n id,\n error: { code, message },\n});\nconst bearer = (req: Request) =>\n (req.headers.get(\"authorization\") ?? \"\").match(/^Bearer (.+)$/)?.[1] ?? null;\n\nfunction escapeHtml(s: string): string {\n return s.replace(\n /[&<>\"']/g,\n (\n ch,\n ) => ({\n \"&\": \"&\",\n \"<\": \"<\",\n \">\": \">\",\n '\"': \""\",\n \"'\": \"'\",\n }[ch]!),\n );\n}\n\nasync function sha256Base64Url(input: string): Promise {\n const digest = await crypto.subtle.digest(\n \"SHA-256\",\n new TextEncoder().encode(input),\n );\n return btoa(String.fromCharCode(...new Uint8Array(digest)))\n .replace(/\\+/g, \"-\")\n .replace(/\\//g, \"_\")\n .replace(/=+$/, \"\");\n}\n\nasync function purgeExpired() {\n await sqlite.execute({\n sql:\n `DELETE FROM mcp_bridge_requests WHERE status != 'done' AND created_at < ?`,\n args: [Date.now() - REQUEST_TTL_MS],\n });\n await sqlite.execute({\n sql: `DELETE FROM oauth_codes WHERE created_at < ?`,\n args: [Date.now() - OAUTH_CODE_TTL_MS],\n });\n}\n\n// Atomic claim: the UPDATE...WHERE...RETURNING is one SQLite statement, so\n// concurrent pollers/streamers can never claim the same row.\nasync function claimOne(\n channel: string,\n): Promise<\n { id: string; method: string; params: Record } | null\n> {\n const { rows } = await sqlite.execute({\n sql: `UPDATE mcp_bridge_requests SET status = 'claimed'\n WHERE id = (SELECT id FROM mcp_bridge_requests WHERE channel = ? AND status = 'pending' ORDER BY created_at LIMIT 1)\n RETURNING id, method, params`,\n args: [channel],\n });\n if (rows.length === 0) return null;\n const row = rows[0] as any as { id: string; method: string; params: string };\n return { id: row.id, method: row.method, params: JSON.parse(row.params) };\n}\n\nasync function enqueue(\n channel: string,\n method: string,\n params: Record,\n): Promise {\n const id = crypto.randomUUID();\n await sqlite.execute({\n sql:\n `INSERT INTO mcp_bridge_requests (id, channel, method, params, status, created_at) VALUES (?, ?, ?, ?, 'pending', ?)`,\n args: [id, channel, method, JSON.stringify(params), Date.now()],\n });\n return id;\n}\n\n// Blocks until the row lands in 'done' or CLIENT_MAX_WAIT_MS elapses.\nasync function dispatch(\n channel: string,\n method: string,\n params: Record,\n) {\n const id = await enqueue(channel, method, params);\n const deadline = Date.now() + CLIENT_MAX_WAIT_MS;\n\n while (Date.now() < deadline) {\n const { rows } = await sqlite.execute({\n sql:\n `SELECT result, error FROM mcp_bridge_requests WHERE id = ? AND status = 'done'`,\n args: [id],\n });\n if (rows.length > 0) {\n const row = rows[0] as any as {\n result: string | null;\n error: string | null;\n };\n await sqlite.execute({\n sql: `DELETE FROM mcp_bridge_requests WHERE id = ?`,\n args: [id],\n });\n return {\n result: row.result != null ? JSON.parse(row.result) : undefined,\n error: row.error ?? undefined,\n };\n }\n await sleep(CLIENT_POLL_INTERVAL_MS);\n }\n await sqlite.execute({\n sql: `DELETE FROM mcp_bridge_requests WHERE id = ?`,\n args: [id],\n });\n return null;\n}\n\ninterface JsonRpcRequest {\n jsonrpc: \"2.0\";\n id?: string | number | null;\n method: string;\n params?: Record;\n}\n\nasync function handleRpc(channel: string, req: JsonRpcRequest) {\n const { id, method, params = {} } = req;\n\n if (method === \"notifications/initialized\") {\n await enqueue(channel, method, params); // fire-and-forget: no response expected\n return null;\n }\n\n const outcome = await dispatch(channel, method, params);\n if (outcome === null) {\n return rpcError(\n id,\n -32001,\n `Timed out waiting for the local agent on channel '${channel}' to answer '${method}'.`,\n );\n }\n return outcome.error\n ? rpcError(id, -32000, outcome.error)\n : rpcResult(id, outcome.result);\n}\n\nconst app = new Hono();\n\nasync function handleMcpPost(c: any, channel: string) {\n if (bearer(c.req.raw) !== CLIENT_TOKEN) {\n const origin = new URL(c.req.url).origin;\n c.header(\n \"WWW-Authenticate\",\n `Bearer resource_metadata=\"${origin}/.well-known/oauth-protected-resource\"`,\n );\n return c.json(rpcError(null, -32000, \"Unauthorized\"), 401);\n }\n await purgeExpired();\n\n let body: JsonRpcRequest | JsonRpcRequest[];\n try {\n body = await c.req.json();\n } catch {\n return c.json(rpcError(null, -32700, \"Parse error\"), 400);\n }\n\n const batch = Array.isArray(body) ? body : [body];\n const responses = (await Promise.all(batch.map((r) => handleRpc(channel, r))))\n .filter((r) => r !== null);\n\n if (responses.length === 0) return c.body(null, 202);\n return c.json(Array.isArray(body) ? responses : responses[0]);\n}\n\napp.post(\"/mcp\", (c) => handleMcpPost(c, DEFAULT_CHANNEL));\napp.post(\"/mcp/:channel\", (c) => handleMcpPost(c, c.req.param(\"channel\")));\napp.get(\n \"/mcp\",\n (c) => c.json({ error: \"POST only — no SSE client transport.\" }, 405),\n);\n\nasync function handlePoll(c: any, channel: string) {\n if (bearer(c.req.raw) !== WORKER_TOKEN) return c.body(null, 401);\n const deadline = Date.now() + WORKER_POLL_WINDOW_MS;\n while (Date.now() < deadline) {\n const job = await claimOne(channel);\n if (job) return c.json(job);\n await sleep(WORKER_POLL_INTERVAL_MS);\n }\n return c.body(null, 204);\n}\napp.get(\"/poll\", (c) => handlePoll(c, DEFAULT_CHANNEL));\napp.get(\"/poll/:channel\", (c) => handlePoll(c, c.req.param(\"channel\")));\n\nfunction handleStream(c: any, channel: string) {\n // if (bearer(c.req.raw) !== WORKER_TOKEN) return c.body(null, 401);\n\n const stream = new ReadableStream({\n async start(controller) {\n const encoder = new TextEncoder();\n const deadline = Date.now() + STREAM_MAX_MS;\n let lastHeartbeat = Date.now();\n try {\n while (Date.now() < deadline) {\n const job = await claimOne(channel);\n if (job) {\n controller.enqueue(encoder.encode(JSON.stringify(job) + \"\\n\"));\n continue;\n }\n if (Date.now() - lastHeartbeat > STREAM_HEARTBEAT_INTERVAL_MS) {\n controller.enqueue(encoder.encode(\"\\n\")); // keeps intermediaries from buffering the whole stream\n lastHeartbeat = Date.now();\n }\n await sleep(STREAM_CHECK_INTERVAL_MS);\n }\n } catch {\n // client disconnected — no job was left half-claimed, nothing to clean up\n } finally {\n controller.close();\n }\n },\n });\n\n return new Response(stream, {\n headers: {\n \"content-type\": \"application/x-ndjson\",\n \"cache-control\": \"no-store\",\n },\n });\n}\napp.get(\"/stream\", (c) => handleStream(c, DEFAULT_CHANNEL));\napp.get(\"/stream/:channel\", (c) => handleStream(c, c.req.param(\"channel\")));\n\napp.post(\"/respond\", async (c) => {\n if (bearer(c.req.raw) !== WORKER_TOKEN) return c.body(null, 401);\n const { id, result, error } = await c.req.json() as {\n id: string;\n result?: unknown;\n error?: string;\n };\n await sqlite.execute({\n sql:\n `UPDATE mcp_bridge_requests SET status = 'done', result = ?, error = ? WHERE id = ?`,\n args: [\n result !== undefined ? JSON.stringify(result) : null,\n error ?? null,\n id,\n ],\n });\n return c.body(null, 204);\n});\n\napp.get(\"/health\", async (c) => {\n await purgeExpired();\n const { rows } = await sqlite.execute(\n `SELECT channel, COUNT(*) as stuck FROM mcp_bridge_requests WHERE status != 'done' GROUP BY channel`,\n );\n return c.json({ ok: rows.length === 0, stuck: rows });\n});\n\n// ── OAuth 2.1 authorization server (single-tenant) ─────────────────────\n\napp.get(\"/.well-known/oauth-authorization-server\", (c) => {\n const origin = new URL(c.req.url).origin;\n return c.json({\n issuer: origin,\n authorization_endpoint: `${origin}/authorize`,\n token_endpoint: `${origin}/token`,\n registration_endpoint: `${origin}/register`,\n response_types_supported: [\"code\"],\n grant_types_supported: [\"authorization_code\", \"refresh_token\"],\n code_challenge_methods_supported: [\"S256\"],\n token_endpoint_auth_methods_supported: [\"none\"],\n });\n});\n\napp.get(\"/.well-known/oauth-protected-resource\", (c) => {\n const origin = new URL(c.req.url).origin;\n return c.json({\n resource: origin,\n authorization_servers: [origin],\n bearer_methods_supported: [\"header\"],\n });\n});\n\n// DCR: client_id isn't a security boundary for this AS (PKCE is), so\n// registration always succeeds and nothing is persisted.\napp.post(\"/register\", async (c) => {\n let body: Record = {};\n try {\n body = await c.req.json();\n } catch {\n // empty registration request is fine\n }\n return c.json({\n client_id: crypto.randomUUID(),\n client_id_issued_at: Math.floor(Date.now() / 1000),\n redirect_uris: body.redirect_uris ?? [],\n token_endpoint_auth_method: \"none\",\n grant_types: [\"authorization_code\", \"refresh_token\"],\n response_types: [\"code\"],\n }, 201);\n});\n\nconst AUTHORIZE_FIELDS = [\n \"response_type\",\n \"client_id\",\n \"redirect_uri\",\n \"code_challenge\",\n \"code_challenge_method\",\n \"state\",\n \"resource\",\n] as const;\n\nfunction renderAuthorizeForm(\n q: Record,\n error: string | null,\n): string {\n const hidden = AUTHORIZE_FIELDS\n .map((k) =>\n q[k] !== undefined\n ? ``\n : \"\"\n )\n .join(\"\\n\");\n return `\n

valtown-mcp-bridge authorization

\n${error ? `

${escapeHtml(error)}

Diff · v9