From 1b76b60b173326d17e90c857e38fb79cdc2f8ac8 Mon Sep 17 00:00:00 2001 From: can1357 Date: Fri, 15 May 2026 09:07:34 +0200 Subject: [PATCH] feat(ai/providers): added configurable fetch overrides to AI provider request paths - Introduced a `fetch` option on `StreamOptions` and threaded it through providers to let callers supply a custom request transport. - Updated provider clients and direct HTTP calls across Anthropic, OpenAI, Azure, Google, GitLab Duo, Gemini CLI, Ollama, and Codex flows to use the injected fetch implementation. - Extended retry helper options to accept a fetch override and preserved preconnect support from the selected fetch function. --- packages/ai/src/providers/anthropic.ts | 9 ++++- .../src/providers/azure-openai-responses.ts | 5 ++- packages/ai/src/providers/gitlab-duo.ts | 12 +++++-- .../ai/src/providers/google-gemini-cli.ts | 3 +- packages/ai/src/providers/google-vertex.ts | 36 ++++++++++++++----- packages/ai/src/providers/google.ts | 18 ++++++++-- packages/ai/src/providers/ollama.ts | 1 + .../ai/src/providers/openai-anthropic-shim.ts | 2 ++ .../src/providers/openai-codex-responses.ts | 3 ++ .../ai/src/providers/openai-completions.ts | 7 ++-- packages/ai/src/providers/openai-responses.ts | 5 ++- packages/ai/src/types.ts | 8 +++++ packages/utils/src/fetch-retry.ts | 9 ++++- 13 files changed, 96 insertions(+), 22 deletions(-) diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 58d56b1a1..9021d207b 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -541,6 +541,7 @@ export type AnthropicClientOptionsArgs = { isOAuth?: boolean; hasTools?: boolean; onSseEvent?: AnthropicOptions["onSseEvent"]; + fetch?: typeof fetch; }; export type AnthropicClientOptionsResult = { @@ -965,6 +966,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( isOAuth: options?.isOAuth, hasTools: !!context.tools?.length, onSseEvent: options?.onSseEvent, + fetch: options?.fetch, }); client = created.client; isOAuthToken = created.isOAuthToken; @@ -1405,7 +1407,12 @@ export function buildAnthropicClientOptions(args: AnthropicClientOptionsArgs): A const baseUrl = resolveAnthropicBaseUrl(model, apiKey); const foundryCustomHeaders = resolveAnthropicCustomHeaders(model); const tlsFetchOptions = buildClaudeCodeTlsFetchOptions(model, baseUrl); - const debugFetch = onSseEvent ? wrapFetchForSseDebug(fetch, event => onSseEvent(event, model)) : undefined; + const baseFetch = args.fetch ?? fetch; + const debugFetch = onSseEvent + ? wrapFetchForSseDebug(baseFetch, event => onSseEvent(event, model)) + : args.fetch + ? baseFetch + : undefined; if (model.provider === "github-copilot") { const copilotApiKey = parseGitHubCopilotApiKey(apiKey).accessToken; const betaFeatures = [...extraBetas]; diff --git a/packages/ai/src/providers/azure-openai-responses.ts b/packages/ai/src/providers/azure-openai-responses.ts index d5f34a45f..bec81251d 100644 --- a/packages/ai/src/providers/azure-openai-responses.ts +++ b/packages/ai/src/providers/azure-openai-responses.ts @@ -241,6 +241,7 @@ function createClient(model: Model<"azure-openai-responses">, apiKey: string, op const { baseUrl, apiVersion } = resolveAzureConfig(model, options); + const baseFetch = options?.fetch ?? fetch; return new AzureOpenAI({ apiKey, apiVersion, @@ -248,7 +249,9 @@ function createClient(model: Model<"azure-openai-responses">, apiKey: string, op maxRetries: 5, defaultHeaders: headers, baseURL: baseUrl, - fetch: options?.onSseEvent ? wrapFetchForSseDebug(fetch, event => options.onSseEvent?.(event, model)) : fetch, + fetch: options?.onSseEvent + ? wrapFetchForSseDebug(baseFetch, event => options.onSseEvent?.(event, model)) + : baseFetch, }); } diff --git a/packages/ai/src/providers/gitlab-duo.ts b/packages/ai/src/providers/gitlab-duo.ts index 8efcda7ec..fbc5bcff4 100644 --- a/packages/ai/src/providers/gitlab-duo.ts +++ b/packages/ai/src/providers/gitlab-duo.ts @@ -172,13 +172,16 @@ interface DirectAccessToken { const directAccessCache = new Map(); -async function getDirectAccessToken(gitlabAccessToken: string): Promise { +async function getDirectAccessToken( + gitlabAccessToken: string, + fetchImpl: typeof fetch = fetch, +): Promise { const cached = directAccessCache.get(gitlabAccessToken); if (cached && cached.expiresAt > Date.now()) { return cached; } - const response = await fetch(`${GITLAB_COM_URL}/api/v4/ai/third_party_agents/direct_access`, { + const response = await fetchImpl(`${GITLAB_COM_URL}/api/v4/ai/third_party_agents/direct_access`, { method: "POST", headers: { Authorization: `Bearer ${gitlabAccessToken}`, @@ -240,7 +243,7 @@ export function streamGitLabDuo( throw new Error(`Unsupported GitLab Duo model: ${model.id}`); } - const directAccess = await getDirectAccessToken(options.apiKey); + const directAccess = await getDirectAccessToken(options.apiKey, options.fetch); const headers = { ...directAccess.headers, ...options.headers, @@ -278,6 +281,7 @@ export function streamGitLabDuo( onPayload: options.onPayload, onResponse: options.onResponse, onSseEvent: options.onSseEvent, + fetch: options.fetch, thinkingEnabled: Boolean(reasoningEffort) && model.reasoning, thinkingBudgetTokens: reasoningEffort ? (options.thinkingBudgets?.[reasoningEffort] ?? ANTHROPIC_THINKING[reasoningEffort]) @@ -314,6 +318,7 @@ export function streamGitLabDuo( onPayload: options.onPayload, onResponse: options.onResponse, onSseEvent: options.onSseEvent, + fetch: options.fetch, reasoning: reasoningEffort, toolChoice: options.toolChoice, } satisfies OpenAIResponsesOptions, @@ -345,6 +350,7 @@ export function streamGitLabDuo( onPayload: options.onPayload, onResponse: options.onResponse, onSseEvent: options.onSseEvent, + fetch: options.fetch, reasoning: reasoningEffort, toolChoice: options.toolChoice, } satisfies OpenAICompletionsOptions, diff --git a/packages/ai/src/providers/google-gemini-cli.ts b/packages/ai/src/providers/google-gemini-cli.ts index 3e75d286a..c8fe2ea3f 100644 --- a/packages/ai/src/providers/google-gemini-cli.ts +++ b/packages/ai/src/providers/google-gemini-cli.ts @@ -362,6 +362,7 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( maxAttempts: MAX_RETRIES + 1, defaultDelayMs: attempt => BASE_DELAY_MS * 2 ** attempt, maxDelayMs: options?.maxRetryDelayMs ?? RATE_LIMIT_BUDGET_MS, + fetch: options?.fetch, }, ); if (!response.ok) { @@ -545,7 +546,7 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( throw new Error("Missing request URL"); } - currentResponse = await fetch(requestUrl, { + currentResponse = await (options?.fetch ?? fetch)(requestUrl, { method: "POST", headers: requestHeaders, body: requestBodyJson, diff --git a/packages/ai/src/providers/google-vertex.ts b/packages/ai/src/providers/google-vertex.ts index 47a72037b..d2f05b4e5 100644 --- a/packages/ai/src/providers/google-vertex.ts +++ b/packages/ai/src/providers/google-vertex.ts @@ -25,7 +25,9 @@ export const streamGoogleVertex: StreamFunction<"google-vertex"> = ( const apiKey = resolveApiKey(options); const project = apiKey ? undefined : resolveProject(options); const location = apiKey ? undefined : resolveLocation(options); - const client = apiKey ? createClientWithApiKey(model, apiKey) : createClient(model, project!, location!); + const client = apiKey + ? createClientWithApiKey(model, apiKey, options?.fetch) + : createClient(model, project!, location!, options?.fetch); const params = buildGoogleGenerateContentParams(model, context, options ?? {}); const url = apiKey ? `https://aiplatform.googleapis.com/${API_VERSION}/publishers/google/models/${model.id}:streamGenerateContent` @@ -34,29 +36,45 @@ export const streamGoogleVertex: StreamFunction<"google-vertex"> = ( }, }); -function buildHttpOptions(model: Model<"google-vertex">): { headers?: Record } | undefined { - if (!model.headers) { - return undefined; +function buildHttpOptions( + model: Model<"google-vertex">, + fetchOverride: typeof fetch | undefined, +): { headers?: Record; fetch?: typeof fetch } | undefined { + const options: { headers?: Record; fetch?: typeof fetch } = {}; + if (model.headers) { + options.headers = { ...model.headers }; } - return { headers: { ...model.headers } }; + if (fetchOverride) { + options.fetch = fetchOverride; + } + return Object.keys(options).length > 0 ? options : undefined; } -function createClient(model: Model<"google-vertex">, project: string, location: string): GoogleGenAI { +function createClient( + model: Model<"google-vertex">, + project: string, + location: string, + fetchOverride: typeof fetch | undefined, +): GoogleGenAI { return new GoogleGenAI({ vertexai: true, project, location, apiVersion: API_VERSION, - httpOptions: buildHttpOptions(model), + httpOptions: buildHttpOptions(model, fetchOverride), }); } -function createClientWithApiKey(model: Model<"google-vertex">, apiKey: string): GoogleGenAI { +function createClientWithApiKey( + model: Model<"google-vertex">, + apiKey: string, + fetchOverride: typeof fetch | undefined, +): GoogleGenAI { return new GoogleGenAI({ vertexai: true, apiKey, apiVersion: API_VERSION, - httpOptions: buildHttpOptions(model), + httpOptions: buildHttpOptions(model, fetchOverride), }); } diff --git a/packages/ai/src/providers/google.ts b/packages/ai/src/providers/google.ts index 48e571564..a7d0ac028 100644 --- a/packages/ai/src/providers/google.ts +++ b/packages/ai/src/providers/google.ts @@ -17,15 +17,24 @@ export const streamGoogle: StreamFunction<"google-generative-ai"> = ( api: "google-generative-ai", prepare: () => { const apiKey = options?.apiKey || getEnvApiKey(model.provider); - const client = createClient(model, apiKey); + const client = createClient(model, apiKey, options?.fetch); const params = buildGoogleGenerateContentParams(model, context, options ?? {}); const url = model.baseUrl ? `${model.baseUrl}/models/${model.id}:streamGenerateContent` : undefined; return { client, params, url }; }, }); -function createClient(model: Model<"google-generative-ai">, apiKey?: string): GoogleGenAI { - const httpOptions: { baseUrl?: string; apiVersion?: string; headers?: Record } = {}; +function createClient( + model: Model<"google-generative-ai">, + apiKey?: string, + fetchOverride?: typeof fetch, +): GoogleGenAI { + const httpOptions: { + baseUrl?: string; + apiVersion?: string; + headers?: Record; + fetch?: typeof fetch; + } = {}; if (model.baseUrl) { httpOptions.baseUrl = model.baseUrl; httpOptions.apiVersion = ""; // baseUrl already includes version path, don't append @@ -33,6 +42,9 @@ function createClient(model: Model<"google-generative-ai">, apiKey?: string): Go if (model.headers) { httpOptions.headers = model.headers; } + if (fetchOverride) { + httpOptions.fetch = fetchOverride; + } return new GoogleGenAI({ apiKey, diff --git a/packages/ai/src/providers/ollama.ts b/packages/ai/src/providers/ollama.ts index b4f987f53..ddba1dd5f 100644 --- a/packages/ai/src/providers/ollama.ts +++ b/packages/ai/src/providers/ollama.ts @@ -378,6 +378,7 @@ export const streamOllama: StreamFunction<"ollama-chat"> = ( body: JSON.stringify(body), signal: options.signal, defaultDelayMs: OLLAMA_RETRY_DELAYS_MS, + fetch: options.fetch, }); if (!response.ok) { throw new Error(`HTTP ${response.status} from ${baseUrl}/api/chat`); diff --git a/packages/ai/src/providers/openai-anthropic-shim.ts b/packages/ai/src/providers/openai-anthropic-shim.ts index 23ae7fe4d..416fabb1d 100644 --- a/packages/ai/src/providers/openai-anthropic-shim.ts +++ b/packages/ai/src/providers/openai-anthropic-shim.ts @@ -88,6 +88,7 @@ export function streamOpenAIAnthropicShim( onPayload: options?.onPayload, onResponse: options?.onResponse, onSseEvent: options?.onSseEvent, + fetch: options?.fetch, thinkingEnabled, thinkingBudgetTokens: thinkingBudget, }); @@ -116,6 +117,7 @@ export function streamOpenAIAnthropicShim( onPayload: options?.onPayload, onResponse: options?.onResponse, onSseEvent: options?.onSseEvent, + fetch: options?.fetch, reasoning: reasoningEffort, }); diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index f2fb641c4..b89427fea 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -735,6 +735,7 @@ async function openCodexSseTransport( state, requestSetup.requestSignal, event => options?.onSseEvent?.(event, model), + options?.fetch, ), ); return { eventStream, requestBodyForState: structuredCloneJSON(body), transport: "sse" }; @@ -2173,6 +2174,7 @@ async function openCodexSseEventStream( state: CodexWebSocketSessionState | undefined, signal?: AbortSignal, onSseEvent?: OpenAICodexResponsesOptions["onSseEvent"], + fetchOverride?: typeof fetch, ): Promise>> { const headers = createCodexHeaders(requestHeaders, accountId, apiKey, sessionId, "sse", state); logCodexDebug("codex request", { @@ -2190,6 +2192,7 @@ async function openCodexSseEventStream( maxAttempts: CODEX_MAX_RETRIES + 1, defaultDelayMs: attempt => CODEX_RETRY_DELAY_MS * (attempt + 1), maxDelayMs: CODEX_RATE_LIMIT_BUDGET_MS, + fetch: fetchOverride, }); logCodexDebug("codex response", { url: response.url, diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index e878e4f1f..5348a64ff 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -362,6 +362,7 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( options?.headers, options?.initiatorOverride, options?.onSseEvent, + options?.fetch, ); const priorityPremiumRequests = getPriorityPremiumRequests(options?.serviceTier, model.provider); const premiumRequestsTotal = @@ -778,6 +779,7 @@ async function createClient( extraHeaders?: Record, initiatorOverride?: MessageAttribution, onSseEvent?: OpenAICompletionsOptions["onSseEvent"], + fetchOverride?: typeof fetch, ): Promise<{ client: OpenAI; copilotPremiumRequests: number | undefined; @@ -847,9 +849,10 @@ async function createClient( azureDefaultQuery = { "api-version": apiVersion }; } let capturedErrorResponse: CapturedHttpErrorResponse | undefined; + const baseFetch = fetchOverride ?? fetch; const wrappedFetch = Object.assign( async (input: string | URL | Request, init?: RequestInit): Promise => { - const response = await fetch(input, init); + const response = await baseFetch(input, init); if (response.ok) { capturedErrorResponse = undefined; return response; @@ -872,7 +875,7 @@ async function createClient( }; return response; }, - { preconnect: fetch.preconnect }, + baseFetch.preconnect ? { preconnect: baseFetch.preconnect } : {}, ); const debugFetch = onSseEvent ? wrapFetchForSseDebug(wrappedFetch, event => onSseEvent(event, model)) : wrappedFetch; return { diff --git a/packages/ai/src/providers/openai-responses.ts b/packages/ai/src/providers/openai-responses.ts index 4f212adb2..292bd89c8 100644 --- a/packages/ai/src/providers/openai-responses.ts +++ b/packages/ai/src/providers/openai-responses.ts @@ -210,6 +210,7 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = ( options?.initiatorOverride, cacheSessionId, options?.onSseEvent, + options?.fetch, ); const priorityPremiumRequests = getPriorityPremiumRequests(options?.serviceTier, model.provider); const premiumRequestsTotal = @@ -312,6 +313,7 @@ function createClient( initiatorOverride?: MessageAttribution, sessionId?: string, onSseEvent?: OpenAIResponsesOptions["onSseEvent"], + fetchOverride?: typeof fetch, ): { client: OpenAI; copilotPremiumRequests: number | undefined; @@ -349,6 +351,7 @@ function createClient( headers.session_id ??= sessionId; headers["x-client-request-id"] ??= sessionId; } + const baseFetch = fetchOverride ?? fetch; return { client: new OpenAI({ apiKey, @@ -356,7 +359,7 @@ function createClient( dangerouslyAllowBrowser: true, maxRetries: 5, defaultHeaders: headers, - fetch: onSseEvent ? wrapFetchForSseDebug(fetch, event => onSseEvent(event, model)) : fetch, + fetch: onSseEvent ? wrapFetchForSseDebug(baseFetch, event => onSseEvent(event, model)) : baseFetch, }), copilotPremiumRequests, baseUrl, diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index b2ef0871c..409979f75 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -275,6 +275,14 @@ export interface StreamOptions { * Set to 0 to disable the inter-event idle watchdog for this request. */ streamIdleTimeoutMs?: number; + /** + * Optional `fetch` implementation override. Providers route every HTTP + * request — direct calls, SDK clients, and retry helpers — through this + * implementation when set. Defaults to `globalThis.fetch`. Providers that + * do not use `fetch` (Bedrock's AWS SDK transport, Cursor's HTTP/2 + * channel) silently ignore the override. + */ + fetch?: typeof fetch; /** Cursor exec/MCP tool handlers (cursor-agent only). */ execHandlers?: CursorExecHandlers; } diff --git a/packages/utils/src/fetch-retry.ts b/packages/utils/src/fetch-retry.ts index 35814eedc..1be105c99 100644 --- a/packages/utils/src/fetch-retry.ts +++ b/packages/utils/src/fetch-retry.ts @@ -96,6 +96,12 @@ export interface FetchWithRetryOptions extends RequestInit { * token refresh or user-agent rotation. */ prepareInit?: (attempt: number) => RequestInit | Promise; + /** + * Optional `fetch` implementation override. Defaults to `globalThis.fetch`. + * Useful for routing requests through a proxy, instrumented transport, or + * mock during tests. + */ + fetch?: typeof fetch; } const DEFAULT_MAX_DELAY_MS = 60_000; @@ -119,6 +125,7 @@ export async function fetchWithRetry( maxDelayMs = DEFAULT_MAX_DELAY_MS, defaultDelayMs, prepareInit, + fetch: fetchImpl = fetch, ...baseInit } = options; const signal = baseInit.signal as AbortSignal | undefined; @@ -130,7 +137,7 @@ export async function fetchWithRetry( let response: Response; try { - response = await fetch(requestUrl, init); + response = await fetchImpl(requestUrl, init); } catch (error) { if (signal?.aborted) throw new Error("Request was aborted"); const wrapped = wrapNetworkError(error);