feat: enhance upstream endpoint handling with new passthrough header extraction and response management
- Introduced new types for endpoint preferences to support 'responses'. - Added functions to extract safe passthrough headers, Claude-specific headers, and responses-specific headers. - Implemented logic to ensure proper handling of stream accept headers. - Updated endpoint resolution logic to prioritize 'responses' in certain contexts. - Enhanced request building for 'messages' and 'responses' endpoints with improved body sanitization and header management. - Added comprehensive documentation for new features and usage instructions.
This commit is contained in:
cita
@@ -338,6 +338,155 @@ describe('chat proxy stream behavior', () => {
|
||||
expect(response.body).toContain('event: message_stop');
|
||||
});
|
||||
|
||||
it('converts OpenAI tool_calls SSE into Claude tool_use stream events on /v1/messages', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-tool","model":"claude-opus-4-6","choices":[{"delta":{"role":"assistant"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-tool","model":"claude-opus-4-6","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_glob","type":"function","function":{"name":"Glob","arguments":""}}]},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-tool","model":"claude-opus-4-6","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"pattern\\":\\"README*\\"}"}}]},"finish_reason":"tool_calls"}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
stream: true,
|
||||
max_tokens: 256,
|
||||
messages: [{ role: 'user', content: 'find readme files' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.body).toContain('event: content_block_start');
|
||||
expect(response.body).toContain('"type":"tool_use"');
|
||||
expect(response.body).toContain('"name":"Glob"');
|
||||
expect(response.body).toContain('"type":"input_json_delta"');
|
||||
expect(response.body).toContain('"partial_json":"{\\"pattern\\":\\"README*\\"}"');
|
||||
expect(response.body).toContain('"stop_reason":"tool_use"');
|
||||
});
|
||||
|
||||
it('keeps final content when upstream chunk carries both delta and finish_reason on /v1/messages', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-2","model":"claude-opus-4-6","choices":[{"delta":{"role":"assistant"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-2","model":"claude-opus-4-6","choices":[{"delta":{"content":"tail-token"},"finish_reason":"stop"}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
stream: true,
|
||||
max_tokens: 256,
|
||||
messages: [{ role: 'user', content: 'hello' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.body).toContain('event: content_block_delta');
|
||||
expect(response.body).toContain('\"text\":\"tail-token\"');
|
||||
expect(response.body).toContain('event: message_stop');
|
||||
});
|
||||
|
||||
it('preserves Claude-specific payload fields and forwards claude headers on /v1/messages', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'msg_headers',
|
||||
type: 'message',
|
||||
model: 'claude-opus-4-6',
|
||||
content: [{ type: 'text', text: 'ok' }],
|
||||
stop_reason: 'end_turn',
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
headers: {
|
||||
'anthropic-beta': 'code-2025-09-30',
|
||||
'x-claude-client': 'claude-code',
|
||||
},
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
max_tokens: 256,
|
||||
messages: [{ role: 'user', content: 'hello' }],
|
||||
metadata: { session_id: 'abc123' },
|
||||
thinking: { type: 'enabled', budget_tokens: 1024 },
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
|
||||
const [_targetUrl, options] = fetchMock.mock.calls[0] as [string, any];
|
||||
expect(options.headers['anthropic-beta']).toBe('code-2025-09-30');
|
||||
expect(options.headers['x-claude-client']).toBe('claude-code');
|
||||
|
||||
const forwardedBody = JSON.parse(options.body);
|
||||
expect(forwardedBody.metadata).toEqual({ session_id: 'abc123' });
|
||||
expect(forwardedBody.thinking).toEqual({ type: 'enabled', budget_tokens: 1024 });
|
||||
});
|
||||
|
||||
it('passes through Claude tool_use SSE events on /v1/messages for CLI tool execution', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('event: message_start\ndata: {"type":"message_start","message":{"id":"msg_tool_1","model":"claude-opus-4-6","role":"assistant","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":1,"output_tokens":0}}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_1","name":"Glob","input":{}}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\\"pattern\\":\\"README*\\"}"}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: content_block_stop\ndata: {"type":"content_block_stop","index":0}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":6}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
stream: true,
|
||||
max_tokens: 256,
|
||||
messages: [{ role: 'user', content: 'find readme files' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.body).toContain('event: content_block_start');
|
||||
expect(response.body).toContain('"type":"tool_use"');
|
||||
expect(response.body).toContain('"name":"Glob"');
|
||||
expect(response.body).toContain('"partial_json":"{\\"pattern\\":\\"README*\\"}"');
|
||||
expect(response.body).toContain('event: message_stop');
|
||||
});
|
||||
|
||||
it('serves /v1/responses via protocol translation when upstream is OpenAI-compatible', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'resp_123',
|
||||
@@ -365,9 +514,10 @@ describe('chat proxy stream behavior', () => {
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
const [targetUrl, options] = fetchMock.mock.calls[0] as [string, any];
|
||||
expect(targetUrl).toContain('/v1/chat/completions');
|
||||
expect(targetUrl).toContain('/v1/responses');
|
||||
const forwarded = JSON.parse(options.body);
|
||||
expect(forwarded.model).toBe('upstream-gpt');
|
||||
expect(forwarded.input).toBe('hello');
|
||||
});
|
||||
|
||||
it('passes through /v1/responses SSE payloads', async () => {
|
||||
@@ -401,6 +551,168 @@ describe('chat proxy stream behavior', () => {
|
||||
expect(response.body).toContain('[DONE]');
|
||||
});
|
||||
|
||||
it('converts chat-completions SSE to Responses stream events for /v1/responses clients', async () => {
|
||||
fetchModelPricingCatalogMock.mockResolvedValue({
|
||||
models: [
|
||||
{
|
||||
modelName: 'upstream-gpt',
|
||||
supportedEndpointTypes: ['/v1/chat/completions'],
|
||||
},
|
||||
],
|
||||
groupRatio: {},
|
||||
});
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-r1","model":"upstream-gpt","choices":[{"delta":{"role":"assistant"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-r1","model":"upstream-gpt","choices":[{"delta":{"content":"hello"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-r1","model":"upstream-gpt","choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
payload: {
|
||||
model: 'gpt-5.2',
|
||||
input: 'hello',
|
||||
stream: true,
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.body).toContain('response.output_item.added');
|
||||
expect(response.body).toContain('response.output_text.delta');
|
||||
expect(response.body).toContain('response.completed');
|
||||
expect(response.body).toContain('[DONE]');
|
||||
});
|
||||
|
||||
it('converts chat tool_calls SSE to Responses function_call events on /v1/responses', async () => {
|
||||
fetchModelPricingCatalogMock.mockResolvedValue({
|
||||
models: [
|
||||
{
|
||||
modelName: 'upstream-gpt',
|
||||
supportedEndpointTypes: ['/v1/chat/completions'],
|
||||
},
|
||||
],
|
||||
groupRatio: {},
|
||||
});
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-r2","model":"upstream-gpt","choices":[{"delta":{"role":"assistant"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-r2","model":"upstream-gpt","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_abc","type":"function","function":{"name":"Glob","arguments":""}}]},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-r2","model":"upstream-gpt","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"pattern\\":\\"README*\\"}"}}]},"finish_reason":"tool_calls"}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
payload: {
|
||||
model: 'gpt-5.2',
|
||||
input: 'find readme',
|
||||
stream: true,
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.body).toContain('"type":"function_call"');
|
||||
expect(response.body).toContain('response.function_call_arguments.delta');
|
||||
expect(response.body).toContain('"name":"Glob"');
|
||||
expect(response.body).toContain('"delta":"{\\"pattern\\":\\"README*\\"}"');
|
||||
expect(response.body).toContain('response.completed');
|
||||
});
|
||||
|
||||
it('preserves Responses-specific payload fields and forwards openai headers on /v1/responses', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'resp_passthrough',
|
||||
object: 'response',
|
||||
status: 'completed',
|
||||
model: 'upstream-gpt',
|
||||
output_text: 'ok',
|
||||
output: [],
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
headers: {
|
||||
'openai-beta': 'responses-2025-03-11',
|
||||
'x-stainless-lang': 'typescript',
|
||||
originator: 'codex_cli_rs',
|
||||
},
|
||||
payload: {
|
||||
model: 'gpt-5.2',
|
||||
input: 'hello',
|
||||
metadata: { session_id: 'abc123' },
|
||||
reasoning: { effort: 'high' },
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
|
||||
const [_targetUrl, options] = fetchMock.mock.calls[0] as [string, any];
|
||||
expect(options.headers['openai-beta']).toBe('responses-2025-03-11');
|
||||
expect(options.headers['x-stainless-lang']).toBe('typescript');
|
||||
expect(options.headers.originator).toBe('codex_cli_rs');
|
||||
|
||||
const forwardedBody = JSON.parse(options.body);
|
||||
expect(forwardedBody.metadata).toEqual({ session_id: 'abc123' });
|
||||
expect(forwardedBody.reasoning).toEqual({ effort: 'high' });
|
||||
});
|
||||
|
||||
it('sets stream accept header for /v1/responses when downstream omits accept', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('event: response.completed\ndata: {"type":"response.completed","response":{"id":"resp_accept","model":"upstream-gpt","status":"completed","usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
payload: {
|
||||
model: 'gpt-5.2',
|
||||
input: 'hello',
|
||||
stream: true,
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
|
||||
const [_targetUrl, options] = fetchMock.mock.calls[0] as [string, any];
|
||||
expect(options.headers.accept).toBe('text/event-stream');
|
||||
});
|
||||
|
||||
it('routes /v1/responses to /v1/messages when upstream catalog is anthropic-only', async () => {
|
||||
fetchModelPricingCatalogMock.mockResolvedValue({
|
||||
models: [
|
||||
|
||||
@@ -29,11 +29,72 @@ import {
|
||||
} from './upstreamEndpoint.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
const CLAUDE_SSE_EVENT_NAMES = new Set([
|
||||
'message_start',
|
||||
'content_block_start',
|
||||
'content_block_delta',
|
||||
'content_block_stop',
|
||||
'message_delta',
|
||||
'message_stop',
|
||||
'ping',
|
||||
'error',
|
||||
]);
|
||||
|
||||
function withUpstreamPath(path: string, message: string): string {
|
||||
return `[upstream:${path}] ${message}`;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object';
|
||||
}
|
||||
|
||||
function isClaudeSseEventName(value: unknown): value is string {
|
||||
return typeof value === 'string' && CLAUDE_SSE_EVENT_NAMES.has(value);
|
||||
}
|
||||
|
||||
function serializeRawSseEvent(event: string, data: string): string {
|
||||
const dataLines = data.split('\n').map((line) => `data: ${line}`).join('\n');
|
||||
if (event) {
|
||||
return `event: ${event}\n${dataLines}\n\n`;
|
||||
}
|
||||
return `${dataLines}\n\n`;
|
||||
}
|
||||
|
||||
function syncClaudeStreamStateFromRawEvent(
|
||||
eventName: string,
|
||||
parsedPayload: unknown,
|
||||
streamContext: { id: string; model: string },
|
||||
claudeContext: { messageStarted: boolean; contentBlockStarted: boolean; doneSent: boolean },
|
||||
) {
|
||||
if (eventName === 'message_start') {
|
||||
claudeContext.messageStarted = true;
|
||||
if (isRecord(parsedPayload) && isRecord(parsedPayload.message)) {
|
||||
const message = parsedPayload.message;
|
||||
if (typeof message.id === 'string' && message.id.trim().length > 0) {
|
||||
streamContext.id = message.id;
|
||||
}
|
||||
if (typeof message.model === 'string' && message.model.trim().length > 0) {
|
||||
streamContext.model = message.model;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (eventName === 'content_block_start') {
|
||||
claudeContext.contentBlockStarted = true;
|
||||
return;
|
||||
}
|
||||
|
||||
if (eventName === 'content_block_stop') {
|
||||
claudeContext.contentBlockStarted = false;
|
||||
return;
|
||||
}
|
||||
|
||||
if (eventName === 'message_stop') {
|
||||
claudeContext.doneSent = true;
|
||||
}
|
||||
}
|
||||
|
||||
export async function chatProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/chat/completions', async (request: FastifyRequest, reply: FastifyReply) =>
|
||||
handleChatProxyRequest(request, reply, 'openai'));
|
||||
@@ -54,7 +115,7 @@ async function handleChatProxyRequest(
|
||||
return reply.code(parsedRequest.error.statusCode).send(parsedRequest.error.payload);
|
||||
}
|
||||
|
||||
const { requestedModel, isStream, upstreamBody } = parsedRequest.value!;
|
||||
const { requestedModel, isStream, upstreamBody, claudeOriginalBody } = parsedRequest.value!;
|
||||
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
@@ -105,6 +166,9 @@ async function handleChatProxyRequest(
|
||||
stream: isStream,
|
||||
tokenValue: selected.tokenValue,
|
||||
openaiBody: upstreamBody,
|
||||
downstreamFormat,
|
||||
claudeOriginalBody,
|
||||
downstreamHeaders: request.headers as Record<string, unknown>,
|
||||
});
|
||||
|
||||
const targetUrl = `${selected.site.url}${endpointRequest.path}`;
|
||||
@@ -312,6 +376,27 @@ async function handleChatProxyRequest(
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(parsedPayload));
|
||||
|
||||
if (downstreamFormat === 'claude') {
|
||||
const payloadType = (isRecord(parsedPayload) && typeof parsedPayload.type === 'string')
|
||||
? parsedPayload.type
|
||||
: '';
|
||||
const claudeEventName = isClaudeSseEventName(eventBlock.event)
|
||||
? eventBlock.event
|
||||
: (isClaudeSseEventName(payloadType) ? payloadType : '');
|
||||
|
||||
if (claudeEventName) {
|
||||
syncClaudeStreamStateFromRawEvent(
|
||||
claudeEventName,
|
||||
parsedPayload,
|
||||
streamContext,
|
||||
claudeContext,
|
||||
);
|
||||
reply.raw.write(serializeRawSseEvent(claudeEventName, eventBlock.data));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
const normalizedEvent = normalizeUpstreamStreamEvent(parsedPayload, streamContext, modelName);
|
||||
writeLines(serializeNormalizedStreamEvent(
|
||||
downstreamFormat,
|
||||
|
||||
@@ -11,18 +11,33 @@ export type StreamTransformContext = {
|
||||
created: number;
|
||||
roleSent: boolean;
|
||||
doneSent: boolean;
|
||||
toolCalls: Record<number, { id?: string; name?: string }>;
|
||||
};
|
||||
|
||||
export type ClaudeDownstreamContext = {
|
||||
messageStarted: boolean;
|
||||
contentBlockStarted: boolean;
|
||||
doneSent: boolean;
|
||||
textBlockIndex: number | null;
|
||||
nextContentBlockIndex: number;
|
||||
toolBlocks: Record<number, {
|
||||
contentIndex: number;
|
||||
id: string;
|
||||
name: string;
|
||||
open: boolean;
|
||||
}>;
|
||||
};
|
||||
|
||||
export type NormalizedStreamEvent = {
|
||||
role?: 'assistant';
|
||||
contentDelta?: string;
|
||||
reasoningDelta?: string;
|
||||
toolCallDeltas?: Array<{
|
||||
index: number;
|
||||
id?: string;
|
||||
name?: string;
|
||||
argumentsDelta?: string;
|
||||
}>;
|
||||
finishReason?: string | null;
|
||||
done?: boolean;
|
||||
};
|
||||
@@ -40,6 +55,7 @@ export type ParsedDownstreamChatRequest = {
|
||||
requestedModel: string;
|
||||
isStream: boolean;
|
||||
upstreamBody: Record<string, unknown>;
|
||||
claudeOriginalBody?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
@@ -182,6 +198,7 @@ export function createStreamTransformContext(modelName: string): StreamTransform
|
||||
created: Math.floor(Date.now() / 1000),
|
||||
roleSent: false,
|
||||
doneSent: false,
|
||||
toolCalls: {},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -190,6 +207,9 @@ export function createClaudeDownstreamContext(): ClaudeDownstreamContext {
|
||||
messageStarted: false,
|
||||
contentBlockStarted: false,
|
||||
doneSent: false,
|
||||
textBlockIndex: null,
|
||||
nextContentBlockIndex: 0,
|
||||
toolBlocks: {},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -355,6 +375,7 @@ export function parseDownstreamChatRequest(
|
||||
requestedModel: converted.model,
|
||||
isStream: converted.stream,
|
||||
upstreamBody: converted.payload,
|
||||
claudeOriginalBody: raw,
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -498,10 +519,43 @@ export function normalizeUpstreamStreamEvent(
|
||||
|| deltaParsed.reasoning
|
||||
|| '';
|
||||
|
||||
const rawToolCalls = Array.isArray((delta as any).tool_calls)
|
||||
? ((delta as any).tool_calls as unknown[])
|
||||
: [];
|
||||
const toolCallDeltas = rawToolCalls
|
||||
.map((item, itemIndex) => {
|
||||
if (!isRecord(item)) return null;
|
||||
const functionPart = isRecord(item.function) ? item.function : {};
|
||||
const index = (
|
||||
typeof item.index === 'number' && Number.isFinite(item.index)
|
||||
? Math.max(0, Math.trunc(item.index))
|
||||
: itemIndex
|
||||
);
|
||||
const id = typeof item.id === 'string' && item.id.trim().length > 0
|
||||
? item.id
|
||||
: undefined;
|
||||
const name = typeof functionPart.name === 'string' && functionPart.name.trim().length > 0
|
||||
? functionPart.name
|
||||
: undefined;
|
||||
const argumentsDelta = typeof functionPart.arguments === 'string'
|
||||
? functionPart.arguments
|
||||
: undefined;
|
||||
|
||||
if (!id && !name && argumentsDelta === undefined) return null;
|
||||
return {
|
||||
index,
|
||||
id,
|
||||
name,
|
||||
argumentsDelta,
|
||||
};
|
||||
})
|
||||
.filter((item): item is NonNullable<typeof item> => !!item);
|
||||
|
||||
return {
|
||||
role: (delta as any).role === 'assistant' ? 'assistant' : undefined,
|
||||
contentDelta: contentDelta || undefined,
|
||||
reasoningDelta: reasoningDelta || undefined,
|
||||
toolCallDeltas: toolCallDeltas.length > 0 ? toolCallDeltas : undefined,
|
||||
finishReason: normalizeStopReason(choice?.finish_reason),
|
||||
};
|
||||
}
|
||||
@@ -539,6 +593,42 @@ export function normalizeUpstreamStreamEvent(
|
||||
}
|
||||
|
||||
if (type === 'content_block_start') {
|
||||
const index = (
|
||||
typeof (payload as any).index === 'number' && Number.isFinite((payload as any).index)
|
||||
? Math.max(0, Math.trunc((payload as any).index))
|
||||
: 0
|
||||
);
|
||||
const contentBlock = isRecord(payload.content_block) ? payload.content_block : {};
|
||||
if (contentBlock.type === 'tool_use') {
|
||||
const id = typeof contentBlock.id === 'string' && contentBlock.id.trim().length > 0
|
||||
? contentBlock.id
|
||||
: undefined;
|
||||
const name = typeof contentBlock.name === 'string' && contentBlock.name.trim().length > 0
|
||||
? contentBlock.name
|
||||
: undefined;
|
||||
let argumentsDelta: string | undefined;
|
||||
const rawInput = contentBlock.input;
|
||||
if (typeof rawInput === 'string') {
|
||||
argumentsDelta = rawInput;
|
||||
} else if (Array.isArray(rawInput) || isRecord(rawInput)) {
|
||||
try {
|
||||
const serialized = JSON.stringify(rawInput);
|
||||
if (serialized && serialized !== '{}' && serialized !== '[]') {
|
||||
argumentsDelta = serialized;
|
||||
}
|
||||
} catch {}
|
||||
}
|
||||
|
||||
return {
|
||||
toolCallDeltas: [{
|
||||
index,
|
||||
id,
|
||||
name,
|
||||
argumentsDelta,
|
||||
}],
|
||||
};
|
||||
}
|
||||
|
||||
const parsed = extractTextAndReasoning(payload.content_block);
|
||||
return {
|
||||
contentDelta: parsed.content || undefined,
|
||||
@@ -551,6 +641,23 @@ export function normalizeUpstreamStreamEvent(
|
||||
const deltaType = typeof delta.type === 'string' ? delta.type : '';
|
||||
const parsed = extractTextAndReasoning(delta);
|
||||
|
||||
if (deltaType === 'input_json_delta') {
|
||||
const index = (
|
||||
typeof (payload as any).index === 'number' && Number.isFinite((payload as any).index)
|
||||
? Math.max(0, Math.trunc((payload as any).index))
|
||||
: 0
|
||||
);
|
||||
const partialJson = typeof (delta as any).partial_json === 'string'
|
||||
? (delta as any).partial_json
|
||||
: undefined;
|
||||
return {
|
||||
toolCallDeltas: [{
|
||||
index,
|
||||
argumentsDelta: partialJson,
|
||||
}],
|
||||
};
|
||||
}
|
||||
|
||||
if (deltaType === 'thinking_delta') {
|
||||
return {
|
||||
reasoningDelta: parsed.content || parsed.reasoning || undefined,
|
||||
@@ -626,6 +733,34 @@ function buildOpenAiStreamChunk(
|
||||
delta.reasoning_content = event.reasoningDelta;
|
||||
}
|
||||
|
||||
if (Array.isArray(event.toolCallDeltas) && event.toolCallDeltas.length > 0) {
|
||||
const toolCalls = event.toolCallDeltas.map((toolDelta) => {
|
||||
const index = Number.isFinite(toolDelta.index) ? Math.max(0, Math.trunc(toolDelta.index)) : 0;
|
||||
const existing = context.toolCalls[index] || {};
|
||||
const id = toolDelta.id || existing.id || `call_meta_${index}`;
|
||||
const name = toolDelta.name || existing.name || '';
|
||||
context.toolCalls[index] = {
|
||||
id,
|
||||
name: name || existing.name,
|
||||
};
|
||||
|
||||
const fn: Record<string, unknown> = {};
|
||||
if (name) fn.name = name;
|
||||
if (toolDelta.argumentsDelta !== undefined) fn.arguments = toolDelta.argumentsDelta;
|
||||
|
||||
return {
|
||||
index,
|
||||
id,
|
||||
type: 'function',
|
||||
function: fn,
|
||||
};
|
||||
});
|
||||
|
||||
if (toolCalls.length > 0) {
|
||||
delta.tool_calls = toolCalls;
|
||||
}
|
||||
}
|
||||
|
||||
// Some OpenAI-compatible clients (e.g. OpenWebUI) expect starter chunk to include empty content.
|
||||
if (isInitialAssistantRoleOnlyEvent) {
|
||||
delta.content = '';
|
||||
@@ -675,15 +810,18 @@ function ensureClaudeStartEvents(
|
||||
return [serializeSse('message_start', payload)];
|
||||
}
|
||||
|
||||
function ensureClaudeContentBlockStart(
|
||||
function ensureClaudeTextBlockStart(
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
if (claudeContext.contentBlockStarted) return [];
|
||||
if (claudeContext.contentBlockStarted && claudeContext.textBlockIndex !== null) return [];
|
||||
const contentIndex = claudeContext.nextContentBlockIndex;
|
||||
claudeContext.nextContentBlockIndex += 1;
|
||||
claudeContext.contentBlockStarted = true;
|
||||
claudeContext.textBlockIndex = contentIndex;
|
||||
|
||||
return [serializeSse('content_block_start', {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
index: contentIndex,
|
||||
content_block: {
|
||||
type: 'text',
|
||||
text: '',
|
||||
@@ -691,6 +829,93 @@ function ensureClaudeContentBlockStart(
|
||||
})];
|
||||
}
|
||||
|
||||
function closeClaudeTextBlock(
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
if (!claudeContext.contentBlockStarted || claudeContext.textBlockIndex === null) return [];
|
||||
|
||||
const contentIndex = claudeContext.textBlockIndex;
|
||||
claudeContext.contentBlockStarted = false;
|
||||
claudeContext.textBlockIndex = null;
|
||||
return [serializeSse('content_block_stop', {
|
||||
type: 'content_block_stop',
|
||||
index: contentIndex,
|
||||
})];
|
||||
}
|
||||
|
||||
function normalizeToolContentIndex(raw: unknown): number {
|
||||
if (typeof raw === 'number' && Number.isFinite(raw)) {
|
||||
return Math.max(0, Math.trunc(raw));
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
function ensureClaudeToolBlockStart(
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
toolDelta: NonNullable<NormalizedStreamEvent['toolCallDeltas']>[number],
|
||||
): { events: string[]; contentIndex: number } {
|
||||
const toolSlot = normalizeToolContentIndex(toolDelta.index);
|
||||
let state = claudeContext.toolBlocks[toolSlot];
|
||||
if (!state) {
|
||||
const fallbackId = `toolu_meta_${toolSlot}`;
|
||||
const fallbackName = `tool_${toolSlot}`;
|
||||
state = {
|
||||
contentIndex: claudeContext.nextContentBlockIndex,
|
||||
id: toolDelta.id || fallbackId,
|
||||
name: toolDelta.name || fallbackName,
|
||||
open: false,
|
||||
};
|
||||
claudeContext.nextContentBlockIndex += 1;
|
||||
claudeContext.toolBlocks[toolSlot] = state;
|
||||
} else {
|
||||
if (toolDelta.id && state.id !== toolDelta.id) {
|
||||
state.id = toolDelta.id;
|
||||
}
|
||||
if (toolDelta.name && state.name !== toolDelta.name) {
|
||||
state.name = toolDelta.name;
|
||||
}
|
||||
}
|
||||
|
||||
const events: string[] = [];
|
||||
if (!state.open) {
|
||||
state.open = true;
|
||||
events.push(serializeSse('content_block_start', {
|
||||
type: 'content_block_start',
|
||||
index: state.contentIndex,
|
||||
content_block: {
|
||||
type: 'tool_use',
|
||||
id: state.id,
|
||||
name: state.name,
|
||||
input: {},
|
||||
},
|
||||
}));
|
||||
}
|
||||
|
||||
return {
|
||||
events,
|
||||
contentIndex: state.contentIndex,
|
||||
};
|
||||
}
|
||||
|
||||
function closeClaudeToolBlocks(
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
const openBlocks = Object.values(claudeContext.toolBlocks)
|
||||
.filter((item) => item.open)
|
||||
.sort((a, b) => a.contentIndex - b.contentIndex);
|
||||
if (openBlocks.length <= 0) return [];
|
||||
|
||||
const events: string[] = [];
|
||||
for (const block of openBlocks) {
|
||||
block.open = false;
|
||||
events.push(serializeSse('content_block_stop', {
|
||||
type: 'content_block_stop',
|
||||
index: block.contentIndex,
|
||||
}));
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
function buildClaudeDoneEvents(
|
||||
context: StreamTransformContext,
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
@@ -701,13 +926,8 @@ function buildClaudeDoneEvents(
|
||||
const events: string[] = [];
|
||||
events.push(...ensureClaudeStartEvents(context, claudeContext));
|
||||
|
||||
if (claudeContext.contentBlockStarted) {
|
||||
events.push(serializeSse('content_block_stop', {
|
||||
type: 'content_block_stop',
|
||||
index: 0,
|
||||
}));
|
||||
claudeContext.contentBlockStarted = false;
|
||||
}
|
||||
events.push(...closeClaudeTextBlock(claudeContext));
|
||||
events.push(...closeClaudeToolBlocks(claudeContext));
|
||||
|
||||
events.push(serializeSse('message_delta', {
|
||||
type: 'message_delta',
|
||||
@@ -736,10 +956,6 @@ export function serializeNormalizedStreamEvent(
|
||||
return chunk ? [serializeSse('', chunk)] : [];
|
||||
}
|
||||
|
||||
if (event.done || event.finishReason) {
|
||||
return buildClaudeDoneEvents(context, claudeContext, event.finishReason);
|
||||
}
|
||||
|
||||
const events: string[] = [];
|
||||
if (event.role === 'assistant' || event.contentDelta || event.reasoningDelta) {
|
||||
events.push(...ensureClaudeStartEvents(context, claudeContext));
|
||||
@@ -750,11 +966,31 @@ export function serializeNormalizedStreamEvent(
|
||||
event.contentDelta || '',
|
||||
]);
|
||||
|
||||
if (Array.isArray(event.toolCallDeltas) && event.toolCallDeltas.length > 0) {
|
||||
events.push(...closeClaudeTextBlock(claudeContext));
|
||||
for (const toolDelta of event.toolCallDeltas) {
|
||||
const toolBlock = ensureClaudeToolBlockStart(claudeContext, toolDelta);
|
||||
events.push(...toolBlock.events);
|
||||
|
||||
if (toolDelta.argumentsDelta !== undefined && toolDelta.argumentsDelta.length > 0) {
|
||||
events.push(serializeSse('content_block_delta', {
|
||||
type: 'content_block_delta',
|
||||
index: toolBlock.contentIndex,
|
||||
delta: {
|
||||
type: 'input_json_delta',
|
||||
partial_json: toolDelta.argumentsDelta,
|
||||
},
|
||||
}));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (mergedText) {
|
||||
events.push(...ensureClaudeContentBlockStart(claudeContext));
|
||||
events.push(...closeClaudeToolBlocks(claudeContext));
|
||||
events.push(...ensureClaudeTextBlockStart(claudeContext));
|
||||
events.push(serializeSse('content_block_delta', {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
index: claudeContext.textBlockIndex ?? 0,
|
||||
delta: {
|
||||
type: 'text_delta',
|
||||
text: mergedText,
|
||||
@@ -762,6 +998,10 @@ export function serializeNormalizedStreamEvent(
|
||||
}));
|
||||
}
|
||||
|
||||
if (event.done || event.finishReason) {
|
||||
events.push(...buildClaudeDoneEvents(context, claudeContext, event.finishReason));
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
|
||||
@@ -8,8 +8,14 @@ import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { estimateProxyCost } from '../../services/modelPricingService.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
|
||||
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
|
||||
import { normalizeUpstreamFinalResponse } from './chatFormats.js';
|
||||
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
|
||||
import {
|
||||
createStreamTransformContext,
|
||||
normalizeUpstreamFinalResponse,
|
||||
normalizeUpstreamStreamEvent,
|
||||
pullSseEventsWithDone,
|
||||
type NormalizedStreamEvent,
|
||||
} from './chatFormats.js';
|
||||
import {
|
||||
buildUpstreamEndpointRequest,
|
||||
isEndpointDowngradeError,
|
||||
@@ -95,23 +101,137 @@ function convertResponsesBodyToOpenAiBody(
|
||||
return payload;
|
||||
}
|
||||
|
||||
type UsageSummary = {
|
||||
promptTokens: number;
|
||||
completionTokens: number;
|
||||
totalTokens: number;
|
||||
};
|
||||
|
||||
type ResponsesToolCall = {
|
||||
id: string;
|
||||
name: string;
|
||||
arguments: string;
|
||||
};
|
||||
|
||||
type ResponsesMessageItemState = {
|
||||
itemId: string;
|
||||
outputIndex: number;
|
||||
contentIndex: number;
|
||||
text: string;
|
||||
};
|
||||
|
||||
type ResponsesToolItemState = {
|
||||
toolIndex: number;
|
||||
itemId: string;
|
||||
callId: string;
|
||||
outputIndex: number;
|
||||
name: string;
|
||||
arguments: string;
|
||||
};
|
||||
|
||||
type ResponsesStreamState = {
|
||||
started: boolean;
|
||||
completed: boolean;
|
||||
responseId: string;
|
||||
model: string;
|
||||
createdAt: number;
|
||||
sequenceNumber: number;
|
||||
outputCursor: number;
|
||||
messageItem: ResponsesMessageItemState | null;
|
||||
toolItems: Map<number, ResponsesToolItemState>;
|
||||
};
|
||||
|
||||
function ensureResponseId(rawId: string): string {
|
||||
const trimmed = rawId.trim() || `resp_${Date.now()}`;
|
||||
return trimmed.startsWith('resp_') ? trimmed : `resp_${trimmed}`;
|
||||
}
|
||||
|
||||
function ensureMessageId(rawId: string): string {
|
||||
const trimmed = rawId.trim() || `msg_${Date.now()}`;
|
||||
return trimmed.startsWith('msg_') ? trimmed : `msg_${trimmed}`;
|
||||
}
|
||||
|
||||
function ensureFunctionCallId(rawId: string): string {
|
||||
const trimmed = rawId.trim();
|
||||
if (!trimmed) return `call_${Date.now()}`;
|
||||
return trimmed.startsWith('call_') ? trimmed : `call_${trimmed}`;
|
||||
}
|
||||
|
||||
function stringifyToolInput(input: unknown): string {
|
||||
if (typeof input === 'string') return input;
|
||||
if (!input || typeof input !== 'object') return '';
|
||||
try {
|
||||
return JSON.stringify(input);
|
||||
} catch {
|
||||
return '';
|
||||
}
|
||||
}
|
||||
|
||||
function extractToolCallsFromUpstream(payload: unknown): ResponsesToolCall[] {
|
||||
if (!isRecord(payload)) return [];
|
||||
|
||||
if (Array.isArray(payload.choices)) {
|
||||
const choice = payload.choices[0];
|
||||
const message = isRecord((choice as any)?.message) ? (choice as any).message : {};
|
||||
const toolCalls = Array.isArray((message as any).tool_calls) ? (message as any).tool_calls : [];
|
||||
return toolCalls
|
||||
.map((item: unknown, index: number) => {
|
||||
if (!isRecord(item)) return null;
|
||||
const fn = isRecord(item.function) ? item.function : {};
|
||||
const id = typeof item.id === 'string' && item.id.trim().length > 0
|
||||
? item.id
|
||||
: `call_${index}`;
|
||||
const name = typeof fn.name === 'string' ? fn.name : '';
|
||||
const args = typeof fn.arguments === 'string' ? fn.arguments : '';
|
||||
return {
|
||||
id: ensureFunctionCallId(id),
|
||||
name,
|
||||
arguments: args,
|
||||
};
|
||||
})
|
||||
.filter((item: ResponsesToolCall | null): item is ResponsesToolCall => !!item);
|
||||
}
|
||||
|
||||
if (payload.type === 'message' && Array.isArray(payload.content)) {
|
||||
return payload.content
|
||||
.map((item: unknown, index: number) => {
|
||||
if (!isRecord(item) || item.type !== 'tool_use') return null;
|
||||
const id = typeof item.id === 'string' && item.id.trim().length > 0
|
||||
? item.id
|
||||
: `call_${index}`;
|
||||
const name = typeof item.name === 'string' ? item.name : '';
|
||||
const args = stringifyToolInput(item.input);
|
||||
return {
|
||||
id: ensureFunctionCallId(id),
|
||||
name,
|
||||
arguments: args,
|
||||
};
|
||||
})
|
||||
.filter((item: ResponsesToolCall | null): item is ResponsesToolCall => !!item);
|
||||
}
|
||||
|
||||
return [];
|
||||
}
|
||||
|
||||
function toResponsesPayload(
|
||||
upstreamPayload: unknown,
|
||||
normalized: ReturnType<typeof normalizeUpstreamFinalResponse>,
|
||||
usage: { promptTokens: number; completionTokens: number; totalTokens: number },
|
||||
usage: UsageSummary,
|
||||
): Record<string, unknown> {
|
||||
if (isRecord(upstreamPayload) && upstreamPayload.object === 'response') {
|
||||
return upstreamPayload;
|
||||
}
|
||||
|
||||
const normalizedId = typeof normalized.id === 'string' && normalized.id.trim()
|
||||
? normalized.id.trim()
|
||||
: `resp_${Date.now()}`;
|
||||
const responseId = normalizedId.startsWith('resp_') ? normalizedId : `resp_${normalizedId}`;
|
||||
const messageId = normalizedId.startsWith('msg_') ? normalizedId : `msg_${normalizedId}`;
|
||||
const responseId = ensureResponseId(normalizedId);
|
||||
const messageId = ensureMessageId(normalizedId);
|
||||
const toolCalls = extractToolCallsFromUpstream(upstreamPayload);
|
||||
|
||||
return {
|
||||
id: responseId,
|
||||
object: 'response',
|
||||
created: normalized.created,
|
||||
status: 'completed',
|
||||
model: normalized.model,
|
||||
output: [{
|
||||
const output: Array<Record<string, unknown>> = [];
|
||||
if (normalized.content || toolCalls.length === 0) {
|
||||
output.push({
|
||||
id: messageId,
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
@@ -120,7 +240,27 @@ function toResponsesPayload(
|
||||
type: 'output_text',
|
||||
text: normalized.content || '',
|
||||
}],
|
||||
}],
|
||||
});
|
||||
}
|
||||
|
||||
for (const toolCall of toolCalls) {
|
||||
output.push({
|
||||
id: toolCall.id,
|
||||
type: 'function_call',
|
||||
status: 'completed',
|
||||
call_id: toolCall.id,
|
||||
name: toolCall.name,
|
||||
arguments: toolCall.arguments,
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
id: responseId,
|
||||
object: 'response',
|
||||
created: normalized.created,
|
||||
status: 'completed',
|
||||
model: normalized.model,
|
||||
output,
|
||||
output_text: normalized.content || '',
|
||||
usage: {
|
||||
input_tokens: usage.promptTokens,
|
||||
@@ -130,6 +270,342 @@ function toResponsesPayload(
|
||||
};
|
||||
}
|
||||
|
||||
function createResponsesStreamState(modelName: string): ResponsesStreamState {
|
||||
return {
|
||||
started: false,
|
||||
completed: false,
|
||||
responseId: `resp_meta_${Date.now()}`,
|
||||
model: modelName,
|
||||
createdAt: Math.floor(Date.now() / 1000),
|
||||
sequenceNumber: 0,
|
||||
outputCursor: 0,
|
||||
messageItem: null,
|
||||
toolItems: new Map(),
|
||||
};
|
||||
}
|
||||
|
||||
function toResponsesUsage(usage: UsageSummary): Record<string, number> {
|
||||
return {
|
||||
input_tokens: usage.promptTokens,
|
||||
output_tokens: usage.completionTokens,
|
||||
total_tokens: usage.totalTokens,
|
||||
};
|
||||
}
|
||||
|
||||
function nextSequence(state: ResponsesStreamState): number {
|
||||
const value = state.sequenceNumber;
|
||||
state.sequenceNumber += 1;
|
||||
return value;
|
||||
}
|
||||
|
||||
function serializeResponsesSse(eventType: string, payload: Record<string, unknown>): string {
|
||||
return `event: ${eventType}\ndata: ${JSON.stringify(payload)}\n\n`;
|
||||
}
|
||||
|
||||
function emitResponsesEvent(
|
||||
state: ResponsesStreamState,
|
||||
eventType: string,
|
||||
payload: Record<string, unknown>,
|
||||
): string {
|
||||
return serializeResponsesSse(eventType, {
|
||||
type: eventType,
|
||||
sequence_number: nextSequence(state),
|
||||
...payload,
|
||||
});
|
||||
}
|
||||
|
||||
function buildResponseObject(
|
||||
state: ResponsesStreamState,
|
||||
status: 'in_progress' | 'completed',
|
||||
usage?: UsageSummary,
|
||||
): Record<string, unknown> {
|
||||
const response: Record<string, unknown> = {
|
||||
id: state.responseId,
|
||||
object: 'response',
|
||||
created_at: state.createdAt,
|
||||
status,
|
||||
model: state.model,
|
||||
output: [],
|
||||
};
|
||||
if (usage) {
|
||||
response.usage = toResponsesUsage(usage);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
function ensureResponsesStarted(
|
||||
state: ResponsesStreamState,
|
||||
streamContext: { id: string; model: string; created: number },
|
||||
): string[] {
|
||||
if (state.started) return [];
|
||||
|
||||
state.started = true;
|
||||
state.responseId = ensureResponseId(streamContext.id || state.responseId);
|
||||
state.model = streamContext.model || state.model;
|
||||
state.createdAt = streamContext.created || state.createdAt;
|
||||
|
||||
const inProgress = buildResponseObject(state, 'in_progress');
|
||||
return [
|
||||
emitResponsesEvent(state, 'response.created', { response: inProgress }),
|
||||
emitResponsesEvent(state, 'response.in_progress', { response: inProgress }),
|
||||
];
|
||||
}
|
||||
|
||||
function ensureMessageItem(state: ResponsesStreamState): string[] {
|
||||
if (state.messageItem) return [];
|
||||
|
||||
const outputIndex = state.outputCursor;
|
||||
state.outputCursor += 1;
|
||||
const itemId = ensureMessageId(`${state.responseId}_${outputIndex}`);
|
||||
state.messageItem = {
|
||||
itemId,
|
||||
outputIndex,
|
||||
contentIndex: 0,
|
||||
text: '',
|
||||
};
|
||||
|
||||
return [
|
||||
emitResponsesEvent(state, 'response.output_item.added', {
|
||||
output_index: outputIndex,
|
||||
item: {
|
||||
id: itemId,
|
||||
type: 'message',
|
||||
status: 'in_progress',
|
||||
role: 'assistant',
|
||||
content: [],
|
||||
},
|
||||
}),
|
||||
emitResponsesEvent(state, 'response.content_part.added', {
|
||||
item_id: itemId,
|
||||
output_index: outputIndex,
|
||||
content_index: 0,
|
||||
part: {
|
||||
type: 'output_text',
|
||||
text: '',
|
||||
},
|
||||
}),
|
||||
];
|
||||
}
|
||||
|
||||
function appendMessageDelta(state: ResponsesStreamState, delta: string): string[] {
|
||||
if (!delta) return [];
|
||||
|
||||
const events: string[] = [];
|
||||
events.push(...ensureMessageItem(state));
|
||||
if (!state.messageItem) return events;
|
||||
|
||||
state.messageItem.text += delta;
|
||||
events.push(emitResponsesEvent(state, 'response.output_text.delta', {
|
||||
item_id: state.messageItem.itemId,
|
||||
output_index: state.messageItem.outputIndex,
|
||||
content_index: state.messageItem.contentIndex,
|
||||
delta,
|
||||
}));
|
||||
return events;
|
||||
}
|
||||
|
||||
function closeMessageItem(state: ResponsesStreamState): string[] {
|
||||
if (!state.messageItem) return [];
|
||||
|
||||
const item = state.messageItem;
|
||||
const events = [
|
||||
emitResponsesEvent(state, 'response.output_text.done', {
|
||||
item_id: item.itemId,
|
||||
output_index: item.outputIndex,
|
||||
content_index: item.contentIndex,
|
||||
text: item.text,
|
||||
}),
|
||||
emitResponsesEvent(state, 'response.content_part.done', {
|
||||
item_id: item.itemId,
|
||||
output_index: item.outputIndex,
|
||||
content_index: item.contentIndex,
|
||||
part: {
|
||||
type: 'output_text',
|
||||
text: item.text,
|
||||
},
|
||||
}),
|
||||
emitResponsesEvent(state, 'response.output_item.done', {
|
||||
output_index: item.outputIndex,
|
||||
item: {
|
||||
id: item.itemId,
|
||||
type: 'message',
|
||||
status: 'completed',
|
||||
role: 'assistant',
|
||||
content: [{
|
||||
type: 'output_text',
|
||||
text: item.text,
|
||||
}],
|
||||
},
|
||||
}),
|
||||
];
|
||||
|
||||
state.messageItem = null;
|
||||
return events;
|
||||
}
|
||||
|
||||
function ensureToolItem(
|
||||
state: ResponsesStreamState,
|
||||
toolIndex: number,
|
||||
id?: string,
|
||||
name?: string,
|
||||
): string[] {
|
||||
const existing = state.toolItems.get(toolIndex);
|
||||
if (existing) {
|
||||
if (id && !existing.callId) existing.callId = ensureFunctionCallId(id);
|
||||
if (id && !existing.itemId) existing.itemId = ensureFunctionCallId(id);
|
||||
if (name && !existing.name) existing.name = name;
|
||||
return [];
|
||||
}
|
||||
|
||||
const outputIndex = state.outputCursor;
|
||||
state.outputCursor += 1;
|
||||
const callId = ensureFunctionCallId(id || `${state.responseId}_${toolIndex}`);
|
||||
const itemId = callId;
|
||||
const toolState: ResponsesToolItemState = {
|
||||
toolIndex,
|
||||
itemId,
|
||||
callId,
|
||||
outputIndex,
|
||||
name: name || '',
|
||||
arguments: '',
|
||||
};
|
||||
state.toolItems.set(toolIndex, toolState);
|
||||
|
||||
return [
|
||||
emitResponsesEvent(state, 'response.output_item.added', {
|
||||
output_index: outputIndex,
|
||||
item: {
|
||||
id: itemId,
|
||||
type: 'function_call',
|
||||
status: 'in_progress',
|
||||
call_id: callId,
|
||||
name: toolState.name,
|
||||
arguments: '',
|
||||
},
|
||||
}),
|
||||
];
|
||||
}
|
||||
|
||||
function appendToolCallDelta(
|
||||
state: ResponsesStreamState,
|
||||
toolDelta: NonNullable<NormalizedStreamEvent['toolCallDeltas']>[number],
|
||||
): string[] {
|
||||
const toolIndex = Number.isFinite(toolDelta.index) ? Math.max(0, Math.trunc(toolDelta.index)) : 0;
|
||||
const events: string[] = [];
|
||||
events.push(...ensureToolItem(state, toolIndex, toolDelta.id, toolDelta.name));
|
||||
|
||||
const toolState = state.toolItems.get(toolIndex);
|
||||
if (!toolState) return events;
|
||||
if (toolDelta.name && !toolState.name) toolState.name = toolDelta.name;
|
||||
|
||||
if (toolDelta.argumentsDelta !== undefined) {
|
||||
toolState.arguments += toolDelta.argumentsDelta;
|
||||
events.push(emitResponsesEvent(state, 'response.function_call_arguments.delta', {
|
||||
item_id: toolState.itemId,
|
||||
output_index: toolState.outputIndex,
|
||||
delta: toolDelta.argumentsDelta,
|
||||
}));
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
function closeToolItems(state: ResponsesStreamState): string[] {
|
||||
if (state.toolItems.size <= 0) return [];
|
||||
|
||||
const ordered = Array.from(state.toolItems.values())
|
||||
.sort((a, b) => a.outputIndex - b.outputIndex);
|
||||
const events: string[] = [];
|
||||
for (const toolItem of ordered) {
|
||||
events.push(emitResponsesEvent(state, 'response.function_call_arguments.done', {
|
||||
item_id: toolItem.itemId,
|
||||
output_index: toolItem.outputIndex,
|
||||
arguments: toolItem.arguments,
|
||||
}));
|
||||
events.push(emitResponsesEvent(state, 'response.output_item.done', {
|
||||
output_index: toolItem.outputIndex,
|
||||
item: {
|
||||
id: toolItem.itemId,
|
||||
type: 'function_call',
|
||||
status: 'completed',
|
||||
call_id: toolItem.callId,
|
||||
name: toolItem.name,
|
||||
arguments: toolItem.arguments,
|
||||
},
|
||||
}));
|
||||
}
|
||||
|
||||
state.toolItems.clear();
|
||||
return events;
|
||||
}
|
||||
|
||||
function completeResponsesStream(
|
||||
state: ResponsesStreamState,
|
||||
streamContext: { id: string; model: string; created: number },
|
||||
usage: UsageSummary,
|
||||
): string[] {
|
||||
if (state.completed) return [];
|
||||
|
||||
const events: string[] = [];
|
||||
events.push(...ensureResponsesStarted(state, streamContext));
|
||||
events.push(...closeMessageItem(state));
|
||||
events.push(...closeToolItems(state));
|
||||
events.push(emitResponsesEvent(state, 'response.completed', {
|
||||
response: buildResponseObject(state, 'completed', usage),
|
||||
}));
|
||||
events.push('data: [DONE]\n\n');
|
||||
state.completed = true;
|
||||
return events;
|
||||
}
|
||||
|
||||
function serializeConvertedResponsesEvents(input: {
|
||||
state: ResponsesStreamState;
|
||||
streamContext: { id: string; model: string; created: number };
|
||||
event: NormalizedStreamEvent;
|
||||
usage: UsageSummary;
|
||||
}): string[] {
|
||||
const { state, streamContext, event, usage } = input;
|
||||
if (state.completed) return [];
|
||||
|
||||
const shouldStart = (
|
||||
event.role === 'assistant'
|
||||
|| !!event.contentDelta
|
||||
|| !!event.reasoningDelta
|
||||
|| (Array.isArray(event.toolCallDeltas) && event.toolCallDeltas.length > 0)
|
||||
|| !!event.done
|
||||
|| !!event.finishReason
|
||||
);
|
||||
|
||||
const events: string[] = [];
|
||||
if (shouldStart) {
|
||||
events.push(...ensureResponsesStarted(state, streamContext));
|
||||
}
|
||||
|
||||
if (event.contentDelta) {
|
||||
events.push(...appendMessageDelta(state, event.contentDelta));
|
||||
}
|
||||
if (event.reasoningDelta) {
|
||||
events.push(...appendMessageDelta(state, event.reasoningDelta));
|
||||
}
|
||||
|
||||
if (Array.isArray(event.toolCallDeltas) && event.toolCallDeltas.length > 0) {
|
||||
events.push(...closeMessageItem(state));
|
||||
for (const toolDelta of event.toolCallDeltas) {
|
||||
events.push(...appendToolCallDelta(state, toolDelta));
|
||||
}
|
||||
}
|
||||
|
||||
if (event.finishReason) {
|
||||
// Keep compatibility with clients that expect finish before completed.
|
||||
}
|
||||
|
||||
if (event.done) {
|
||||
events.push(...completeResponsesStream(state, streamContext, usage));
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/responses', async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const body = request.body as any;
|
||||
@@ -166,18 +642,16 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
|
||||
const modelName = selected.actualModel || requestedModel;
|
||||
const openAiBody = convertResponsesBodyToOpenAiBody(body, modelName, isStream);
|
||||
const resolvedCandidates = await resolveUpstreamEndpointCandidates(
|
||||
const endpointCandidates = await resolveUpstreamEndpointCandidates(
|
||||
{
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
},
|
||||
modelName,
|
||||
'openai',
|
||||
'responses',
|
||||
);
|
||||
const endpointCandidates = resolvedCandidates
|
||||
.filter((endpoint) => endpoint !== 'responses');
|
||||
if (endpointCandidates.length === 0) {
|
||||
endpointCandidates.push('chat', 'messages');
|
||||
endpointCandidates.push('responses', 'chat', 'messages');
|
||||
}
|
||||
|
||||
const startTime = Date.now();
|
||||
@@ -196,6 +670,9 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
stream: isStream,
|
||||
tokenValue: selected.tokenValue,
|
||||
openaiBody: openAiBody,
|
||||
downstreamFormat: 'responses',
|
||||
responsesOriginalBody: body,
|
||||
downstreamHeaders: request.headers as Record<string, unknown>,
|
||||
});
|
||||
const targetUrl = `${selected.site.url}${endpointRequest.path}`;
|
||||
|
||||
@@ -256,11 +733,11 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
}
|
||||
|
||||
if (isStream) {
|
||||
reply.raw.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
});
|
||||
reply.raw.statusCode = 200;
|
||||
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
reply.raw.setHeader('Connection', 'keep-alive');
|
||||
reply.raw.setHeader('X-Accel-Buffering', 'no');
|
||||
|
||||
const reader = upstream.body?.getReader();
|
||||
if (!reader) {
|
||||
@@ -269,34 +746,87 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
}
|
||||
|
||||
const decoder = new TextDecoder();
|
||||
let parsedUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };
|
||||
let parsedUsage: UsageSummary = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };
|
||||
let sseBuffer = '';
|
||||
|
||||
const passthroughResponsesStream = successfulUpstreamPath === '/v1/responses';
|
||||
const streamContext = createStreamTransformContext(modelName);
|
||||
const responsesState = createResponsesStreamState(modelName);
|
||||
|
||||
const writeLines = (lines: string[]) => {
|
||||
for (const line of lines) reply.raw.write(line);
|
||||
};
|
||||
|
||||
const consumeSseBuffer = (incoming: string): string => {
|
||||
const pulled = pullSseEventsWithDone(incoming);
|
||||
for (const eventBlock of pulled.events) {
|
||||
if (eventBlock.data === '[DONE]') {
|
||||
if (passthroughResponsesStream) {
|
||||
reply.raw.write('data: [DONE]\n\n');
|
||||
} else {
|
||||
writeLines(completeResponsesStream(responsesState, streamContext, parsedUsage));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
let parsedPayload: unknown = null;
|
||||
try {
|
||||
parsedPayload = JSON.parse(eventBlock.data);
|
||||
} catch {
|
||||
parsedPayload = null;
|
||||
}
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(parsedPayload));
|
||||
}
|
||||
|
||||
if (passthroughResponsesStream) {
|
||||
const eventName = eventBlock.event ? `event: ${eventBlock.event}\n` : '';
|
||||
reply.raw.write(`${eventName}data: ${eventBlock.data}\n\n`);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
const normalizedEvent = normalizeUpstreamStreamEvent(parsedPayload, streamContext, modelName);
|
||||
writeLines(serializeConvertedResponsesEvents({
|
||||
state: responsesState,
|
||||
streamContext,
|
||||
event: normalizedEvent,
|
||||
usage: parsedUsage,
|
||||
}));
|
||||
continue;
|
||||
}
|
||||
|
||||
writeLines(serializeConvertedResponsesEvents({
|
||||
state: responsesState,
|
||||
streamContext,
|
||||
event: { contentDelta: eventBlock.data },
|
||||
usage: parsedUsage,
|
||||
}));
|
||||
}
|
||||
|
||||
return pulled.rest;
|
||||
};
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
reply.raw.write(chunk);
|
||||
if (!value) continue;
|
||||
|
||||
sseBuffer += chunk;
|
||||
const pulled = pullSseDataEvents(sseBuffer);
|
||||
sseBuffer = pulled.rest;
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
sseBuffer += decoder.decode(value, { stream: true });
|
||||
sseBuffer = consumeSseBuffer(sseBuffer);
|
||||
}
|
||||
|
||||
sseBuffer += decoder.decode();
|
||||
if (sseBuffer.trim().length > 0) {
|
||||
const pulled = pullSseDataEvents(`${sseBuffer}\n\n`);
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
sseBuffer = consumeSseBuffer(`${sseBuffer}\n\n`);
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
if (!passthroughResponsesStream) {
|
||||
writeLines(completeResponsesStream(responsesState, streamContext, parsedUsage));
|
||||
}
|
||||
reply.raw.end();
|
||||
}
|
||||
|
||||
@@ -350,7 +880,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
modelName,
|
||||
rawText,
|
||||
);
|
||||
const downstreamData = toResponsesPayload(normalized, parsedUsage);
|
||||
const downstreamData = toResponsesPayload(upstreamData, normalized, parsedUsage);
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
|
||||
@@ -2,6 +2,7 @@ import { fetchModelPricingCatalog } from '../../services/modelPricingService.js'
|
||||
import type { DownstreamFormat } from './chatFormats.js';
|
||||
|
||||
export type UpstreamEndpoint = 'chat' | 'messages' | 'responses';
|
||||
export type EndpointPreference = DownstreamFormat | 'responses';
|
||||
|
||||
type ChannelContext = {
|
||||
site: {
|
||||
@@ -25,10 +26,150 @@ function asTrimmedString(value: unknown): string {
|
||||
return typeof value === 'string' ? value.trim() : '';
|
||||
}
|
||||
|
||||
function headerValueToString(value: unknown): string | null {
|
||||
if (typeof value === 'string') {
|
||||
const trimmed = value.trim();
|
||||
return trimmed || null;
|
||||
}
|
||||
|
||||
if (Array.isArray(value)) {
|
||||
for (const item of value) {
|
||||
if (typeof item !== 'string') continue;
|
||||
const trimmed = item.trim();
|
||||
if (trimmed) return trimmed;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
const HOP_BY_HOP_HEADERS = new Set([
|
||||
'connection',
|
||||
'keep-alive',
|
||||
'proxy-authenticate',
|
||||
'proxy-authorization',
|
||||
'te',
|
||||
'trailer',
|
||||
'transfer-encoding',
|
||||
'upgrade',
|
||||
]);
|
||||
|
||||
const BLOCKED_PASSTHROUGH_HEADERS = new Set([
|
||||
'host',
|
||||
'content-length',
|
||||
'accept-encoding',
|
||||
'cookie',
|
||||
'authorization',
|
||||
'x-api-key',
|
||||
'x-goog-api-key',
|
||||
'sec-websocket-key',
|
||||
'sec-websocket-version',
|
||||
'sec-websocket-extensions',
|
||||
]);
|
||||
|
||||
function shouldSkipPassthroughHeader(key: string): boolean {
|
||||
return HOP_BY_HOP_HEADERS.has(key) || BLOCKED_PASSTHROUGH_HEADERS.has(key);
|
||||
}
|
||||
|
||||
function extractSafePassthroughHeaders(
|
||||
headers?: Record<string, unknown>,
|
||||
): Record<string, string> {
|
||||
if (!headers) return {};
|
||||
|
||||
const forwarded: Record<string, string> = {};
|
||||
for (const [rawKey, rawValue] of Object.entries(headers)) {
|
||||
const key = rawKey.toLowerCase();
|
||||
if (!key || shouldSkipPassthroughHeader(key)) continue;
|
||||
|
||||
const value = headerValueToString(rawValue);
|
||||
if (!value) continue;
|
||||
forwarded[key] = value;
|
||||
}
|
||||
|
||||
return forwarded;
|
||||
}
|
||||
|
||||
function extractClaudePassthroughHeaders(
|
||||
headers?: Record<string, unknown>,
|
||||
): Record<string, string> {
|
||||
if (!headers) return {};
|
||||
|
||||
const forwarded: Record<string, string> = {};
|
||||
for (const [rawKey, rawValue] of Object.entries(headers)) {
|
||||
const key = rawKey.toLowerCase();
|
||||
const shouldForward = (
|
||||
key.startsWith('anthropic-')
|
||||
|| key.startsWith('x-claude-')
|
||||
|| key.startsWith('x-stainless-')
|
||||
);
|
||||
if (!shouldForward) continue;
|
||||
|
||||
const value = headerValueToString(rawValue);
|
||||
if (!value) continue;
|
||||
forwarded[key] = value;
|
||||
}
|
||||
|
||||
return forwarded;
|
||||
}
|
||||
|
||||
function extractResponsesPassthroughHeaders(
|
||||
headers?: Record<string, unknown>,
|
||||
): Record<string, string> {
|
||||
if (!headers) return {};
|
||||
|
||||
const forwarded: Record<string, string> = {};
|
||||
for (const [rawKey, rawValue] of Object.entries(headers)) {
|
||||
const key = rawKey.toLowerCase();
|
||||
const shouldForward = (
|
||||
key.startsWith('openai-')
|
||||
|| key.startsWith('x-openai-')
|
||||
|| key.startsWith('x-stainless-')
|
||||
|| key.startsWith('chatgpt-')
|
||||
|| key === 'originator'
|
||||
);
|
||||
if (!shouldForward) continue;
|
||||
|
||||
const value = headerValueToString(rawValue);
|
||||
if (!value) continue;
|
||||
forwarded[key] = value;
|
||||
}
|
||||
|
||||
return forwarded;
|
||||
}
|
||||
|
||||
function ensureStreamAcceptHeader(
|
||||
headers: Record<string, string>,
|
||||
stream: boolean,
|
||||
): Record<string, string> {
|
||||
if (!stream) return headers;
|
||||
|
||||
const existingAccept = (
|
||||
headerValueToString(headers.accept)
|
||||
|| headerValueToString((headers as Record<string, unknown>).Accept)
|
||||
);
|
||||
if (existingAccept) return headers;
|
||||
|
||||
return {
|
||||
...headers,
|
||||
accept: 'text/event-stream',
|
||||
};
|
||||
}
|
||||
|
||||
function toFiniteNumber(value: unknown): number | null {
|
||||
return typeof value === 'number' && Number.isFinite(value) ? value : null;
|
||||
}
|
||||
|
||||
function sanitizeMessagesBodyForAnthropic(body: Record<string, unknown>): Record<string, unknown> {
|
||||
const sanitized: Record<string, unknown> = { ...body };
|
||||
const hasTemperature = toFiniteNumber(sanitized.temperature) !== null;
|
||||
const hasTopP = toFiniteNumber(sanitized.top_p) !== null;
|
||||
// Some Anthropic-compatible upstreams reject requests carrying both fields.
|
||||
if (hasTemperature && hasTopP) {
|
||||
delete sanitized.top_p;
|
||||
}
|
||||
return sanitized;
|
||||
}
|
||||
|
||||
function normalizeContentText(content: unknown): string {
|
||||
if (typeof content === 'string') return content;
|
||||
if (Array.isArray(content)) {
|
||||
@@ -173,8 +314,7 @@ function normalizeEndpointTypes(value: unknown): UpstreamEndpoint[] {
|
||||
|| raw === 'responses'
|
||||
|| raw.includes('response')
|
||||
) {
|
||||
// Treat responses family as OpenAI-compatible for upstream selection.
|
||||
normalized.add('chat');
|
||||
normalized.add('responses');
|
||||
}
|
||||
|
||||
if (
|
||||
@@ -191,12 +331,17 @@ function normalizeEndpointTypes(value: unknown): UpstreamEndpoint[] {
|
||||
// Some upstreams return protocol families instead of concrete endpoint paths.
|
||||
if (raw === 'openai' || raw.includes('openai')) {
|
||||
normalized.add('chat');
|
||||
normalized.add('responses');
|
||||
}
|
||||
|
||||
return Array.from(normalized);
|
||||
}
|
||||
|
||||
function preferredEndpointOrder(downstreamFormat: DownstreamFormat): UpstreamEndpoint[] {
|
||||
function preferredEndpointOrder(downstreamFormat: EndpointPreference): UpstreamEndpoint[] {
|
||||
if (downstreamFormat === 'responses') {
|
||||
return ['responses', 'chat', 'messages'];
|
||||
}
|
||||
|
||||
return downstreamFormat === 'claude'
|
||||
? ['messages', 'chat']
|
||||
: ['chat', 'messages'];
|
||||
@@ -205,11 +350,13 @@ function preferredEndpointOrder(downstreamFormat: DownstreamFormat): UpstreamEnd
|
||||
export async function resolveUpstreamEndpointCandidates(
|
||||
context: ChannelContext,
|
||||
modelName: string,
|
||||
downstreamFormat: DownstreamFormat,
|
||||
downstreamFormat: EndpointPreference,
|
||||
): Promise<UpstreamEndpoint[]> {
|
||||
if ((context.site.platform || '').toLowerCase() === 'anyrouter') {
|
||||
// anyrouter deployments are effectively anthropic-protocol first.
|
||||
return ['messages', 'chat'];
|
||||
return downstreamFormat === 'responses'
|
||||
? ['responses', 'messages', 'chat']
|
||||
: ['messages', 'chat'];
|
||||
}
|
||||
|
||||
const preferred = preferredEndpointOrder(downstreamFormat);
|
||||
@@ -269,39 +416,86 @@ export function buildUpstreamEndpointRequest(input: {
|
||||
stream: boolean;
|
||||
tokenValue: string;
|
||||
openaiBody: Record<string, unknown>;
|
||||
downstreamFormat: EndpointPreference;
|
||||
claudeOriginalBody?: Record<string, unknown>;
|
||||
responsesOriginalBody?: Record<string, unknown>;
|
||||
downstreamHeaders?: Record<string, unknown>;
|
||||
}): {
|
||||
path: string;
|
||||
headers: Record<string, string>;
|
||||
body: Record<string, unknown>;
|
||||
} {
|
||||
const passthroughHeaders = extractSafePassthroughHeaders(input.downstreamHeaders);
|
||||
const commonHeaders = {
|
||||
...passthroughHeaders,
|
||||
'Content-Type': 'application/json',
|
||||
Authorization: `Bearer ${input.tokenValue}`,
|
||||
};
|
||||
|
||||
if (input.endpoint === 'messages') {
|
||||
const claudeHeaders = input.downstreamFormat === 'claude'
|
||||
? extractClaudePassthroughHeaders(input.downstreamHeaders)
|
||||
: {};
|
||||
const anthropicVersion = (
|
||||
claudeHeaders['anthropic-version']
|
||||
|| passthroughHeaders['anthropic-version']
|
||||
|| '2023-06-01'
|
||||
);
|
||||
const body = (
|
||||
input.downstreamFormat === 'claude' && input.claudeOriginalBody
|
||||
? {
|
||||
...input.claudeOriginalBody,
|
||||
model: input.modelName,
|
||||
stream: input.stream,
|
||||
}
|
||||
: convertOpenAiBodyToMessagesBody(input.openaiBody, input.modelName, input.stream)
|
||||
);
|
||||
const sanitizedBody = sanitizeMessagesBodyForAnthropic(body);
|
||||
|
||||
const headers = ensureStreamAcceptHeader({
|
||||
...commonHeaders,
|
||||
...claudeHeaders,
|
||||
'x-api-key': input.tokenValue,
|
||||
'anthropic-version': anthropicVersion,
|
||||
}, input.stream);
|
||||
|
||||
return {
|
||||
path: '/v1/messages',
|
||||
headers: {
|
||||
...commonHeaders,
|
||||
'x-api-key': input.tokenValue,
|
||||
'anthropic-version': '2023-06-01',
|
||||
},
|
||||
body: convertOpenAiBodyToMessagesBody(input.openaiBody, input.modelName, input.stream),
|
||||
headers,
|
||||
body: sanitizedBody,
|
||||
};
|
||||
}
|
||||
|
||||
if (input.endpoint === 'responses') {
|
||||
const responsesHeaders = input.downstreamFormat === 'responses'
|
||||
? extractResponsesPassthroughHeaders(input.downstreamHeaders)
|
||||
: {};
|
||||
const body = (
|
||||
input.downstreamFormat === 'responses' && input.responsesOriginalBody
|
||||
? {
|
||||
...input.responsesOriginalBody,
|
||||
model: input.modelName,
|
||||
stream: input.stream,
|
||||
}
|
||||
: convertOpenAiBodyToResponsesBody(input.openaiBody, input.modelName, input.stream)
|
||||
);
|
||||
|
||||
const headers = ensureStreamAcceptHeader({
|
||||
...commonHeaders,
|
||||
...responsesHeaders,
|
||||
}, input.stream);
|
||||
|
||||
return {
|
||||
path: '/v1/responses',
|
||||
headers: commonHeaders,
|
||||
body: convertOpenAiBodyToResponsesBody(input.openaiBody, input.modelName, input.stream),
|
||||
headers,
|
||||
body,
|
||||
};
|
||||
}
|
||||
|
||||
const headers = ensureStreamAcceptHeader(commonHeaders, input.stream);
|
||||
return {
|
||||
path: '/v1/chat/completions',
|
||||
headers: commonHeaders,
|
||||
headers,
|
||||
body: {
|
||||
...input.openaiBody,
|
||||
model: input.modelName,
|
||||
|
||||
Reference in New Issue
Block a user