diag(mcp): comprehensive HTTP logging for OAuth + streamable-HTTP transport

apps/sim/lib/mcp/client.test.ts

@@ -381,6 +381,9 @@ describe('McpClient notification handler', () => {
  {
  authProvider,
  requestInit: { headers: { 'X-Sim-Via': 'workflow' } },
+ // The transport fetch is always wrapped for diagnostics (a no-op passthrough
+ // under test); it defaults to globalThis.fetch when the server isn't pinned.
+ fetch: expect.any(Function),
  }
  )
  })

apps/sim/lib/mcp/client.ts

@@ -12,6 +12,7 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { getMcpSafeErrorDiagnostics } from '@/lib/mcp/error-diagnostics'
+import { withMcpHttpDiagnostics } from '@/lib/mcp/http-diagnostics'
 import { McpOauthRedirectRequired } from '@/lib/mcp/oauth'
 import { createPinnedMcpFetch } from '@/lib/mcp/pinned-fetch'
 import {
@@ -101,7 +102,9 @@ export class McpClient {
  this.transport = new StreamableHTTPClientTransport(new URL(this.config.url), {
  authProvider: useOauth ? this.authProvider : undefined,
  requestInit: { headers: this.config.headers },
- ...(pinned ? { fetch: pinned.fetch } : {}),
+ // Wrap whether pinned or not (SDK's default is globalThis.fetch, so passing it is
+ // behavior-neutral) so the diagnostic also covers unpinned/allowlisted servers.
+ fetch: withMcpHttpDiagnostics(pinned?.fetch ?? globalThis.fetch, 'transport'),
  })
 
  this.client = new Client(

apps/sim/lib/mcp/http-diagnostics.ts

@@ -0,0 +1,176 @@
+import type { FetchLike } from '@modelcontextprotocol/sdk/shared/transport.js'
+import { createLogger } from '@sim/logger'
+import { sanitizeForLogging } from '@/lib/core/security/redaction'
+import { sanitizeUrlForLog } from '@/lib/core/utils/logging'
+import { getMcpSafeErrorDiagnostics } from '@/lib/mcp/error-diagnostics'
+
+const logger = createLogger('McpHttpDiag')
+
+const MAX_LOGGED_CHUNKS = 40
+const PREVIEW_CHARS = 500
+
+/**
+ * TEMPORARY diagnostic for the MCP OAuth + streamable-HTTP transport HTTP layer.
+ * Enabled by default; set `MCP_HTTP_DIAGNOSTICS=false` to silence. Remove once the
+ * Gauge `initialize` hang is root-caused.
+ *
+ * Secret-safety (the wrapped transport fetch also carries in-transport OAuth
+ * refresh/registration, and every tool result):
+ * - Request and OAuth response bodies are NEVER logged.
+ * - The only response body streamed is the one whose REQUEST is an MCP `initialize`
+ * JSON-RPC call — which excludes token/refresh responses AND `tools/call` results
+ * (tool output can be PII/file contents/credentials). The `initialize` result is
+ * protocol metadata only (serverInfo, capabilities), no credentials.
+ * - URLs are logged origin+path only; query strings (`?code=`, `?token=`, …) are
+ * redacted. Sensitive headers (authorization/cookie/www-authenticate) are omitted.
+ */
+function diagnosticsEnabled(): boolean {
+ if (process.env.NODE_ENV === 'test' || process.env.VITEST) return false
+ return process.env.MCP_HTTP_DIAGNOSTICS !== 'false'
+}
+
+function rawUrl(input: string | URL | Request): string {
+ return typeof input === 'string' ? input : input instanceof URL ? input.href : input.url
+}
+
+/**
+ * The `initialize` request is a small fixed structure (protocolVersion, capabilities,
+ * clientInfo). This length gate lets us skip parsing large `tools/call` payloads —
+ * which can be multi-MB — entirely, keeping the check off the hot path.
+ */
+const MAX_INIT_BODY_CHARS = 4096
+
+/**
+ * True only when the request body is an MCP `initialize` JSON-RPC message. Used to
+ * scope response-body logging to the initialize handshake and nothing else — token
+ * requests (form-encoded) and tool calls (`method: 'tools/call'`) both fail this.
+ * Large bodies are rejected by length before any parse (see {@link MAX_INIT_BODY_CHARS}).
+ */
+function isInitializeRequest(body: unknown): boolean {
+ if (typeof body !== 'string' || body.length > MAX_INIT_BODY_CHARS) return false
+ try {
+ return (JSON.parse(body) as { method?: unknown })?.method === 'initialize'
+ } catch {
+ return false
+ }
+}
+
+/**
+ * Wraps a `FetchLike` so every request/response in the MCP OAuth or transport flow is
+ * logged with timing. Only the `initialize` response body is streamed (see secret-safety
+ * note above) — that's the suspected hang.
+ */
+export function withMcpHttpDiagnostics(
+ fetchFn: FetchLike,
+ phase: 'oauth' | 'transport'
+): FetchLike {
+ if (!diagnosticsEnabled()) return fetchFn
+
+ return async (input, init) => {
+ const url = sanitizeUrlForLog(rawUrl(input as string | URL | Request))
+ const method = init?.method ?? 'GET'
+ const reqHeaders = new Headers((init?.headers as HeadersInit | undefined) ?? undefined)
+ const startedAt = Date.now()
+
+ logger.warn('request', {
+ phase,
+ method,
+ url,
+ hasAuth: reqHeaders.has('authorization'),
+ accept: reqHeaders.get('accept') ?? undefined,
+ })
+
+ let res: Response
+ try {
+ res = (await fetchFn(input, init)) as Response
+ } catch (error) {
+ logger.warn('fetch rejected', {
+ phase,
+ method,
+ url,
+ ms: Date.now() - startedAt,
+ error: getMcpSafeErrorDiagnostics(error),
+ })
+ throw error
+ }
+
+ logger.warn('response', {
+ phase,
+ method,
+ url,
+ status: res.status,
+ contentType: res.headers.get('content-type') ?? '',
+ mcpSessionId: res.headers.get('mcp-session-id') ? 'present' : 'absent',
+ headerMs: Date.now() - startedAt,
+ })
+
+ // Stream-log ONLY the initialize handshake response — never OAuth bodies or tool results.
+ if (phase !== 'transport' || !isInitializeRequest(init?.body) || !res.body) return res
+
+ let logBranch: ReadableStream<Uint8Array>
+ let passBranch: ReadableStream<Uint8Array>
+ try {
+ ;[logBranch, passBranch] = res.body.tee()
+ } catch {
+ return res
+ }
+
+ // Cancel the detached log reader when the caller aborts (e.g. the SDK's 30s
+ // initialize timeout — exactly the hang we're tracing) so this tee branch can't
+ // keep the response stream / connection alive after the SDK has given up.
+ const signal = init?.signal
+ void (async () => {
+ const reader = logBranch.getReader()
+ const cancelReader = () => void reader.cancel().catch(() => {})
+ if (signal?.aborted) {
+ cancelReader()
+ return
+ }
+ signal?.addEventListener('abort', cancelReader, { once: true })
+ const decoder = new TextDecoder()
+ let chunks = 0
+ let bytes = 0
+ try {
+ for (;;) {
+ const { done, value } = await reader.read()
+ if (done) {
+ logger.warn('initialize body complete', {
+ url,
+ chunks,
+ bytes,
+ ms: Date.now() - startedAt,
+ })
+ break
+ }
+ chunks += 1
+ bytes += value.byteLength
+ if (chunks <= MAX_LOGGED_CHUNKS) {
+ logger.warn('initialize body chunk', {
+ url,
+ chunk: chunks,
+ size: value.byteLength,
+ ms: Date.now() - startedAt,
+ preview: sanitizeForLogging(decoder.decode(value, { stream: true }), PREVIEW_CHARS),
+ })
+ }
+ }
+ } catch (error) {
+ logger.warn('initialize body read error', {
+ url,
+ chunks,
+ bytes,
+ ms: Date.now() - startedAt,
+ error: getMcpSafeErrorDiagnostics(error),
+ })
+ } finally {
+ signal?.removeEventListener('abort', cancelReader)
+ }
+ })()
+
+ return new Response(passBranch, {
+ status: res.status,
+ statusText: res.statusText,
+ headers: res.headers,
+ })
+ }
+}

apps/sim/lib/mcp/oauth/auth.ts

@@ -1,4 +1,5 @@
 import { auth, type OAuthClientProvider } from '@modelcontextprotocol/sdk/client/auth.js'
+import { withMcpHttpDiagnostics } from '@/lib/mcp/http-diagnostics'
 import { createSsrfGuardedMcpFetch } from '@/lib/mcp/pinned-fetch'
 
 type McpAuthOptions = Parameters<typeof auth>[1]
@@ -15,5 +16,8 @@ export function mcpAuthGuarded(
  provider: OAuthClientProvider,
  options: McpAuthOptions
 ): ReturnType<typeof auth> {
- return auth(provider, { ...options, fetchFn: options.fetchFn ?? createSsrfGuardedMcpFetch() })
+ return auth(provider, {
+ ...options,
+ fetchFn: withMcpHttpDiagnostics(options.fetchFn ?? createSsrfGuardedMcpFetch(), 'oauth'),
+ })
 }