// ---------------------------------------------------------------------------// Cloud MCP handler — Effect-native HTTP app for /mcp + /.well-known/*// ---------------------------------------------------------------------------//// Built on `@effect/platform`'s `HttpApp.toWebHandler`. start.ts's// mcpRequestMiddleware calls `mcpFetch` and falls through to `next()` when it// returns `null` (non-MCP path) so TanStack Start keeps routing.//// Streaming passthrough — the MCP session Durable Object returns a `Response`// whose body is a `ReadableStream` (SSE). We wrap that `Response` in// `HttpServerResponse.raw(response)`; the platform's `toWeb` conversion// recognises `body.body instanceof Response` and returns it as-is (only// merging headers we set on the outer response, which is none), so the// underlying `ReadableStream` passes through untouched.// --------------------------------------------------------------------------- import { env } from "cloudflare:workers";import { HttpApp, HttpServerRequest, HttpServerResponse } from "@effect/platform";import * as Sentry from "@sentry/cloudflare";import { Context, Effect, Layer, Option, Schema } from "effect";import { createRemoteJWKSet, jwtVerify } from "jose"; import { TelemetryLive } from "./services/telemetry"; // ---------------------------------------------------------------------------// Constants// --------------------------------------------------------------------------- const AUTHKIT_DOMAIN = env.MCP_AUTHKIT_DOMAIN ?? "https://signin.executor.sh";const RESOURCE_ORIGIN = env.MCP_RESOURCE_ORIGIN ?? "https://executor.sh"; const jwks = createRemoteJWKSet(new URL(`${AUTHKIT_DOMAIN}/oauth2/jwks`)); const BEARER_PREFIX = "Bearer "; const CORS_ALLOW_ORIGIN = { "access-control-allow-origin": "*" } as const; const CORS_PREFLIGHT_HEADERS = { ...CORS_ALLOW_ORIGIN, "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", "access-control-allow-headers": "authorization, content-type, mcp-session-id, accept, mcp-protocol-version", "access-control-expose-headers": "mcp-session-id",} as const; const WWW_AUTHENTICATE = `Bearer resource_metadata="${RESOURCE_ORIGIN}/.well-known/oauth-protected-resource"`; // ---------------------------------------------------------------------------// Response helpers// --------------------------------------------------------------------------- const jsonResponse = (body: unknown, status = 200) => HttpServerResponse.unsafeJson(body, { status, headers: CORS_ALLOW_ORIGIN }); const jsonRpcError = (status: number, code: number, message: string) => HttpServerResponse.unsafeJson({ jsonrpc: "2.0", error: { code, message }, id: null }, { status }); const unauthorized = HttpServerResponse.unsafeJson( { error: "unauthorized" }, { status: 401, headers: { ...CORS_ALLOW_ORIGIN, "www-authenticate": WWW_AUTHENTICATE }, },); const corsPreflight = HttpServerResponse.empty({ status: 204, headers: CORS_PREFLIGHT_HEADERS,}); // ---------------------------------------------------------------------------// Auth// --------------------------------------------------------------------------- export type VerifiedToken = { /** The WorkOS account ID (user ID). */ accountId: string; /** The WorkOS organization ID, if the session has org context. */ organizationId: string | null;}; export class McpAuth extends Context.Tag("@executor/cloud/McpAuth")< McpAuth, { readonly verifyBearer: (request: Request) => Effect.Effect<VerifiedToken | null>; }>() {} export const McpAuthLive = Layer.succeed(McpAuth, { verifyBearer: (request) => Effect.promise(async () => { const authHeader = request.headers.get("authorization"); if (!authHeader?.startsWith(BEARER_PREFIX)) return null; try { const { payload } = await jwtVerify(authHeader.slice(BEARER_PREFIX.length), jwks, { issuer: AUTHKIT_DOMAIN, }); if (!payload.sub) return null; return { accountId: payload.sub, organizationId: (payload.org_id as string | undefined) ?? null, }; } catch { return null; } }),}); // ---------------------------------------------------------------------------// Client fingerprint capture// ---------------------------------------------------------------------------// Annotates the Effect span (which nests under the otel-cf-workers fetch// span) with everything we can learn about a connecting MCP client: the// parsed JSON-RPC body, whitelisted request headers, CF request metadata,// and verified-JWT claims. Lets us compare how each client (Claude Code,// Claude.ai web, ChatGPT, custom scripts, ...) actually reports over the// wire. Runs before dispatch so unauthorized requests still get fingerprinted.// --------------------------------------------------------------------------- type CfRequestMetadata = { country?: string; city?: string; region?: string; timezone?: string; asn?: number; asOrganization?: string; tlsVersion?: string; tlsCipher?: string; httpProtocol?: string; colo?: string;}; const getCfMeta = (request: Request): CfRequestMetadata => ((request as unknown as { cf?: CfRequestMetadata }).cf ?? {}) as CfRequestMetadata; const HEADERS_TO_DUMP = [ "accept", "accept-encoding", "accept-language", "cache-control", "content-type", "mcp-protocol-version", "origin", "referer", "sec-fetch-dest", "sec-fetch-mode", "sec-fetch-site", "user-agent", "x-client-name", "x-client-version", "x-requested-with",] as const; const dumpHeaders = (request: Request): Record<string, string> => { const out: Record<string, string> = {}; for (const name of HEADERS_TO_DUMP) { const value = request.headers.get(name); if (value !== null) out[`mcp.http.header.${name}`] = value; } const authHeader = request.headers.get("authorization"); if (authHeader) { out["mcp.http.header.authorization.scheme"] = authHeader.split(" ", 1)[0] ?? ""; out["mcp.http.header.authorization.length"] = String(authHeader.length); } // Record the full header name list too — surfaces anything unexpected // without us having to enumerate every possibility up front. out["mcp.http.header.names"] = Array.from(request.headers.keys()).sort().join(","); return out;}; // JSON-RPC shapes — narrow to just the fields we fingerprint. Using Schema// collapses the typeof-guard pile and surfaces "what does an MCP client// actually send us" as declarative types. Unknown/malformed input decodes// to None and contributes no span attrs. const UnknownRecord = Schema.Record({ key: Schema.String, value: Schema.Unknown }); const JsonRpcEnvelope = Schema.Struct({ method: Schema.optional(Schema.String), id: Schema.optional(Schema.Union(Schema.String, Schema.Number, Schema.Null)), params: Schema.optional(UnknownRecord), // Responses to server-initiated requests arrive as POST bodies too — // notably elicitation replies (`result.action = "accept" | "decline" | "cancel"`). result: Schema.optional(UnknownRecord),});type JsonRpcEnvelope = typeof JsonRpcEnvelope.Type; const ElicitationReplyResult = Schema.Struct({ action: Schema.optional(Schema.Literal("accept", "decline", "cancel")),}); const InitializeParams = Schema.Struct({ protocolVersion: Schema.optional(Schema.String), clientInfo: Schema.optional( Schema.Struct({ name: Schema.optional(Schema.String), version: Schema.optional(Schema.String), title: Schema.optional(Schema.String), }), ), capabilities: Schema.optional(UnknownRecord),}); const NamedParams = Schema.Struct({ name: Schema.optional(Schema.String) });const UriParams = Schema.Struct({ uri: Schema.optional(Schema.String) }); const decode = <A, I>(schema: Schema.Schema<A, I>, input: unknown): Option.Option<A> => Schema.decodeUnknownOption(schema)(input); const readJsonRpcEnvelope = (request: Request): Effect.Effect<Option.Option<JsonRpcEnvelope>> => Effect.promise(async () => { try { const text = await request.clone().text(); if (!text) return Option.none(); return decode(JsonRpcEnvelope, JSON.parse(text)); } catch { return Option.none(); } }); const methodAttrs = (envelope: JsonRpcEnvelope): Record<string, unknown> => { const params = envelope.params ?? {}; switch (envelope.method) { case "initialize": return Option.match(decode(InitializeParams, params), { onNone: () => ({}), onSome: (init) => ({ ...(init.protocolVersion && { "mcp.client.protocol_version": init.protocolVersion }), ...(init.clientInfo?.name && { "mcp.client.name": init.clientInfo.name }), ...(init.clientInfo?.version && { "mcp.client.version": init.clientInfo.version }), ...(init.clientInfo?.title && { "mcp.client.title": init.clientInfo.title }), "mcp.client.capability.keys": Object.keys(init.capabilities ?? {}).sort().join(","), }), }); case "tools/call": return Option.match(decode(NamedParams, params), { onNone: () => ({}), onSome: ({ name }) => (name ? { "mcp.tool.name": name } : {}), }); case "resources/read": case "resources/subscribe": return Option.match(decode(UriParams, params), { onNone: () => ({}), onSome: ({ uri }) => (uri ? { "mcp.resource.uri": uri } : {}), }); case "prompts/get": return Option.match(decode(NamedParams, params), { onNone: () => ({}), onSome: ({ name }) => (name ? { "mcp.prompt.name": name } : {}), }); default: return {}; }}; const replyAttrs = (envelope: JsonRpcEnvelope): Record<string, unknown> => { if (!envelope.result || envelope.method) return {}; return Option.match(decode(ElicitationReplyResult, envelope.result), { onNone: () => ({}), onSome: ({ action }) => (action ? { "mcp.elicitation.action": action } : {}), });}; const rpcAttrs = (envelope: Option.Option<JsonRpcEnvelope>): Record<string, unknown> => Option.match(envelope, { onNone: () => ({}), onSome: (e) => ({ ...(e.method && { "mcp.rpc.method": e.method }), ...(e.id !== undefined && e.id !== null && { "mcp.rpc.id": String(e.id) }), ...methodAttrs(e), ...replyAttrs(e), }), }); const annotateMcpRequest = ( request: Request, opts: { token: VerifiedToken | null; parseBody: boolean },): Effect.Effect<void> => Effect.gen(function* () { const cf = getCfMeta(request); const baseAttrs: Record<string, unknown> = { "mcp.request.method": request.method, "mcp.request.session_id_present": !!request.headers.get("mcp-session-id"), "mcp.request.session_id": request.headers.get("mcp-session-id") ?? "", "mcp.auth.has_bearer": (request.headers.get("authorization") ?? "").startsWith(BEARER_PREFIX), "mcp.auth.verified": !!opts.token, "mcp.auth.organization_id": opts.token?.organizationId ?? "", "mcp.auth.account_id": opts.token?.accountId ?? "", "cf.country": cf.country ?? "", "cf.city": cf.city ?? "", "cf.region": cf.region ?? "", "cf.timezone": cf.timezone ?? "", "cf.asn": cf.asn ?? 0, "cf.as_organization": cf.asOrganization ?? "", "cf.tls_version": cf.tlsVersion ?? "", "cf.tls_cipher": cf.tlsCipher ?? "", "cf.http_protocol": cf.httpProtocol ?? "", "cf.colo": cf.colo ?? "", ...dumpHeaders(request), }; const envelope = opts.parseBody ? yield* readJsonRpcEnvelope(request) : Option.none(); yield* Effect.annotateCurrentSpan({ ...baseAttrs, ...rpcAttrs(envelope), }); }); // ---------------------------------------------------------------------------// OAuth metadata endpoints// --------------------------------------------------------------------------- const protectedResourceMetadata = Effect.sync(() => jsonResponse({ resource: RESOURCE_ORIGIN, authorization_servers: [AUTHKIT_DOMAIN], bearer_methods_supported: ["header"], scopes_supported: [], }),); const authorizationServerMetadata = Effect.promise(async () => { try { const res = await fetch(`${AUTHKIT_DOMAIN}/.well-known/oauth-authorization-server`); if (!res.ok) return jsonResponse({ error: "upstream_error" }, 502); return jsonResponse(await res.json()); } catch { return jsonResponse({ error: "upstream_error" }, 502); }}); // ---------------------------------------------------------------------------// Response-body peek for JSON-RPC error attrs// ---------------------------------------------------------------------------//// The MCP protocol wraps protocol-level failures (malformed envelope, method// not found, invalid params) as `{ error: { code, message } }` in the// JSON-RPC response body with HTTP 200 — none of which surface at the HTTP// layer or as an Effect failure. Same for tool results carrying// `isError: true`. To make those visible in Axiom we peek the first// JSON-RPC message out of the DO's response and stamp it onto the// surrounding `mcp.request` span.//// Only applied to non-streaming response shapes (POST/DELETE). GET hops// onto a long-lived SSE channel we don't want to consume.// --------------------------------------------------------------------------- type SandboxOutcome = { readonly status?: string; readonly error?: { readonly kind?: string; readonly message?: string };}; type JsonRpcErrorBody = { readonly jsonrpc?: string; readonly error?: { readonly code?: number; readonly message?: string }; readonly result?: { readonly isError?: boolean; readonly structuredContent?: SandboxOutcome; };}; const parseFirstJsonRpc = (contentType: string, body: string): JsonRpcErrorBody | null => { if (!body) return null; try { if (contentType.includes("text/event-stream")) { // Grab the first `data:` line from the first SSE event. for (const line of body.split(/\r?\n/)) { if (line.startsWith("data:")) return JSON.parse(line.slice(5).trimStart()); } return null; } if (contentType.includes("application/json")) { return JSON.parse(body); } return null; } catch { return null; }}; const rpcResponseAttrs = (payload: JsonRpcErrorBody | null): Record<string, unknown> => { // Require a JSON-RPC 2.0 envelope so we don't false-positive on other // JSON shapes the edge happens to return (e.g. the auth-failure body // `{ "error": "unauthorized" }` — not a JSON-RPC error). if (!payload || payload.jsonrpc !== "2.0") return {}; const attrs: Record<string, unknown> = {}; const err = payload.error; if (err && typeof err === "object") { attrs["mcp.rpc.is_error"] = true; if (typeof err.code === "number") attrs["mcp.rpc.error.code"] = err.code; if (typeof err.message === "string") { attrs["mcp.rpc.error.message"] = err.message.slice(0, 500); } } if (payload.result?.isError === true) { attrs["mcp.tool.result.is_error"] = true; } const sc = payload.result?.structuredContent; if (sc && typeof sc.status === "string") { attrs["mcp.tool.sandbox.status"] = sc.status; if (sc.error?.kind) attrs["mcp.tool.sandbox.error.kind"] = sc.error.kind; if (typeof sc.error?.message === "string") { attrs["mcp.tool.sandbox.error.message"] = sc.error.message.slice(0, 500); } } return attrs;}; const peekAndAnnotate = (response: Response): Effect.Effect<Response> => Effect.gen(function* () { if (!response.body) return response; // The DO returns a streaming SSE Response (POST responses aren't // `enableJsonResponse`'d in prod), so `response.text()` blocks on the // entire downstream execution — RPC into the dynamic Worker, tool // invocations back to the host, result serialisation. Carving this // into its own span pins "worker drain time" on the trace so you can // tell worker-side waiting apart from any pre/post work on the edge. const text = yield* Effect.promise(() => response.text()).pipe( Effect.withSpan("mcp.peek_response", { attributes: { "http.response.content_type": response.headers.get("content-type") ?? "", "http.response.status_code": response.status, }, }), ); const payload = parseFirstJsonRpc(response.headers.get("content-type") ?? "", text); const attrs = rpcResponseAttrs(payload); if (Object.keys(attrs).length > 0) { yield* Effect.annotateCurrentSpan(attrs); } // Internal-error code -32603 means our server failed handling a // structurally valid request. Unlike -32601 / -32602 ("the client // fucked up"), this is a real bug in our code — route to Sentry so // we get alerted. Protocol-level client errors stay in Axiom. if (payload?.error?.code === -32603) { yield* Effect.sync(() => { const msg = payload.error?.message ?? "unknown"; Sentry.captureException(new Error(`MCP internal error (-32603): ${msg}`)); }); } return new Response(text, { status: response.status, statusText: response.statusText, headers: response.headers, }); }); // ---------------------------------------------------------------------------// DO dispatch// --------------------------------------------------------------------------- // Worker and DO run in separate isolates with independent WebSdk tracer// providers. Neither one can see the other's OTEL context, so the DO used// to emit a brand-new root trace on every stub call. Ferry the worker span// context across with W3C headers: `traceparent` generated from the active// Effect span plus passthrough `tracestate` / `baggage` from the inbound// request.type IncomingPropagationHeaders = { readonly traceparent?: string; readonly tracestate?: string; readonly baggage?: string;}; const currentTraceparent = Effect.map(Effect.currentSpan, (span) => { if (!span || !span.traceId || !span.spanId) return undefined; const flags = span.sampled ? "01" : "00"; return `00-${span.traceId}-${span.spanId}-${flags}`;}).pipe(Effect.orElseSucceed(() => undefined)); const currentPropagationHeaders = ( request: Request,): Effect.Effect<IncomingPropagationHeaders> => Effect.map(currentTraceparent, (traceparent) => ({ traceparent, tracestate: request.headers.get("tracestate") ?? undefined, baggage: request.headers.get("baggage") ?? undefined, })); const withPropagationHeaders = ( request: Request, propagation: IncomingPropagationHeaders,): Request => { const headers = new Headers(request.headers); if (propagation.traceparent) { headers.set("traceparent", propagation.traceparent); } if (propagation.tracestate) { headers.set("tracestate", propagation.tracestate); } if (propagation.baggage) { headers.set("baggage", propagation.baggage); } return new Request(request, { headers });}; /** * Forward a request to an existing session DO. Wrapping the DO's `Response` * with `HttpServerResponse.raw` lets streaming bodies (SSE) pass through * `HttpApp.toWebHandler`'s conversion unchanged.Lines 1–500 of 636. Load more
// --------------------------------------------------------------------------- // Cloud MCP handler — Effect-native HTTP app for /mcp + /.well-known/* // --------------------------------------------------------------------------- // // Built on `@effect/platform`'s `HttpApp.toWebHandler`. start.ts's // mcpRequestMiddleware calls `mcpFetch` and falls through to `next()` when it // returns `null` (non-MCP path) so TanStack Start keeps routing. // // Streaming passthrough — the MCP session Durable Object returns a `Response` // whose body is a `ReadableStream` (SSE). We wrap that `Response` in // `HttpServerResponse.raw(response)`; the platform's `toWeb` conversion // recognises `body.body instanceof Response` and returns it as-is (only // merging headers we set on the outer response, which is none), so the // underlying `ReadableStream` passes through untouched. // --------------------------------------------------------------------------- import { env } from "cloudflare:workers"; import { HttpApp, HttpServerRequest, HttpServerResponse } from "@effect/platform"; import * as Sentry from "@sentry/cloudflare"; import { Context, Effect, Layer, Option, Schema } from "effect"; import { createRemoteJWKSet, jwtVerify } from "jose"; import { TelemetryLive } from "./services/telemetry"; // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- const AUTHKIT_DOMAIN = env.MCP_AUTHKIT_DOMAIN ?? "https://signin.executor.sh"; const RESOURCE_ORIGIN = env.MCP_RESOURCE_ORIGIN ?? "https://executor.sh"; const jwks = createRemoteJWKSet(new URL(`${AUTHKIT_DOMAIN}/oauth2/jwks`)); const BEARER_PREFIX = "Bearer "; const CORS_ALLOW_ORIGIN = { "access-control-allow-origin": "*" } as const; const CORS_PREFLIGHT_HEADERS = { ...CORS_ALLOW_ORIGIN, "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", "access-control-allow-headers": "authorization, content-type, mcp-session-id, accept, mcp-protocol-version", "access-control-expose-headers": "mcp-session-id", } as const; const WWW_AUTHENTICATE = `Bearer resource_metadata="${RESOURCE_ORIGIN}/.well-known/oauth-protected-resource"`; // --------------------------------------------------------------------------- // Response helpers // --------------------------------------------------------------------------- const jsonResponse = (body: unknown, status = 200) => HttpServerResponse.unsafeJson(body, { status, headers: CORS_ALLOW_ORIGIN }); const jsonRpcError = (status: number, code: number, message: string) => HttpServerResponse.unsafeJson({ jsonrpc: "2.0", error: { code, message }, id: null }, { status }); const unauthorized = HttpServerResponse.unsafeJson( { error: "unauthorized" }, { status: 401, headers: { ...CORS_ALLOW_ORIGIN, "www-authenticate": WWW_AUTHENTICATE }, }, ); const corsPreflight = HttpServerResponse.empty({ status: 204, headers: CORS_PREFLIGHT_HEADERS, }); // --------------------------------------------------------------------------- // Auth // --------------------------------------------------------------------------- export type VerifiedToken = { /** The WorkOS account ID (user ID). */ accountId: string; /** The WorkOS organization ID, if the session has org context. */ organizationId: string | null; }; export class McpAuth extends Context.Tag("@executor/cloud/McpAuth")< McpAuth, { readonly verifyBearer: (request: Request) => Effect.Effect<VerifiedToken | null>; } >() {} export const McpAuthLive = Layer.succeed(McpAuth, { verifyBearer: (request) => Effect.promise(async () => { const authHeader = request.headers.get("authorization"); if (!authHeader?.startsWith(BEARER_PREFIX)) return null; try { const { payload } = await jwtVerify(authHeader.slice(BEARER_PREFIX.length), jwks, { issuer: AUTHKIT_DOMAIN, }); if (!payload.sub) return null; return { accountId: payload.sub, organizationId: (payload.org_id as string | undefined) ?? null, }; } catch { return null; } }), }); // --------------------------------------------------------------------------- // Client fingerprint capture // --------------------------------------------------------------------------- // Annotates the Effect span (which nests under the otel-cf-workers fetch // span) with everything we can learn about a connecting MCP client: the // parsed JSON-RPC body, whitelisted request headers, CF request metadata, // and verified-JWT claims. Lets us compare how each client (Claude Code, // Claude.ai web, ChatGPT, custom scripts, ...) actually reports over the // wire. Runs before dispatch so unauthorized requests still get fingerprinted. // --------------------------------------------------------------------------- type CfRequestMetadata = { country?: string; city?: string; region?: string; timezone?: string; asn?: number; asOrganization?: string; tlsVersion?: string; tlsCipher?: string; httpProtocol?: string; colo?: string; }; const getCfMeta = (request: Request): CfRequestMetadata => ((request as unknown as { cf?: CfRequestMetadata }).cf ?? {}) as CfRequestMetadata; const HEADERS_TO_DUMP = [ "accept", "accept-encoding", "accept-language", "cache-control", "content-type", "mcp-protocol-version", "origin", "referer", "sec-fetch-dest", "sec-fetch-mode", "sec-fetch-site", "user-agent", "x-client-name", "x-client-version", "x-requested-with", ] as const; const dumpHeaders = (request: Request): Record<string, string> => { const out: Record<string, string> = {}; for (const name of HEADERS_TO_DUMP) { const value = request.headers.get(name); if (value !== null) out[`mcp.http.header.${name}`] = value; } const authHeader = request.headers.get("authorization"); if (authHeader) { out["mcp.http.header.authorization.scheme"] = authHeader.split(" ", 1)[0] ?? ""; out["mcp.http.header.authorization.length"] = String(authHeader.length); } // Record the full header name list too — surfaces anything unexpected // without us having to enumerate every possibility up front. out["mcp.http.header.names"] = Array.from(request.headers.keys()).sort().join(","); return out; }; // JSON-RPC shapes — narrow to just the fields we fingerprint. Using Schema // collapses the typeof-guard pile and surfaces "what does an MCP client // actually send us" as declarative types. Unknown/malformed input decodes // to None and contributes no span attrs. const UnknownRecord = Schema.Record({ key: Schema.String, value: Schema.Unknown }); const JsonRpcEnvelope = Schema.Struct({ method: Schema.optional(Schema.String), id: Schema.optional(Schema.Union(Schema.String, Schema.Number, Schema.Null)), params: Schema.optional(UnknownRecord), // Responses to server-initiated requests arrive as POST bodies too — // notably elicitation replies (`result.action = "accept" | "decline" | "cancel"`). result: Schema.optional(UnknownRecord), }); type JsonRpcEnvelope = typeof JsonRpcEnvelope.Type; const ElicitationReplyResult = Schema.Struct({ action: Schema.optional(Schema.Literal("accept", "decline", "cancel")), }); const InitializeParams = Schema.Struct({ protocolVersion: Schema.optional(Schema.String), clientInfo: Schema.optional( Schema.Struct({ name: Schema.optional(Schema.String), version: Schema.optional(Schema.String), title: Schema.optional(Schema.String), }), ), capabilities: Schema.optional(UnknownRecord), }); const NamedParams = Schema.Struct({ name: Schema.optional(Schema.String) }); const UriParams = Schema.Struct({ uri: Schema.optional(Schema.String) }); const decode = <A, I>(schema: Schema.Schema<A, I>, input: unknown): Option.Option<A> => Schema.decodeUnknownOption(schema)(input); const readJsonRpcEnvelope = (request: Request): Effect.Effect<Option.Option<JsonRpcEnvelope>> => Effect.promise(async () => { try { const text = await request.clone().text(); if (!text) return Option.none(); return decode(JsonRpcEnvelope, JSON.parse(text)); } catch { return Option.none(); } }); const methodAttrs = (envelope: JsonRpcEnvelope): Record<string, unknown> => { const params = envelope.params ?? {}; switch (envelope.method) { case "initialize": return Option.match(decode(InitializeParams, params), { onNone: () => ({}), onSome: (init) => ({ ...(init.protocolVersion && { "mcp.client.protocol_version": init.protocolVersion }), ...(init.clientInfo?.name && { "mcp.client.name": init.clientInfo.name }), ...(init.clientInfo?.version && { "mcp.client.version": init.clientInfo.version }), ...(init.clientInfo?.title && { "mcp.client.title": init.clientInfo.title }), "mcp.client.capability.keys": Object.keys(init.capabilities ?? {}).sort().join(","), }), }); case "tools/call": return Option.match(decode(NamedParams, params), { onNone: () => ({}), onSome: ({ name }) => (name ? { "mcp.tool.name": name } : {}), }); case "resources/read": case "resources/subscribe": return Option.match(decode(UriParams, params), { onNone: () => ({}), onSome: ({ uri }) => (uri ? { "mcp.resource.uri": uri } : {}), }); case "prompts/get": return Option.match(decode(NamedParams, params), { onNone: () => ({}), onSome: ({ name }) => (name ? { "mcp.prompt.name": name } : {}), }); default: return {}; } }; const replyAttrs = (envelope: JsonRpcEnvelope): Record<string, unknown> => { if (!envelope.result || envelope.method) return {}; return Option.match(decode(ElicitationReplyResult, envelope.result), { onNone: () => ({}), onSome: ({ action }) => (action ? { "mcp.elicitation.action": action } : {}), }); }; const rpcAttrs = (envelope: Option.Option<JsonRpcEnvelope>): Record<string, unknown> => Option.match(envelope, { onNone: () => ({}), onSome: (e) => ({ ...(e.method && { "mcp.rpc.method": e.method }), ...(e.id !== undefined && e.id !== null && { "mcp.rpc.id": String(e.id) }), ...methodAttrs(e), ...replyAttrs(e), }), }); const annotateMcpRequest = ( request: Request, opts: { token: VerifiedToken | null; parseBody: boolean }, ): Effect.Effect<void> => Effect.gen(function* () { const cf = getCfMeta(request); const baseAttrs: Record<string, unknown> = { "mcp.request.method": request.method, "mcp.request.session_id_present": !!request.headers.get("mcp-session-id"), "mcp.request.session_id": request.headers.get("mcp-session-id") ?? "", "mcp.auth.has_bearer": (request.headers.get("authorization") ?? "").startsWith(BEARER_PREFIX), "mcp.auth.verified": !!opts.token, "mcp.auth.organization_id": opts.token?.organizationId ?? "", "mcp.auth.account_id": opts.token?.accountId ?? "", "cf.country": cf.country ?? "", "cf.city": cf.city ?? "", "cf.region": cf.region ?? "", "cf.timezone": cf.timezone ?? "", "cf.asn": cf.asn ?? 0, "cf.as_organization": cf.asOrganization ?? "", "cf.tls_version": cf.tlsVersion ?? "", "cf.tls_cipher": cf.tlsCipher ?? "", "cf.http_protocol": cf.httpProtocol ?? "", "cf.colo": cf.colo ?? "", ...dumpHeaders(request), }; const envelope = opts.parseBody ? yield* readJsonRpcEnvelope(request) : Option.none(); yield* Effect.annotateCurrentSpan({ ...baseAttrs, ...rpcAttrs(envelope), }); }); // --------------------------------------------------------------------------- // OAuth metadata endpoints // --------------------------------------------------------------------------- const protectedResourceMetadata = Effect.sync(() => jsonResponse({ resource: RESOURCE_ORIGIN, authorization_servers: [AUTHKIT_DOMAIN], bearer_methods_supported: ["header"], scopes_supported: [], }), ); const authorizationServerMetadata = Effect.promise(async () => { try { const res = await fetch(`${AUTHKIT_DOMAIN}/.well-known/oauth-authorization-server`); if (!res.ok) return jsonResponse({ error: "upstream_error" }, 502); return jsonResponse(await res.json()); } catch { return jsonResponse({ error: "upstream_error" }, 502); } }); // --------------------------------------------------------------------------- // Response-body peek for JSON-RPC error attrs // --------------------------------------------------------------------------- // // The MCP protocol wraps protocol-level failures (malformed envelope, method // not found, invalid params) as `{ error: { code, message } }` in the // JSON-RPC response body with HTTP 200 — none of which surface at the HTTP // layer or as an Effect failure. Same for tool results carrying // `isError: true`. To make those visible in Axiom we peek the first // JSON-RPC message out of the DO's response and stamp it onto the // surrounding `mcp.request` span. // // Only applied to non-streaming response shapes (POST/DELETE). GET hops // onto a long-lived SSE channel we don't want to consume. // --------------------------------------------------------------------------- type SandboxOutcome = { readonly status?: string; readonly error?: { readonly kind?: string; readonly message?: string }; }; type JsonRpcErrorBody = { readonly jsonrpc?: string; readonly error?: { readonly code?: number; readonly message?: string }; readonly result?: { readonly isError?: boolean; readonly structuredContent?: SandboxOutcome; }; }; const parseFirstJsonRpc = (contentType: string, body: string): JsonRpcErrorBody | null => { if (!body) return null; try { if (contentType.includes("text/event-stream")) { // Grab the first `data:` line from the first SSE event. for (const line of body.split(/\r?\n/)) { if (line.startsWith("data:")) return JSON.parse(line.slice(5).trimStart()); } return null; } if (contentType.includes("application/json")) { return JSON.parse(body); } return null; } catch { return null; } }; const rpcResponseAttrs = (payload: JsonRpcErrorBody | null): Record<string, unknown> => { // Require a JSON-RPC 2.0 envelope so we don't false-positive on other // JSON shapes the edge happens to return (e.g. the auth-failure body // `{ "error": "unauthorized" }` — not a JSON-RPC error). if (!payload || payload.jsonrpc !== "2.0") return {}; const attrs: Record<string, unknown> = {}; const err = payload.error; if (err && typeof err === "object") { attrs["mcp.rpc.is_error"] = true; if (typeof err.code === "number") attrs["mcp.rpc.error.code"] = err.code; if (typeof err.message === "string") { attrs["mcp.rpc.error.message"] = err.message.slice(0, 500); } } if (payload.result?.isError === true) { attrs["mcp.tool.result.is_error"] = true; } const sc = payload.result?.structuredContent; if (sc && typeof sc.status === "string") { attrs["mcp.tool.sandbox.status"] = sc.status; if (sc.error?.kind) attrs["mcp.tool.sandbox.error.kind"] = sc.error.kind; if (typeof sc.error?.message === "string") { attrs["mcp.tool.sandbox.error.message"] = sc.error.message.slice(0, 500); } } return attrs; }; const peekAndAnnotate = (response: Response): Effect.Effect<Response> => Effect.gen(function* () { if (!response.body) return response; // The DO returns a streaming SSE Response (POST responses aren't // `enableJsonResponse`'d in prod), so `response.text()` blocks on the // entire downstream execution — RPC into the dynamic Worker, tool // invocations back to the host, result serialisation. Carving this // into its own span pins "worker drain time" on the trace so you can // tell worker-side waiting apart from any pre/post work on the edge. const text = yield* Effect.promise(() => response.text()).pipe( Effect.withSpan("mcp.peek_response", { attributes: { "http.response.content_type": response.headers.get("content-type") ?? "", "http.response.status_code": response.status, }, }), ); const payload = parseFirstJsonRpc(response.headers.get("content-type") ?? "", text); const attrs = rpcResponseAttrs(payload); if (Object.keys(attrs).length > 0) { yield* Effect.annotateCurrentSpan(attrs); } // Internal-error code -32603 means our server failed handling a // structurally valid request. Unlike -32601 / -32602 ("the client // fucked up"), this is a real bug in our code — route to Sentry so // we get alerted. Protocol-level client errors stay in Axiom. if (payload?.error?.code === -32603) { yield* Effect.sync(() => { const msg = payload.error?.message ?? "unknown"; Sentry.captureException(new Error(`MCP internal error (-32603): ${msg}`)); }); } return new Response(text, { status: response.status, statusText: response.statusText, headers: response.headers, }); }); // --------------------------------------------------------------------------- // DO dispatch // --------------------------------------------------------------------------- // Worker and DO run in separate isolates with independent WebSdk tracer // providers. Neither one can see the other's OTEL context, so the DO used // to emit a brand-new root trace on every stub call. Ferry the worker span // context across with W3C headers: `traceparent` generated from the active // Effect span plus passthrough `tracestate` / `baggage` from the inbound // request. type IncomingPropagationHeaders = { readonly traceparent?: string; readonly tracestate?: string; readonly baggage?: string; }; const currentTraceparent = Effect.map(Effect.currentSpan, (span) => { if (!span || !span.traceId || !span.spanId) return undefined; const flags = span.sampled ? "01" : "00"; return `00-${span.traceId}-${span.spanId}-${flags}`; }).pipe(Effect.orElseSucceed(() => undefined)); const currentPropagationHeaders = ( request: Request, ): Effect.Effect<IncomingPropagationHeaders> => Effect.map(currentTraceparent, (traceparent) => ({ traceparent, tracestate: request.headers.get("tracestate") ?? undefined, baggage: request.headers.get("baggage") ?? undefined, })); const withPropagationHeaders = ( request: Request, propagation: IncomingPropagationHeaders, ): Request => { const headers = new Headers(request.headers); if (propagation.traceparent) { headers.set("traceparent", propagation.traceparent); } if (propagation.tracestate) { headers.set("tracestate", propagation.tracestate); } if (propagation.baggage) { headers.set("baggage", propagation.baggage); } return new Request(request, { headers }); }; /** * Forward a request to an existing session DO. Wrapping the DO's `Response` * with `HttpServerResponse.raw` lets streaming bodies (SSE) pass through * `HttpApp.toWebHandler`'s conversion unchanged.