mirror of
https://github.com/tiennm99/ccs.git
synced 2026-10-14 12:16:38 +00:00
feat(cliproxy): add ToolSanitizationProxy for Gemini 64-char limit
Fixes #219 - MCP tool names exceeding Gemini's 64-character limit now get sanitized automatically. - Add tool-name-sanitizer: dedupe segments + smart truncate with hash - Add tool-name-mapper: bidirectional mapping for response restoration - Add tool-sanitization-proxy: HTTP proxy layer for all CLIProxy providers - Chain: Claude CLI → ToolSanitizationProxy → [CodexReasoningProxy] → CLIProxy - Add unit tests for sanitizer and mapper modules
This commit is contained in:
1 parent
7f7dcb21f9
commit
63633507d2
7 files changed
+1251
-15
No files matched your search
@@ -38,6 +38,7 @@ import { resolveProxyConfig, PROXY_CLI_FLAGS } from './proxy-config-resolver';
|
||||
import { getWebSearchHookEnv } from '../utils/websearch-manager';
|
||||
import { supportsModelConfig, isModelBroken, getModelIssueUrl, findModel } from './model-catalog';
|
||||
import { CodexReasoningProxy } from './codex-reasoning-proxy';
|
||||
import { ToolSanitizationProxy } from './tool-sanitization-proxy';
|
||||
import {
|
||||
findAccountByQuery,
|
||||
getProviderAccounts,
|
||||
@@ -841,6 +842,37 @@ export async function execClaudeWithCLIProxy(
|
||||
// This adds thinking suffix like model(high) or model(8192) for CLIProxyAPIPlus
|
||||
applyThinkingConfig(envVars, provider, thinkingOverride);
|
||||
|
||||
// Tool sanitization proxy - applies to ALL CLIProxy providers.
|
||||
// Sanitizes MCP tool names exceeding Gemini's 64-char limit.
|
||||
// Chain: Claude CLI → ToolSanitizationProxy → [CodexReasoningProxy] → CLIProxy → Gemini
|
||||
let toolSanitizationProxy: ToolSanitizationProxy | null = null;
|
||||
let toolSanitizationPort: number | null = null;
|
||||
|
||||
if (envVars.ANTHROPIC_BASE_URL) {
|
||||
try {
|
||||
toolSanitizationProxy = new ToolSanitizationProxy({
|
||||
upstreamBaseUrl: envVars.ANTHROPIC_BASE_URL,
|
||||
verbose,
|
||||
warnOnSanitize: true,
|
||||
});
|
||||
toolSanitizationPort = await toolSanitizationProxy.start();
|
||||
log(`Tool sanitization proxy active on port ${toolSanitizationPort}`);
|
||||
} catch (error) {
|
||||
// Non-fatal: continue without sanitization
|
||||
const err = error as Error;
|
||||
toolSanitizationProxy = null;
|
||||
toolSanitizationPort = null;
|
||||
if (verbose) {
|
||||
console.error(warn(`Tool sanitization proxy disabled: ${err.message}`));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Determine the effective upstream URL after tool sanitization
|
||||
const postSanitizationBaseUrl = toolSanitizationPort
|
||||
? `http://127.0.0.1:${toolSanitizationPort}`
|
||||
: envVars.ANTHROPIC_BASE_URL;
|
||||
|
||||
// Codex-only: inject OpenAI reasoning effort based on tier model mapping.
|
||||
// Maps by request.model:
|
||||
// - OPUS/default model → xhigh
|
||||
@@ -850,7 +882,7 @@ export async function execClaudeWithCLIProxy(
|
||||
let codexReasoningProxy: CodexReasoningProxy | null = null;
|
||||
let codexReasoningPort: number | null = null;
|
||||
if (provider === 'codex') {
|
||||
if (!envVars.ANTHROPIC_BASE_URL) {
|
||||
if (!postSanitizationBaseUrl) {
|
||||
log('ANTHROPIC_BASE_URL not set for Codex, reasoning proxy disabled');
|
||||
} else {
|
||||
try {
|
||||
@@ -861,7 +893,7 @@ export async function execClaudeWithCLIProxy(
|
||||
// because remote CLIProxyAPI uses root paths (/v1/messages), not provider-prefixed
|
||||
const stripPathPrefix = useRemoteProxy ? '/api/provider/codex' : undefined;
|
||||
codexReasoningProxy = new CodexReasoningProxy({
|
||||
upstreamBaseUrl: envVars.ANTHROPIC_BASE_URL,
|
||||
upstreamBaseUrl: postSanitizationBaseUrl,
|
||||
verbose,
|
||||
defaultEffort: 'medium',
|
||||
traceFilePath: traceEnabled
|
||||
@@ -890,13 +922,22 @@ export async function execClaudeWithCLIProxy(
|
||||
}
|
||||
}
|
||||
|
||||
const effectiveEnvVars =
|
||||
codexReasoningProxy && codexReasoningPort
|
||||
? {
|
||||
...envVars,
|
||||
ANTHROPIC_BASE_URL: `http://127.0.0.1:${codexReasoningPort}/api/provider/codex`,
|
||||
}
|
||||
: envVars;
|
||||
// Determine the final ANTHROPIC_BASE_URL based on active proxies
|
||||
// Chain order: Claude CLI → [CodexReasoningProxy] → [ToolSanitizationProxy] → CLIProxy
|
||||
// The outermost active proxy becomes the effective base URL
|
||||
let finalBaseUrl = envVars.ANTHROPIC_BASE_URL;
|
||||
if (toolSanitizationPort) {
|
||||
finalBaseUrl = `http://127.0.0.1:${toolSanitizationPort}`;
|
||||
}
|
||||
if (codexReasoningPort) {
|
||||
// Codex reasoning proxy is the outermost layer for codex provider
|
||||
finalBaseUrl = `http://127.0.0.1:${codexReasoningPort}/api/provider/codex`;
|
||||
}
|
||||
|
||||
const effectiveEnvVars = {
|
||||
...envVars,
|
||||
ANTHROPIC_BASE_URL: finalBaseUrl,
|
||||
};
|
||||
const webSearchEnv = getWebSearchHookEnv();
|
||||
const env = {
|
||||
...process.env,
|
||||
@@ -906,16 +947,12 @@ export async function execClaudeWithCLIProxy(
|
||||
};
|
||||
|
||||
log(`Claude env: ANTHROPIC_BASE_URL=${effectiveEnvVars.ANTHROPIC_BASE_URL}`);
|
||||
log(`Claude env: ANTHROPIC_MODEL=${effectiveEnvVars.ANTHROPIC_MODEL}`);
|
||||
log(`Claude env: ANTHROPIC_MODEL=${envVars.ANTHROPIC_MODEL}`);
|
||||
if (Object.keys(webSearchEnv).length > 0) {
|
||||
log(`Claude env: WebSearch config=${JSON.stringify(webSearchEnv)}`);
|
||||
}
|
||||
// Log global env vars for visibility
|
||||
if (
|
||||
effectiveEnvVars.DISABLE_TELEMETRY ||
|
||||
effectiveEnvVars.DISABLE_ERROR_REPORTING ||
|
||||
effectiveEnvVars.DISABLE_BUG_COMMAND
|
||||
) {
|
||||
if (envVars.DISABLE_TELEMETRY || envVars.DISABLE_ERROR_REPORTING || envVars.DISABLE_BUG_COMMAND) {
|
||||
log(`Claude env: Global env applied (telemetry/reporting disabled)`);
|
||||
}
|
||||
|
||||
@@ -983,6 +1020,10 @@ export async function execClaudeWithCLIProxy(
|
||||
codexReasoningProxy.stop();
|
||||
}
|
||||
|
||||
if (toolSanitizationProxy) {
|
||||
toolSanitizationProxy.stop();
|
||||
}
|
||||
|
||||
if (httpsTunnel) {
|
||||
httpsTunnel.stop();
|
||||
}
|
||||
@@ -1007,6 +1048,10 @@ export async function execClaudeWithCLIProxy(
|
||||
codexReasoningProxy.stop();
|
||||
}
|
||||
|
||||
if (toolSanitizationProxy) {
|
||||
toolSanitizationProxy.stop();
|
||||
}
|
||||
|
||||
if (httpsTunnel) {
|
||||
httpsTunnel.stop();
|
||||
}
|
||||
@@ -1026,6 +1071,10 @@ export async function execClaudeWithCLIProxy(
|
||||
codexReasoningProxy.stop();
|
||||
}
|
||||
|
||||
if (toolSanitizationProxy) {
|
||||
toolSanitizationProxy.stop();
|
||||
}
|
||||
|
||||
if (httpsTunnel) {
|
||||
httpsTunnel.stop();
|
||||
}
|
||||
|
||||
@@ -189,3 +189,19 @@ export {
|
||||
getSyncableProfileCount,
|
||||
isProfileSyncable,
|
||||
} from './sync';
|
||||
|
||||
// Tool name sanitization (for Gemini 64-char limit compliance)
|
||||
export type { SanitizeResult } from './tool-name-sanitizer';
|
||||
export {
|
||||
sanitizeToolName,
|
||||
isValidToolName,
|
||||
removeDuplicateSegments,
|
||||
smartTruncate,
|
||||
GEMINI_MAX_TOOL_NAME_LENGTH,
|
||||
} from './tool-name-sanitizer';
|
||||
|
||||
export type { Tool, ToolUseBlock, ContentBlock, SanitizationChange } from './tool-name-mapper';
|
||||
export { ToolNameMapper } from './tool-name-mapper';
|
||||
|
||||
export type { ToolSanitizationProxyConfig } from './tool-sanitization-proxy';
|
||||
export { ToolSanitizationProxy } from './tool-sanitization-proxy';
|
||||
@@ -0,0 +1,153 @@
|
||||
/**
|
||||
* Tool Name Mapper
|
||||
*
|
||||
* Bidirectional mapping class to track original ↔ sanitized tool names.
|
||||
* Used to restore original names in API responses after sanitization.
|
||||
*
|
||||
* Flow:
|
||||
* 1. Request: registerTools() sanitizes names and stores mapping
|
||||
* 2. Response: restoreToolUse() restores original names using mapping
|
||||
*/
|
||||
|
||||
import { sanitizeToolName, type SanitizeResult } from './tool-name-sanitizer';
|
||||
|
||||
/** MCP tool definition from Claude API */
|
||||
export interface Tool {
|
||||
name: string;
|
||||
description?: string;
|
||||
input_schema?: Record<string, unknown>;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
/** Tool use content block from Claude API response */
|
||||
export interface ToolUseBlock {
|
||||
type: 'tool_use';
|
||||
id: string;
|
||||
name: string;
|
||||
input: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** Content block from Claude API response (union type) */
|
||||
export interface ContentBlock {
|
||||
type: string;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
/** Record of a sanitization change */
|
||||
export interface SanitizationChange {
|
||||
original: string;
|
||||
sanitized: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Bidirectional mapper for tool name sanitization.
|
||||
*
|
||||
* Maintains a per-request mapping between sanitized and original tool names.
|
||||
* Must be cleared between requests to avoid memory leaks.
|
||||
*/
|
||||
export class ToolNameMapper {
|
||||
/** Map from sanitized name → original name */
|
||||
private mapping: Map<string, string> = new Map();
|
||||
|
||||
/** List of all changes made during registration */
|
||||
private changes: SanitizationChange[] = [];
|
||||
|
||||
/**
|
||||
* Register tools and sanitize their names.
|
||||
* Stores mapping for later restoration.
|
||||
*
|
||||
* @param tools Array of tool definitions
|
||||
* @returns Array of tools with sanitized names
|
||||
*/
|
||||
registerTools(tools: Tool[]): Tool[] {
|
||||
return tools.map((tool) => {
|
||||
const result: SanitizeResult = sanitizeToolName(tool.name);
|
||||
|
||||
if (result.changed) {
|
||||
// Store mapping: sanitized → original
|
||||
this.mapping.set(result.sanitized, tool.name);
|
||||
this.changes.push({
|
||||
original: tool.name,
|
||||
sanitized: result.sanitized,
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
...tool,
|
||||
name: result.sanitized,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Restore original tool names in content blocks.
|
||||
* Looks for tool_use blocks and restores their names.
|
||||
*
|
||||
* @param content Array of content blocks from API response
|
||||
* @returns Array with restored tool names
|
||||
*/
|
||||
restoreToolUse(content: ContentBlock[]): ContentBlock[] {
|
||||
return content.map((block) => {
|
||||
if (block.type !== 'tool_use') {
|
||||
return block;
|
||||
}
|
||||
|
||||
// Type guard: we know this is a tool_use block
|
||||
const toolUseName = block.name as string | undefined;
|
||||
if (!toolUseName) {
|
||||
return block;
|
||||
}
|
||||
|
||||
const originalName = this.mapping.get(toolUseName);
|
||||
|
||||
if (originalName) {
|
||||
return {
|
||||
...block,
|
||||
name: originalName,
|
||||
};
|
||||
}
|
||||
|
||||
return block;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Restore tool name in a single tool_use block.
|
||||
* Useful for streaming responses.
|
||||
*
|
||||
* @param name The sanitized tool name
|
||||
* @returns Original name if mapped, otherwise the input name
|
||||
*/
|
||||
restoreName(name: string): string {
|
||||
return this.mapping.get(name) ?? name;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if any sanitization occurred during registration.
|
||||
*/
|
||||
hasChanges(): boolean {
|
||||
return this.changes.length > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all sanitization changes for logging.
|
||||
*/
|
||||
getChanges(): SanitizationChange[] {
|
||||
return [...this.changes];
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of tools that were sanitized.
|
||||
*/
|
||||
getChangeCount(): number {
|
||||
return this.changes.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all mappings. Call between requests.
|
||||
*/
|
||||
clear(): void {
|
||||
this.mapping.clear();
|
||||
this.changes = [];
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
/**
|
||||
* Tool Name Sanitizer
|
||||
*
|
||||
* Sanitizes MCP tool names to comply with Gemini API constraints:
|
||||
* - Max 64 characters
|
||||
* - Must start with letter or underscore
|
||||
* - Only a-z A-Z 0-9 _ . : - allowed
|
||||
*
|
||||
* Strategies:
|
||||
* 1. Remove duplicate segments (e.g., gitmcp__foo__foo → gitmcp__foo)
|
||||
* 2. Smart truncate with hash suffix if still >64 chars
|
||||
*/
|
||||
|
||||
import { createHash } from 'crypto';
|
||||
|
||||
/** Maximum tool name length allowed by Gemini API */
|
||||
export const GEMINI_MAX_TOOL_NAME_LENGTH = 64;
|
||||
|
||||
/** Valid characters pattern for Gemini tool names */
|
||||
const VALID_CHARS_REGEX = /^[a-zA-Z_][a-zA-Z0-9_.:/-]*$/;
|
||||
|
||||
/** Result of sanitization operation */
|
||||
export interface SanitizeResult {
|
||||
/** The sanitized tool name */
|
||||
sanitized: string;
|
||||
/** Whether the name was changed */
|
||||
changed: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a tool name is valid for Gemini API.
|
||||
*
|
||||
* Requirements:
|
||||
* - Length <= 64 characters
|
||||
* - Starts with letter or underscore
|
||||
* - Contains only valid characters: a-z A-Z 0-9 _ . : -
|
||||
*/
|
||||
export function isValidToolName(name: string): boolean {
|
||||
if (!name || name.length === 0) {
|
||||
return false;
|
||||
}
|
||||
if (name.length > GEMINI_MAX_TOOL_NAME_LENGTH) {
|
||||
return false;
|
||||
}
|
||||
return VALID_CHARS_REGEX.test(name);
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove consecutive duplicate segments separated by '__'.
|
||||
*
|
||||
* Examples:
|
||||
* - 'gitmcp__foo__foo' → 'gitmcp__foo'
|
||||
* - 'a__b__c__b__c' → 'a__b__c'
|
||||
* - 'no_dupes' → 'no_dupes'
|
||||
*/
|
||||
export function removeDuplicateSegments(name: string): string {
|
||||
const segments = name.split('__');
|
||||
const deduped: string[] = [];
|
||||
|
||||
for (const segment of segments) {
|
||||
// Only add if different from previous segment
|
||||
if (deduped.length === 0 || deduped[deduped.length - 1] !== segment) {
|
||||
deduped.push(segment);
|
||||
}
|
||||
}
|
||||
|
||||
return deduped.join('__');
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate a short hash from a string for truncation suffix.
|
||||
* Uses first 6 characters of MD5 hash (16M combinations).
|
||||
*/
|
||||
function generateShortHash(input: string): string {
|
||||
return createHash('md5').update(input).digest('hex').slice(0, 6);
|
||||
}
|
||||
|
||||
/**
|
||||
* Smart truncate a name to fit within maxLen.
|
||||
* Preserves start of name and appends hash suffix for uniqueness.
|
||||
*
|
||||
* Format: <prefix>_<6-char-hash>
|
||||
*
|
||||
* @param name The name to truncate
|
||||
* @param maxLen Maximum allowed length (default: 64)
|
||||
*/
|
||||
export function smartTruncate(name: string, maxLen: number = GEMINI_MAX_TOOL_NAME_LENGTH): string {
|
||||
if (name.length <= maxLen) {
|
||||
return name;
|
||||
}
|
||||
|
||||
// Format: prefix + '_' + 6-char hash = 7 chars for suffix
|
||||
const hash = generateShortHash(name);
|
||||
const prefixLen = maxLen - 7; // 7 = '_' (1) + hash (6)
|
||||
const prefix = name.slice(0, prefixLen);
|
||||
|
||||
return `${prefix}_${hash}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sanitize a tool name to comply with Gemini API constraints.
|
||||
*
|
||||
* Process:
|
||||
* 1. Remove duplicate segments (always, as duplicates are likely unintentional)
|
||||
* 2. Truncate with hash if >64 chars
|
||||
* 3. Return result with changed flag
|
||||
*
|
||||
* @param name The original tool name
|
||||
* @returns Sanitization result with sanitized name and changed flag
|
||||
*/
|
||||
export function sanitizeToolName(name: string): SanitizeResult {
|
||||
// Step 1: Always try to remove duplicate segments
|
||||
// Duplicates like gitmcp__foo__foo are likely unintentional from MCP naming
|
||||
let sanitized = removeDuplicateSegments(name);
|
||||
|
||||
// Step 2: Truncate if still too long
|
||||
if (sanitized.length > GEMINI_MAX_TOOL_NAME_LENGTH) {
|
||||
sanitized = smartTruncate(sanitized);
|
||||
}
|
||||
|
||||
// Step 3: If still invalid after sanitization, apply truncation to original
|
||||
if (!isValidToolName(sanitized)) {
|
||||
sanitized = smartTruncate(name);
|
||||
}
|
||||
|
||||
return {
|
||||
sanitized,
|
||||
changed: sanitized !== name,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,457 @@
|
||||
/**
|
||||
* Tool Sanitization Proxy
|
||||
*
|
||||
* HTTP proxy that intercepts Claude CLI → CLIProxy requests to:
|
||||
* 1. Sanitize MCP tool names exceeding Gemini's 64-char limit
|
||||
* 2. Forward sanitized requests to upstream
|
||||
* 3. Restore original names in responses
|
||||
*
|
||||
* Follows CodexReasoningProxy pattern for consistency.
|
||||
*/
|
||||
|
||||
import * as http from 'http';
|
||||
import * as https from 'https';
|
||||
import { URL } from 'url';
|
||||
import { ToolNameMapper, type Tool, type ContentBlock } from './tool-name-mapper';
|
||||
|
||||
export interface ToolSanitizationProxyConfig {
|
||||
/** Upstream CLIProxy URL */
|
||||
upstreamBaseUrl: string;
|
||||
/** Enable verbose logging */
|
||||
verbose?: boolean;
|
||||
/** Log warnings when sanitization occurs */
|
||||
warnOnSanitize?: boolean;
|
||||
/** Request timeout in milliseconds */
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
export class ToolSanitizationProxy {
|
||||
private server: http.Server | null = null;
|
||||
private port: number | null = null;
|
||||
private readonly config: Required<ToolSanitizationProxyConfig>;
|
||||
|
||||
constructor(config: ToolSanitizationProxyConfig) {
|
||||
this.config = {
|
||||
upstreamBaseUrl: config.upstreamBaseUrl,
|
||||
verbose: config.verbose ?? false,
|
||||
warnOnSanitize: config.warnOnSanitize ?? true,
|
||||
timeoutMs: config.timeoutMs ?? 120000,
|
||||
};
|
||||
}
|
||||
|
||||
private log(message: string): void {
|
||||
if (this.config.verbose) {
|
||||
console.error(`[tool-sanitization-proxy] ${message}`);
|
||||
}
|
||||
}
|
||||
|
||||
private warn(message: string): void {
|
||||
if (this.config.warnOnSanitize) {
|
||||
console.error(`[!] ${message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the proxy server on an ephemeral port.
|
||||
* @returns The assigned port number
|
||||
*/
|
||||
async start(): Promise<number> {
|
||||
if (this.server) return this.port ?? 0;
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
this.server = http.createServer((req, res) => {
|
||||
void this.handleRequest(req, res);
|
||||
});
|
||||
|
||||
this.server.listen(0, '127.0.0.1', () => {
|
||||
const address = this.server?.address();
|
||||
this.port = typeof address === 'object' && address ? address.port : 0;
|
||||
resolve(this.port);
|
||||
});
|
||||
|
||||
this.server.on('error', (err) => reject(err));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the proxy server.
|
||||
*/
|
||||
stop(): void {
|
||||
if (!this.server) return;
|
||||
this.server.close();
|
||||
this.server = null;
|
||||
this.port = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the port the proxy is listening on.
|
||||
*/
|
||||
getPort(): number | null {
|
||||
return this.port;
|
||||
}
|
||||
|
||||
private readBody(req: http.IncomingMessage): Promise<string> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const chunks: Buffer[] = [];
|
||||
const maxSize = 10 * 1024 * 1024; // 10MB
|
||||
let total = 0;
|
||||
|
||||
req.on('data', (chunk: Buffer) => {
|
||||
total += chunk.length;
|
||||
if (total > maxSize) {
|
||||
req.destroy();
|
||||
reject(new Error('Request body too large (max 10MB)'));
|
||||
return;
|
||||
}
|
||||
chunks.push(chunk);
|
||||
});
|
||||
|
||||
req.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')));
|
||||
req.on('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
private async handleRequest(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
const method = req.method || 'GET';
|
||||
const requestPath = req.url || '/';
|
||||
const upstreamBase = new URL(this.config.upstreamBaseUrl);
|
||||
const fullUpstreamUrl = new URL(requestPath, upstreamBase);
|
||||
|
||||
this.log(`${method} ${requestPath} → ${fullUpstreamUrl.href}`);
|
||||
|
||||
// Only buffer+rewrite JSON POST requests
|
||||
const contentType = String(req.headers['content-type'] || '');
|
||||
const isJson = contentType.includes('application/json');
|
||||
const shouldRewrite = isJson && ['POST', 'PUT', 'PATCH'].includes(method.toUpperCase());
|
||||
|
||||
try {
|
||||
if (!shouldRewrite) {
|
||||
await this.forwardRaw(req, res, fullUpstreamUrl);
|
||||
return;
|
||||
}
|
||||
|
||||
const rawBody = await this.readBody(req);
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = rawBody.length ? JSON.parse(rawBody) : {};
|
||||
} catch {
|
||||
res.writeHead(400, { 'Content-Type': 'application/json' });
|
||||
res.end(JSON.stringify({ error: 'Invalid JSON in request body' }));
|
||||
return;
|
||||
}
|
||||
|
||||
// Create mapper for this request
|
||||
const mapper = new ToolNameMapper();
|
||||
|
||||
// Sanitize tools if present
|
||||
let modifiedBody = parsed;
|
||||
if (isRecord(parsed) && Array.isArray(parsed.tools)) {
|
||||
const sanitizedTools = mapper.registerTools(parsed.tools as Tool[]);
|
||||
modifiedBody = { ...parsed, tools: sanitizedTools };
|
||||
|
||||
// Log sanitization warnings
|
||||
if (mapper.hasChanges()) {
|
||||
const changes = mapper.getChanges();
|
||||
for (const change of changes) {
|
||||
this.warn(`Tool name sanitized: "${change.original}" → "${change.sanitized}"`);
|
||||
}
|
||||
this.log(`Sanitized ${changes.length} tool name(s)`);
|
||||
}
|
||||
}
|
||||
|
||||
// Check if streaming is requested
|
||||
const isStreaming = isRecord(parsed) && parsed.stream === true;
|
||||
|
||||
if (isStreaming) {
|
||||
await this.forwardJsonStreaming(req, res, fullUpstreamUrl, modifiedBody, mapper);
|
||||
} else {
|
||||
await this.forwardJsonBuffered(req, res, fullUpstreamUrl, modifiedBody, mapper);
|
||||
}
|
||||
} catch (error) {
|
||||
const err = error as Error;
|
||||
this.log(`Error: ${err.message}`);
|
||||
if (!res.headersSent) {
|
||||
res.writeHead(502, { 'Content-Type': 'application/json' });
|
||||
}
|
||||
res.end(JSON.stringify({ error: err.message }));
|
||||
}
|
||||
}
|
||||
|
||||
private buildForwardHeaders(
|
||||
originalHeaders: http.IncomingHttpHeaders,
|
||||
bodyString?: string
|
||||
): http.OutgoingHttpHeaders {
|
||||
const headers: http.OutgoingHttpHeaders = {};
|
||||
|
||||
// RFC 7230 hop-by-hop headers that should not be forwarded
|
||||
const hopByHop = new Set([
|
||||
'host',
|
||||
'content-length',
|
||||
'connection',
|
||||
'transfer-encoding',
|
||||
'keep-alive',
|
||||
'proxy-authenticate',
|
||||
'proxy-authorization',
|
||||
'te',
|
||||
'trailer',
|
||||
'upgrade',
|
||||
]);
|
||||
|
||||
for (const [key, value] of Object.entries(originalHeaders)) {
|
||||
if (!value) continue;
|
||||
const lower = key.toLowerCase();
|
||||
if (hopByHop.has(lower)) continue;
|
||||
headers[key] = value;
|
||||
}
|
||||
|
||||
if (bodyString !== undefined) {
|
||||
headers['Content-Type'] = headers['Content-Type'] || 'application/json';
|
||||
headers['Content-Length'] = Buffer.byteLength(bodyString);
|
||||
}
|
||||
|
||||
return headers;
|
||||
}
|
||||
|
||||
private getRequestFn(url: URL): typeof http.request | typeof https.request {
|
||||
return url.protocol === 'https:' ? https.request : http.request;
|
||||
}
|
||||
|
||||
private forwardRaw(
|
||||
originalReq: http.IncomingMessage,
|
||||
clientRes: http.ServerResponse,
|
||||
upstreamUrl: URL
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const requestFn = this.getRequestFn(upstreamUrl);
|
||||
const upstreamReq = requestFn(
|
||||
{
|
||||
protocol: upstreamUrl.protocol,
|
||||
hostname: upstreamUrl.hostname,
|
||||
port: upstreamUrl.port,
|
||||
path: upstreamUrl.pathname + upstreamUrl.search,
|
||||
method: originalReq.method,
|
||||
timeout: this.config.timeoutMs,
|
||||
headers: this.buildForwardHeaders(originalReq.headers),
|
||||
},
|
||||
(upstreamRes) => {
|
||||
clientRes.writeHead(upstreamRes.statusCode || 200, upstreamRes.headers);
|
||||
upstreamRes.pipe(clientRes);
|
||||
upstreamRes.on('end', () => resolve());
|
||||
upstreamRes.on('error', reject);
|
||||
}
|
||||
);
|
||||
|
||||
upstreamReq.on('timeout', () => upstreamReq.destroy(new Error('Upstream request timeout')));
|
||||
upstreamReq.on('error', reject);
|
||||
originalReq.pipe(upstreamReq);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward JSON request and buffer response for tool name restoration.
|
||||
*/
|
||||
private forwardJsonBuffered(
|
||||
originalReq: http.IncomingMessage,
|
||||
clientRes: http.ServerResponse,
|
||||
upstreamUrl: URL,
|
||||
body: unknown,
|
||||
mapper: ToolNameMapper
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const bodyString = JSON.stringify(body);
|
||||
const requestFn = this.getRequestFn(upstreamUrl);
|
||||
const upstreamReq = requestFn(
|
||||
{
|
||||
protocol: upstreamUrl.protocol,
|
||||
hostname: upstreamUrl.hostname,
|
||||
port: upstreamUrl.port,
|
||||
path: upstreamUrl.pathname + upstreamUrl.search,
|
||||
method: originalReq.method,
|
||||
timeout: this.config.timeoutMs,
|
||||
headers: this.buildForwardHeaders(originalReq.headers, bodyString),
|
||||
},
|
||||
(upstreamRes) => {
|
||||
const chunks: Buffer[] = [];
|
||||
|
||||
upstreamRes.on('data', (chunk: Buffer) => chunks.push(chunk));
|
||||
upstreamRes.on('end', () => {
|
||||
try {
|
||||
const responseBody = Buffer.concat(chunks).toString('utf8');
|
||||
const contentType = upstreamRes.headers['content-type'] || '';
|
||||
|
||||
// Only process JSON responses with tool_use blocks
|
||||
if (contentType.includes('application/json') && mapper.hasChanges()) {
|
||||
try {
|
||||
const parsed = JSON.parse(responseBody);
|
||||
if (isRecord(parsed) && Array.isArray(parsed.content)) {
|
||||
parsed.content = mapper.restoreToolUse(parsed.content as ContentBlock[]);
|
||||
const modifiedResponse = JSON.stringify(parsed);
|
||||
|
||||
// Update content-length header
|
||||
const headers = { ...upstreamRes.headers };
|
||||
headers['content-length'] = String(Buffer.byteLength(modifiedResponse));
|
||||
|
||||
clientRes.writeHead(upstreamRes.statusCode || 200, headers);
|
||||
clientRes.end(modifiedResponse);
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
} catch {
|
||||
// JSON parse failed, pass through unchanged
|
||||
}
|
||||
}
|
||||
|
||||
// Pass through unchanged
|
||||
clientRes.writeHead(upstreamRes.statusCode || 200, upstreamRes.headers);
|
||||
clientRes.end(responseBody);
|
||||
resolve();
|
||||
} catch (err) {
|
||||
reject(err);
|
||||
}
|
||||
});
|
||||
upstreamRes.on('error', reject);
|
||||
}
|
||||
);
|
||||
|
||||
upstreamReq.on('timeout', () => upstreamReq.destroy(new Error('Upstream request timeout')));
|
||||
upstreamReq.on('error', reject);
|
||||
upstreamReq.write(bodyString);
|
||||
upstreamReq.end();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward JSON request and stream response with tool name restoration.
|
||||
* Handles SSE (Server-Sent Events) format.
|
||||
*/
|
||||
private forwardJsonStreaming(
|
||||
originalReq: http.IncomingMessage,
|
||||
clientRes: http.ServerResponse,
|
||||
upstreamUrl: URL,
|
||||
body: unknown,
|
||||
mapper: ToolNameMapper
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const bodyString = JSON.stringify(body);
|
||||
const requestFn = this.getRequestFn(upstreamUrl);
|
||||
const upstreamReq = requestFn(
|
||||
{
|
||||
protocol: upstreamUrl.protocol,
|
||||
hostname: upstreamUrl.hostname,
|
||||
port: upstreamUrl.port,
|
||||
path: upstreamUrl.pathname + upstreamUrl.search,
|
||||
method: originalReq.method,
|
||||
timeout: this.config.timeoutMs,
|
||||
headers: this.buildForwardHeaders(originalReq.headers, bodyString),
|
||||
},
|
||||
(upstreamRes) => {
|
||||
clientRes.writeHead(upstreamRes.statusCode || 200, upstreamRes.headers);
|
||||
|
||||
// If no changes were made, just pipe through
|
||||
if (!mapper.hasChanges()) {
|
||||
upstreamRes.pipe(clientRes);
|
||||
upstreamRes.on('end', () => resolve());
|
||||
upstreamRes.on('error', reject);
|
||||
return;
|
||||
}
|
||||
|
||||
// Process SSE events for tool name restoration
|
||||
let buffer = '';
|
||||
|
||||
upstreamRes.on('data', (chunk: Buffer) => {
|
||||
buffer += chunk.toString('utf8');
|
||||
|
||||
// Process complete SSE events
|
||||
const events = buffer.split('\n\n');
|
||||
buffer = events.pop() || ''; // Keep incomplete event in buffer
|
||||
|
||||
for (const event of events) {
|
||||
if (!event.trim()) continue;
|
||||
|
||||
const processedEvent = this.processSSEEvent(event, mapper);
|
||||
clientRes.write(processedEvent + '\n\n');
|
||||
}
|
||||
});
|
||||
|
||||
upstreamRes.on('end', () => {
|
||||
// Process any remaining buffer
|
||||
if (buffer.trim()) {
|
||||
const processedEvent = this.processSSEEvent(buffer, mapper);
|
||||
clientRes.write(processedEvent + '\n\n');
|
||||
}
|
||||
clientRes.end();
|
||||
resolve();
|
||||
});
|
||||
|
||||
upstreamRes.on('error', reject);
|
||||
}
|
||||
);
|
||||
|
||||
upstreamReq.on('timeout', () => upstreamReq.destroy(new Error('Upstream request timeout')));
|
||||
upstreamReq.on('error', reject);
|
||||
upstreamReq.write(bodyString);
|
||||
upstreamReq.end();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a single SSE event, restoring tool names if present.
|
||||
*/
|
||||
private processSSEEvent(event: string, mapper: ToolNameMapper): string {
|
||||
// Parse SSE format: data: {...}
|
||||
const lines = event.split('\n');
|
||||
const processedLines: string[] = [];
|
||||
|
||||
for (const line of lines) {
|
||||
if (line.startsWith('data: ')) {
|
||||
const jsonStr = line.slice(6); // Remove 'data: ' prefix
|
||||
|
||||
// Skip [DONE] marker
|
||||
if (jsonStr.trim() === '[DONE]') {
|
||||
processedLines.push(line);
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const data = JSON.parse(jsonStr);
|
||||
|
||||
// Handle content_block_start with tool_use
|
||||
if (
|
||||
isRecord(data) &&
|
||||
data.type === 'content_block_start' &&
|
||||
isRecord(data.content_block) &&
|
||||
data.content_block.type === 'tool_use' &&
|
||||
typeof data.content_block.name === 'string'
|
||||
) {
|
||||
const originalName = mapper.restoreName(data.content_block.name);
|
||||
data.content_block.name = originalName;
|
||||
processedLines.push('data: ' + JSON.stringify(data));
|
||||
continue;
|
||||
}
|
||||
|
||||
// Handle message with content array (final message)
|
||||
if (isRecord(data) && Array.isArray(data.content)) {
|
||||
data.content = mapper.restoreToolUse(data.content as ContentBlock[]);
|
||||
processedLines.push('data: ' + JSON.stringify(data));
|
||||
continue;
|
||||
}
|
||||
|
||||
// Pass through unchanged
|
||||
processedLines.push(line);
|
||||
} catch {
|
||||
// Not valid JSON, pass through unchanged
|
||||
processedLines.push(line);
|
||||
}
|
||||
} else {
|
||||
// Non-data lines (event:, id:, etc.) pass through
|
||||
processedLines.push(line);
|
||||
}
|
||||
}
|
||||
|
||||
return processedLines.join('\n');
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user