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.
This commit is contained in:
can1357
2026-05-15 09:07:34 +02:00
parent d436555419
commit 1b76b60b17
13 changed files with 96 additions and 22 deletions
+8 -1
View File
@@ -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];
@@ -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,
});
}
+9 -3
View File
@@ -172,13 +172,16 @@ interface DirectAccessToken {
const directAccessCache = new Map<string, DirectAccessToken>();
async function getDirectAccessToken(gitlabAccessToken: string): Promise<DirectAccessToken> {
async function getDirectAccessToken(
gitlabAccessToken: string,
fetchImpl: typeof fetch = fetch,
): Promise<DirectAccessToken> {
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,
@@ -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,
+27 -9
View File
@@ -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<string, string> } | undefined {
if (!model.headers) {
return undefined;
function buildHttpOptions(
model: Model<"google-vertex">,
fetchOverride: typeof fetch | undefined,
): { headers?: Record<string, string>; fetch?: typeof fetch } | undefined {
const options: { headers?: Record<string, string>; 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),
});
}
+15 -3
View File
@@ -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<string, string> } = {};
function createClient(
model: Model<"google-generative-ai">,
apiKey?: string,
fetchOverride?: typeof fetch,
): GoogleGenAI {
const httpOptions: {
baseUrl?: string;
apiVersion?: string;
headers?: Record<string, string>;
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,
+1
View File
@@ -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`);
@@ -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,
});
@@ -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<AsyncGenerator<Record<string, unknown>>> {
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,
@@ -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<string, string>,
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<Response> => {
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 {
@@ -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,
+8
View File
@@ -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;
}
+8 -1
View File
@@ -96,6 +96,12 @@ export interface FetchWithRetryOptions extends RequestInit {
* token refresh or user-agent rotation.
*/
prepareInit?: (attempt: number) => RequestInit | Promise<RequestInit>;
/**
* 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);