from __future__ import annotations import base64 import hashlib import json import os import re import time import uuid from dataclasses import dataclass, field from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from urllib import error, parse, request class ConfigError(ValueError): """Raised when configuration is invalid.""" class UpstreamHTTPError(Exception): def __init__(self, status_code: int, body: bytes, headers: dict[str, str]) -> None: super().__init__(f"upstream returned {status_code}") self.status_code = status_code self.body = body self.headers = headers def _parse_bool(value: str | None, default: bool) -> bool: if value is None or value == "": return default normalized = value.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False raise ConfigError(f"invalid boolean value: {value}") def _parse_int(value: str | None, default: int) -> int: if value is None or value == "": return default try: return int(value) except ValueError as exc: raise ConfigError(f"invalid integer value: {value}") from exc def _parse_json_object(value: str | None, default: dict[str, Any]) -> dict[str, Any]: if value is None or value.strip() == "": return dict(default) try: parsed = json.loads(value) except json.JSONDecodeError as exc: raise ConfigError(f"invalid JSON object: {value}") from exc if not isinstance(parsed, dict): raise ConfigError("expected a JSON object") return parsed def _parse_json_string_list(value: str | None, default: list[str]) -> list[str]: if value is None or value.strip() == "": return list(default) try: parsed = json.loads(value) except json.JSONDecodeError as exc: raise ConfigError(f"invalid JSON list: {value}") from exc if not isinstance(parsed, list): raise ConfigError("expected a JSON array") items: list[str] = [] for item in parsed: if not isinstance(item, str) or not item.strip(): raise ConfigError("JSON string list entries must be non-empty strings") items.append(item.strip()) return items def _normalize_requested_model_name(value: str) -> str: normalized = value.strip() # Claude Code can append local context-window hints such as "[1m]". # Upstream OpenAI-compatible providers typically do not recognize them. while normalized.endswith("[1m]"): normalized = normalized[:-4].rstrip() return normalized @dataclass class AppConfig: host: str = "0.0.0.0" port: int = 8080 upstream_base_url: str = "http://127.0.0.1:3000" upstream_timeout_seconds: int = 240 upstream_auth_header: str | None = None upstream_extra_body: dict[str, Any] = field(default_factory=dict) model_map: dict[str, str] = field(default_factory=dict) allow_unmapped_model_passthrough: bool = True native_tool_models: set[str] = field(default_factory=set) public_model_ids: list[str] = field(default_factory=list) tool_prompt_preamble: str = "You are a tool-using assistant operating on the user's real local machine through a compatibility bridge. You take actions only by calling the tools provided below. You have no sandbox or storage of your own; never fabricate file paths, sandbox links, or download URLs." @classmethod def from_env(cls) -> "AppConfig": config = cls( host=os.environ.get("HOST", "0.0.0.0").strip(), port=_parse_int(os.environ.get("PORT"), 8080), upstream_base_url=os.environ.get("UPSTREAM_BASE_URL", "http://127.0.0.1:3000").rstrip("/"), upstream_timeout_seconds=_parse_int(os.environ.get("UPSTREAM_TIMEOUT_SECONDS"), 240), upstream_auth_header=(os.environ.get("UPSTREAM_AUTH_HEADER") or "").strip() or None, upstream_extra_body=_parse_json_object(os.environ.get("UPSTREAM_EXTRA_BODY_JSON"), {}), model_map={ str(k).strip(): str(v).strip() for k, v in _parse_json_object(os.environ.get("MODEL_MAP_JSON"), {}).items() }, allow_unmapped_model_passthrough=_parse_bool( os.environ.get("ALLOW_UNMAPPED_MODEL_PASSTHROUGH"), True, ), native_tool_models=set( _parse_json_string_list( os.environ.get("NATIVE_TOOL_MODELS_JSON"), [], ) ), public_model_ids=_parse_json_string_list( os.environ.get("PUBLIC_MODEL_IDS_JSON"), [], ), tool_prompt_preamble=( os.environ.get("TOOL_PROMPT_PREAMBLE") or "You are a tool-using assistant operating on the user's real local machine through a compatibility bridge. You take actions only by calling the tools provided below. You have no sandbox or storage of your own; never fabricate file paths, sandbox links, or download URLs." ).strip(), ) if not config.upstream_base_url: raise ConfigError("UPSTREAM_BASE_URL must not be empty") if config.port < 0 or config.port > 65535: raise ConfigError("PORT must be between 0 and 65535") return config def resolve_upstream_model(self, requested_model: str) -> str: requested_model = _normalize_requested_model_name(requested_model) if requested_model in self.model_map: return self.model_map[requested_model] if self.allow_unmapped_model_passthrough: return requested_model raise ConfigError(f"no upstream mapping found for model={requested_model!r}") def should_passthrough_native_tools(self, requested_model: str) -> bool: requested_model = _normalize_requested_model_name(requested_model) upstream_model = self.resolve_upstream_model(requested_model) return ( requested_model in self.native_tool_models or upstream_model in self.native_tool_models ) def list_public_model_ids(self) -> list[str]: if self.public_model_ids: return list(self.public_model_ids) seen: set[str] = set() ordered: list[str] = [] for model_id in self.model_map: if model_id not in seen: ordered.append(model_id) seen.add(model_id) for model_id in self.native_tool_models: if model_id not in seen: ordered.append(model_id) seen.add(model_id) return ordered @dataclass(frozen=True) class ParsedToolCall: name: str arguments: dict[str, Any] @dataclass(frozen=True) class ParsedAssistantOutput: assistant_response: str | None tool_calls: list[ParsedToolCall] def _serialize_content(content: Any) -> str: if content is None: return "" if isinstance(content, str): return content if isinstance(content, list): rendered: list[str] = [] for item in content: if isinstance(item, str): rendered.append(item) continue if not isinstance(item, dict): rendered.append(json.dumps(item, ensure_ascii=False)) continue item_type = item.get("type") if item_type in {"text", "input_text", "output_text"} and isinstance(item.get("text"), str): rendered.append(item["text"]) continue if item_type == "image_url": image_url = item.get("image_url") if isinstance(image_url, dict) and isinstance(image_url.get("url"), str): rendered.append(f"[Image URL: {image_url['url']}]") continue rendered.append(json.dumps(item, ensure_ascii=False)) return "\n".join(part for part in rendered if part).strip() return json.dumps(content, ensure_ascii=False) def _normalize_openai_tool_calls(tool_calls: Any) -> list[dict[str, Any]]: if not isinstance(tool_calls, list): return [] normalized: list[dict[str, Any]] = [] for item in tool_calls: if not isinstance(item, dict): continue function_payload = item.get("function") name = None arguments = None if isinstance(function_payload, dict): name = function_payload.get("name") arguments = function_payload.get("arguments") if not isinstance(name, str) or not name.strip(): name = item.get("name") arguments = item.get("arguments") if isinstance(name, str) and name.strip(): normalized.append( { "id": item.get("id"), "name": name.strip(), "arguments": arguments, } ) return normalized def _parse_arguments(raw_arguments: Any) -> dict[str, Any]: if isinstance(raw_arguments, dict): return raw_arguments if isinstance(raw_arguments, str): stripped = raw_arguments.strip() if not stripped: return {} try: parsed = json.loads(stripped) except json.JSONDecodeError: return {"input": stripped} if isinstance(parsed, dict): return parsed return {"input": parsed} if raw_arguments is None: return {} return {"input": raw_arguments} def _normalize_messages(messages: Any) -> list[dict[str, str]]: if not isinstance(messages, list): raise ValueError("messages must be an array") tool_name_by_id: dict[str, str] = {} normalized: list[dict[str, str]] = [] for item in messages: if not isinstance(item, dict): continue role = str(item.get("role") or "").strip().lower() if role not in {"system", "user", "assistant", "tool"}: continue if role == "tool": tool_call_id = str(item.get("tool_call_id") or "").strip() tool_name = str(item.get("name") or "").strip() or tool_name_by_id.get(tool_call_id, "unknown_tool") normalized.append( { "role": "user", "content": ( f"Tool `{tool_name}` returned for tool_call_id `{tool_call_id or 'unknown'}`:\n" f"{_serialize_content(item.get('content'))}" ).strip(), } ) continue content = _serialize_content(item.get("content")) tool_calls = _normalize_openai_tool_calls(item.get("tool_calls")) if role == "assistant" and tool_calls: summary = [] for tool_call in tool_calls: tool_call_id = tool_call.get("id") if isinstance(tool_call_id, str) and tool_call_id: tool_name_by_id[tool_call_id] = tool_call["name"] summary.append( { "name": tool_call["name"], "arguments": _parse_arguments(tool_call.get("arguments")), } ) summary_text = "Assistant requested tool calls:\n" + json.dumps( summary, ensure_ascii=False, indent=2, ) content = "\n\n".join(part for part in [content, summary_text] if part).strip() if content: normalized.append({"role": role, "content": content}) return normalized def _anthropic_blocks(content: Any) -> list[dict[str, Any]]: if isinstance(content, str): stripped = content.strip() return [{"type": "text", "text": stripped}] if stripped else [] if isinstance(content, list): return [item for item in content if isinstance(item, dict)] return [] def _serialize_anthropic_block(block: dict[str, Any], include_thinking: bool = False) -> str: block_type = str(block.get("type") or "").strip() if block_type == "text" and isinstance(block.get("text"), str): return block["text"] if block_type == "thinking" and include_thinking and isinstance(block.get("thinking"), str): return block["thinking"] if block_type == "redacted_thinking" and include_thinking and isinstance(block.get("data"), str): return f"[Redacted thinking: {block['data']}]" if block_type == "image": source = block.get("source") if isinstance(source, dict): media_type = source.get("media_type") or "image" source_type = source.get("type") or "source" return f"[Image: {media_type} via {source_type}]" return "[Image]" if block_type == "tool_use": name = block.get("name") or "unknown_tool" tool_input = block.get("input") return f"[Tool use: {name} {json.dumps(tool_input, ensure_ascii=False)}]" if block_type == "tool_result": tool_use_id = block.get("tool_use_id") or "unknown" result_content = _serialize_content(block.get("content")) if bool(block.get("is_error")): return f"[Tool result error for {tool_use_id}] {result_content}".strip() return f"[Tool result for {tool_use_id}] {result_content}".strip() return json.dumps(block, ensure_ascii=False) def _serialize_anthropic_blocks_to_text(content: Any, include_thinking: bool = False) -> str: parts = [ _serialize_anthropic_block(block, include_thinking=include_thinking) for block in _anthropic_blocks(content) ] return "\n".join(part for part in parts if part).strip() def _normalize_anthropic_system_messages(system: Any) -> list[dict[str, Any]]: if system is None: return [] if isinstance(system, str): content = system.strip() return [{"role": "system", "content": content}] if content else [] if isinstance(system, list): content = _serialize_anthropic_blocks_to_text(system) return [{"role": "system", "content": content}] if content else [] if isinstance(system, dict): content = _serialize_anthropic_blocks_to_text([system]) return [{"role": "system", "content": content}] if content else [] return [] def _normalize_anthropic_messages(messages: Any) -> list[dict[str, Any]]: if not isinstance(messages, list): raise ValueError("messages must be an array") normalized: list[dict[str, Any]] = [] tool_name_by_id: dict[str, str] = {} def flush_text(role: str, text_parts: list[str], tool_calls: list[dict[str, Any]] | None = None) -> None: content = "\n".join(part for part in text_parts if part).strip() if role == "assistant": if not content and not tool_calls: return message: dict[str, Any] = {"role": "assistant", "content": content} if tool_calls: message["tool_calls"] = tool_calls normalized.append(message) return if content: normalized.append({"role": role, "content": content}) for item in messages: if not isinstance(item, dict): continue role = str(item.get("role") or "").strip().lower() if role not in {"user", "assistant"}: continue blocks = _anthropic_blocks(item.get("content")) if not blocks and isinstance(item.get("content"), str): blocks = [{"type": "text", "text": str(item["content"])}] if role == "user": text_parts: list[str] = [] for block in blocks: block_type = str(block.get("type") or "").strip() if block_type == "tool_result": flush_text("user", text_parts) text_parts = [] tool_use_id = str(block.get("tool_use_id") or "").strip() tool_name = tool_name_by_id.get(tool_use_id, str(block.get("name") or "").strip() or "unknown_tool") tool_content = _serialize_content(block.get("content")) if bool(block.get("is_error")): tool_content = f"Tool execution failed:\n{tool_content}".strip() tool_message: dict[str, Any] = { "role": "tool", "tool_call_id": tool_use_id or f"toolu_{uuid.uuid4().hex[:24]}", "name": tool_name, "content": tool_content, } normalized.append(tool_message) continue rendered = _serialize_anthropic_block(block) if rendered: text_parts.append(rendered) flush_text("user", text_parts) continue assistant_text_parts: list[str] = [] assistant_tool_calls: list[dict[str, Any]] = [] for block in blocks: block_type = str(block.get("type") or "").strip() if block_type in {"thinking", "redacted_thinking"}: continue if block_type == "tool_use": tool_call_id = str(block.get("id") or "").strip() or f"toolu_{uuid.uuid4().hex[:24]}" tool_name = str(block.get("name") or "").strip() if tool_name: tool_name_by_id[tool_call_id] = tool_name assistant_tool_calls.append( { "id": tool_call_id, "type": "function", "function": { "name": tool_name or "unknown_tool", "arguments": json.dumps(block.get("input") or {}, ensure_ascii=False, separators=(",", ":")), }, } ) continue rendered = _serialize_anthropic_block(block) if rendered: assistant_text_parts.append(rendered) flush_text("assistant", assistant_text_parts, assistant_tool_calls) return normalized def _normalize_anthropic_tools(tools: Any) -> list[dict[str, Any]]: if not isinstance(tools, list): return [] normalized: list[dict[str, Any]] = [] for tool in tools: if not isinstance(tool, dict): continue name = str(tool.get("name") or "").strip() if not name: continue normalized.append( { "type": "function", "function": { "name": name, "description": tool.get("description") or "", "parameters": tool.get("input_schema") or {"type": "object", "properties": {}}, }, } ) return normalized def _normalize_anthropic_tool_choice(tool_choice: Any) -> Any: if tool_choice is None: return "auto" if isinstance(tool_choice, str): normalized = tool_choice.strip().lower() if normalized in {"auto", "none"}: return normalized if normalized in {"any", "required"}: return "required" return "auto" if isinstance(tool_choice, dict): choice_type = str(tool_choice.get("type") or "").strip().lower() if choice_type in {"auto", "none"}: return choice_type if choice_type in {"any", "required"}: return "required" if choice_type == "tool": name = str(tool_choice.get("name") or "").strip() if name: return {"type": "function", "function": {"name": name}} return "auto" def _build_openai_payload_from_anthropic(payload: dict[str, Any]) -> dict[str, Any]: requested_model = str(payload.get("model") or "").strip() if not requested_model: raise ValueError("model is required") openai_payload: dict[str, Any] = {} for key, value in payload.items(): if key in {"messages", "system", "tools", "tool_choice", "stream", "thinking"}: continue if key == "stop_sequences": openai_payload["stop"] = value continue openai_payload[key] = value openai_payload["model"] = requested_model openai_payload["messages"] = _normalize_anthropic_system_messages(payload.get("system")) + _normalize_anthropic_messages(payload.get("messages")) openai_payload["stream"] = bool(payload.get("stream")) tools = _normalize_anthropic_tools(payload.get("tools")) if tools: openai_payload["tools"] = tools openai_payload["tool_choice"] = _normalize_anthropic_tool_choice(payload.get("tool_choice")) openai_payload["parallel_tool_calls"] = True return openai_payload def _anthropic_stop_reason_from_openai(finish_reason: str | None, content_blocks: list[dict[str, Any]]) -> str | None: if any(block.get("type") == "tool_use" for block in content_blocks): return "tool_use" if finish_reason in {None, ""}: return None if finish_reason == "stop": return "end_turn" if finish_reason == "tool_calls": return "tool_use" if finish_reason == "length": return "max_tokens" if finish_reason == "content_filter": return "end_turn" return finish_reason def _build_thinking_signature(text: str) -> str: digest = hashlib.sha256(text.encode("utf-8")).digest() return base64.b64encode(digest).decode("ascii") def _anthropic_usage_from_openai(usage: Any) -> dict[str, int]: if not isinstance(usage, dict): return {"input_tokens": 0, "output_tokens": 0} input_tokens = int(usage.get("prompt_tokens") or 0) output_tokens = int(usage.get("completion_tokens") or 0) result = {"input_tokens": input_tokens, "output_tokens": output_tokens} if isinstance(usage.get("cache_creation_input_tokens"), int): result["cache_creation_input_tokens"] = usage["cache_creation_input_tokens"] if isinstance(usage.get("cache_read_input_tokens"), int): result["cache_read_input_tokens"] = usage["cache_read_input_tokens"] return result def _anthropic_message_from_openai_response( upstream_response: dict[str, Any], requested_model: str, had_tools: bool = False, ) -> dict[str, Any]: choices = upstream_response.get("choices") if not isinstance(choices, list) or not choices: raise ValueError("upstream returned no choices") choice = choices[0] message = choice.get("message") if not isinstance(message, dict): raise ValueError("upstream returned invalid message") content_blocks: list[dict[str, Any]] = [] reasoning_content = message.get("reasoning_content") if isinstance(reasoning_content, str) and reasoning_content.strip(): content_blocks.append( { "type": "thinking", "thinking": _strip_upstream_artifacts(reasoning_content), "signature": _build_thinking_signature(reasoning_content), } ) native_tool_calls = _normalize_openai_tool_calls(message.get("tool_calls")) raw_content = _serialize_content(message.get("content")) parsed_output = _parse_virtual_tool_output(raw_content) if had_tools and not native_tool_calls else None tool_calls_to_emit = native_tool_calls if parsed_output and parsed_output.tool_calls: tool_calls_to_emit = [ { "id": f"call_{uuid.uuid4().hex[:24]}", "name": tool_call.name, "arguments": tool_call.arguments, } for tool_call in parsed_output.tool_calls ] for tool_call in tool_calls_to_emit: content_blocks.append( { "type": "tool_use", "id": str(tool_call.get("id") or f"toolu_{uuid.uuid4().hex[:24]}"), "name": tool_call["name"], "input": ( tool_call["arguments"] if isinstance(tool_call.get("arguments"), dict) else _parse_arguments(tool_call.get("arguments")) ), } ) if parsed_output is not None: rendered_content = parsed_output.assistant_response or "" else: rendered_content = raw_content rendered_content = _strip_upstream_artifacts(rendered_content) if rendered_content: content_blocks.append({"type": "text", "text": rendered_content}) finish_reason = choice.get("finish_reason") stop_reason = _anthropic_stop_reason_from_openai(str(finish_reason) if finish_reason is not None else None, content_blocks) response_id = str(upstream_response.get("id") or f"msg_{uuid.uuid4().hex}") if not response_id.startswith("msg_"): response_id = f"msg_{response_id.replace('-', '')}" return { "id": response_id, "type": "message", "role": "assistant", "content": content_blocks, "model": requested_model, "stop_reason": stop_reason, "stop_sequence": None, "usage": _anthropic_usage_from_openai(upstream_response.get("usage")), } def _has_tool_history(messages: Any) -> bool: if not isinstance(messages, list): return False for item in messages: if not isinstance(item, dict): continue role = str(item.get("role") or "").strip().lower() if role == "tool": return True if role == "assistant" and _normalize_openai_tool_calls(item.get("tool_calls")): return True return False def _build_tool_instruction( tools: list[dict[str, Any]], tool_choice: Any, parallel_tool_calls: bool, preamble: str, ) -> str: tool_defs: list[dict[str, Any]] = [] for tool in tools: function_payload = tool.get("function") if not isinstance(function_payload, dict): continue tool_defs.append( { "name": function_payload.get("name"), "description": function_payload.get("description") or "", "parameters": function_payload.get("parameters") or {"type": "object", "properties": {}}, } ) instructions = [ preamble, "Available tools:", json.dumps(tool_defs, ensure_ascii=False, separators=(",", ":")), 'When you need tools, output only raw JSON: {"assistant_response": null, "tool_calls": [{"name": "tool_name", "arguments": {...}}]}', 'When you can answer directly, output only raw JSON: {"assistant_response": "your answer", "tool_calls": []}', "Rules:", "- Output raw JSON only, no markdown fences.", "- tool_calls must be an array.", "- arguments must be a JSON object.", "- Never invent tool names.", "- You have no sandbox, code interpreter, Python runtime, browser, or file storage of your own.", "- Never output sandbox paths, `/mnt/data` references, or download links, and never claim you saved, ran, or executed anything yourself.", "- The only way to read/write files, run code, or take any action is to call one of the tools above; all effects happen on the user's real machine.", "- If the available tools cannot accomplish the request, say so in assistant_response instead of pretending.", ] if tool_choice == "required": instructions.append("- You must call at least one tool.") elif isinstance(tool_choice, dict): function_payload = tool_choice.get("function") if isinstance(function_payload, dict) and isinstance(function_payload.get("name"), str): instructions.append(f"- You must call the tool `{function_payload['name']}`.") elif tool_choice == "none": instructions.append("- You must not call any tool.") if not parallel_tool_calls: instructions.append("- Return at most one tool call.") return "\n".join(instructions) def _extract_code_block(text: str) -> str | None: match = re.search(r"```(?:json)?\s*(\{.*\})\s*```", text, re.DOTALL) return match.group(1).strip() if match else None def _extract_first_json_object(text: str) -> str | None: start = text.find("{") if start < 0: return None depth = 0 in_string = False escape = False for index in range(start, len(text)): char = text[index] if in_string: if escape: escape = False elif char == "\\": escape = True elif char == '"': in_string = False continue if char == '"': in_string = True continue if char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: return text[start : index + 1] return None def _parse_virtual_tool_output(text: str) -> ParsedAssistantOutput | None: candidates: list[str] = [] stripped = text.strip() if stripped: candidates.append(stripped) code_block = _extract_code_block(stripped) if code_block and code_block not in candidates: candidates.append(code_block) first_object = _extract_first_json_object(stripped) if first_object and first_object not in candidates: candidates.append(first_object) for candidate in candidates: try: parsed = json.loads(candidate) except json.JSONDecodeError: continue if not isinstance(parsed, dict): continue raw_tool_calls = parsed.get("tool_calls") if raw_tool_calls is None and "name" in parsed: raw_tool_calls = [parsed] tool_calls: list[ParsedToolCall] = [] if isinstance(raw_tool_calls, list): for raw_tool_call in raw_tool_calls: if not isinstance(raw_tool_call, dict): continue name = raw_tool_call.get("name") if not isinstance(name, str) or not name.strip(): function_payload = raw_tool_call.get("function") if isinstance(function_payload, dict): name = function_payload.get("name") if not isinstance(name, str) or not name.strip(): continue arguments = _parse_arguments( raw_tool_call.get("arguments") if "arguments" in raw_tool_call else ( raw_tool_call.get("function", {}).get("arguments") if isinstance(raw_tool_call.get("function"), dict) else raw_tool_call.get("args") ) ) tool_calls.append(ParsedToolCall(name=name.strip(), arguments=arguments)) assistant_response = parsed.get("assistant_response") if assistant_response is None: assistant_response = parsed.get("content") or parsed.get("response") or parsed.get("answer") if assistant_response is not None and not isinstance(assistant_response, str): assistant_response = json.dumps(assistant_response, ensure_ascii=False) if tool_calls or isinstance(assistant_response, str): return ParsedAssistantOutput( assistant_response=assistant_response.strip() if isinstance(assistant_response, str) and assistant_response.strip() else None, tool_calls=tool_calls, ) return None # ChatGPT web leaks inline annotation markers delimited by private-use chars # (U+E200 start, U+E201 end, U+E202 separator). Some upstream paths surface a # NUL/SOH-delimited variant instead. Examples: # entity["city","Paris","Paris, France"] # citeturn0search0turn0search1 # \x00entity\x00["city","Paris"]\x01 _PUA_START = chr(0xE200) _PUA_END = chr(0xE201) _PUA_SEP = chr(0xE202) _PUA_MARKER_RE = re.compile(_PUA_START + "(.*?)" + _PUA_END, re.DOTALL) _NUL_MARKER_RE = re.compile("\x00([0-9A-Za-z_]+)\x00(.*?)\x01", re.DOTALL) _RESIDUAL_CONTROL_RE = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f" + chr(0xE200) + "-" + chr(0xE20F) + "]") def _marker_display_text(tag: str, payload: str) -> str: """Map a ChatGPT inline annotation marker to the text that should remain. Entity markers carry a JSON array like ["type", "display", "canonical"]; keep the display text (index 1). Citation and other markers are dropped. """ tag = (tag or "").strip().lower() if tag == "entity": segment = payload.split(_PUA_SEP, 1)[0] try: data = json.loads(segment) except json.JSONDecodeError: return "" if isinstance(data, list): for candidate in (data[1:] or data): if isinstance(candidate, str) and candidate.strip(): return candidate return "" def _strip_upstream_artifacts(text: Any) -> Any: """Remove ChatGPT web annotation markers leaked into assistant text.""" if not isinstance(text, str) or not text: return text if not _RESIDUAL_CONTROL_RE.search(text): return text def _pua_repl(match: "re.Match[str]") -> str: tag, _, payload = match.group(1).partition(_PUA_SEP) return _marker_display_text(tag, payload) cleaned = _PUA_MARKER_RE.sub(_pua_repl, text) cleaned = _NUL_MARKER_RE.sub(lambda m: _marker_display_text(m.group(1), m.group(2)), cleaned) cleaned = _RESIDUAL_CONTROL_RE.sub("", cleaned) return cleaned def _scrub_chat_completion(parsed: dict[str, Any]) -> None: """In-place scrub of assistant text in an OpenAI chat completion dict.""" choices = parsed.get("choices") if not isinstance(choices, list): return for choice in choices: if not isinstance(choice, dict): continue message = choice.get("message") if isinstance(message, dict) and isinstance(message.get("content"), str): message["content"] = _strip_upstream_artifacts(message["content"]) def _normalize_responses_input(input_value: Any, instructions: Any) -> list[dict[str, Any]]: """Convert an OpenAI Responses `input` (+ `instructions`) to chat messages.""" messages: list[dict[str, Any]] = [] if isinstance(instructions, str) and instructions.strip(): messages.append({"role": "system", "content": instructions.strip()}) if input_value is None: return messages if isinstance(input_value, str): if input_value.strip(): messages.append({"role": "user", "content": input_value}) return messages if not isinstance(input_value, list): return messages tool_name_by_call_id: dict[str, str] = {} for item in input_value: if isinstance(item, str): if item.strip(): messages.append({"role": "user", "content": item}) continue if not isinstance(item, dict): continue item_type = str(item.get("type") or "message").strip() if item_type == "message": role = str(item.get("role") or "user").strip().lower() if role == "developer": role = "system" if role not in {"system", "user", "assistant"}: role = "user" content = _serialize_content(item.get("content")) if content: messages.append({"role": role, "content": content}) elif item_type == "function_call": call_id = str(item.get("call_id") or item.get("id") or "").strip() or f"call_{uuid.uuid4().hex[:24]}" name = str(item.get("name") or "").strip() or "unknown_tool" tool_name_by_call_id[call_id] = name raw_args = item.get("arguments") arguments = raw_args if isinstance(raw_args, str) else json.dumps(raw_args or {}, ensure_ascii=False, separators=(",", ":")) messages.append( { "role": "assistant", "content": None, "tool_calls": [ {"id": call_id, "type": "function", "function": {"name": name, "arguments": arguments}} ], } ) elif item_type == "function_call_output": call_id = str(item.get("call_id") or "").strip() output = item.get("output") content = output if isinstance(output, str) else _serialize_content(output) messages.append( { "role": "tool", "tool_call_id": call_id or f"call_{uuid.uuid4().hex[:24]}", "name": tool_name_by_call_id.get(call_id, "unknown_tool"), "content": content, } ) # reasoning / item_reference / other item types are dropped return messages def _normalize_responses_tools(tools: Any) -> list[dict[str, Any]]: """Convert Responses (flat) function tools to chat (nested) tools.""" if not isinstance(tools, list): return [] normalized: list[dict[str, Any]] = [] for tool in tools: if not isinstance(tool, dict): continue tool_type = str(tool.get("type") or "function").strip() name = str(tool.get("name") or "").strip() description = tool.get("description") or "" parameters = tool.get("parameters") function_payload = tool.get("function") if not name and isinstance(function_payload, dict): name = str(function_payload.get("name") or "").strip() description = function_payload.get("description") or description parameters = function_payload.get("parameters") or parameters if not name: # built-in tools (web_search, file_search, ...) are not bridgeable continue if tool_type not in {"function", "custom"}: continue normalized.append( { "type": "function", "function": { "name": name, "description": description, "parameters": parameters or {"type": "object", "properties": {}}, }, } ) return normalized def _normalize_responses_tool_choice(tool_choice: Any) -> Any: if tool_choice is None: return "auto" if isinstance(tool_choice, str): normalized = tool_choice.strip().lower() if normalized in {"auto", "none", "required"}: return normalized if normalized == "any": return "required" return "auto" if isinstance(tool_choice, dict): choice_type = str(tool_choice.get("type") or "").strip().lower() if choice_type in {"auto", "none", "required"}: return choice_type if choice_type == "function": name = str(tool_choice.get("name") or "").strip() if not name: function_payload = tool_choice.get("function") if isinstance(function_payload, dict): name = str(function_payload.get("name") or "").strip() if name: return {"type": "function", "function": {"name": name}} return "auto" _RESPONSES_SAMPLING_KEYS = ( "temperature", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "n", "logit_bias", "user", "logprobs", "top_logprobs", ) def _build_openai_payload_from_responses(payload: dict[str, Any]) -> dict[str, Any]: requested_model = str(payload.get("model") or "").strip() if not requested_model: raise ValueError("model is required") openai_payload: dict[str, Any] = {"model": requested_model} openai_payload["messages"] = _normalize_responses_input(payload.get("input"), payload.get("instructions")) openai_payload["stream"] = bool(payload.get("stream")) for key in _RESPONSES_SAMPLING_KEYS: if key in payload: openai_payload[key] = payload[key] max_output_tokens = payload.get("max_output_tokens") if isinstance(max_output_tokens, int): openai_payload["max_tokens"] = max_output_tokens tools = _normalize_responses_tools(payload.get("tools")) if tools: openai_payload["tools"] = tools openai_payload["tool_choice"] = _normalize_responses_tool_choice(payload.get("tool_choice")) openai_payload["parallel_tool_calls"] = bool(payload.get("parallel_tool_calls", True)) return openai_payload def _responses_usage_from_openai(usage: Any) -> dict[str, int]: if not isinstance(usage, dict): return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0} input_tokens = int(usage.get("prompt_tokens") or 0) output_tokens = int(usage.get("completion_tokens") or 0) total = usage.get("total_tokens") return { "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": int(total) if isinstance(total, int) else input_tokens + output_tokens, } def _responses_object_from_openai_response( upstream_response: dict[str, Any], requested_model: str, had_tools: bool = False, ) -> dict[str, Any]: choices = upstream_response.get("choices") if not isinstance(choices, list) or not choices: raise ValueError("upstream returned no choices") choice = choices[0] message = choice.get("message") if not isinstance(message, dict): raise ValueError("upstream returned invalid message") native_tool_calls = _normalize_openai_tool_calls(message.get("tool_calls")) raw_content = _serialize_content(message.get("content")) parsed_output = _parse_virtual_tool_output(raw_content) if had_tools and not native_tool_calls else None tool_calls_to_emit = native_tool_calls if parsed_output and parsed_output.tool_calls: tool_calls_to_emit = [ {"id": None, "name": tool_call.name, "arguments": tool_call.arguments} for tool_call in parsed_output.tool_calls ] if parsed_output is not None: text_content = parsed_output.assistant_response or "" else: text_content = raw_content text_content = _strip_upstream_artifacts(text_content) output_items: list[dict[str, Any]] = [] if text_content: output_items.append( { "type": "message", "id": f"msg_{uuid.uuid4().hex}", "status": "completed", "role": "assistant", "content": [{"type": "output_text", "text": text_content, "annotations": []}], } ) for tool_call in tool_calls_to_emit: raw_args = tool_call.get("arguments") if isinstance(raw_args, dict): arguments = json.dumps(raw_args, ensure_ascii=False, separators=(",", ":")) elif isinstance(raw_args, str): arguments = raw_args else: arguments = json.dumps(_parse_arguments(raw_args), ensure_ascii=False, separators=(",", ":")) output_items.append( { "type": "function_call", "id": f"fc_{uuid.uuid4().hex}", "call_id": str(tool_call.get("id") or f"call_{uuid.uuid4().hex[:24]}"), "name": tool_call["name"], "arguments": arguments, "status": "completed", } ) return { "id": f"resp_{uuid.uuid4().hex}", "object": "response", "created_at": int(upstream_response.get("created") or time.time()), "status": "completed", "error": None, "incomplete_details": None, "model": requested_model, "output": output_items, "parallel_tool_calls": True, "tool_choice": "auto", "tools": [], "usage": _responses_usage_from_openai(upstream_response.get("usage")), "metadata": {}, } class BridgeHTTPServer(ThreadingHTTPServer): daemon_threads = True def __init__( self, server_address: tuple[str, int], request_handler_class: type[BaseHTTPRequestHandler], config: AppConfig, ) -> None: super().__init__(server_address, request_handler_class) self.bridge_config = config class ToolBridgeHandler(BaseHTTPRequestHandler): server_version = "newapi-tool-bridge/0.1.0" protocol_version = "HTTP/1.1" @property def config(self) -> AppConfig: return self.server.bridge_config # type: ignore[attr-defined] def log_message(self, fmt: str, *args: Any) -> None: print("%s - - [%s] %s" % (self.address_string(), self.log_date_time_string(), fmt % args)) def _request_path(self) -> str: return parse.urlsplit(self.path).path def _read_body_bytes(self) -> bytes: content_length = int(self.headers.get("Content-Length", "0")) return self.rfile.read(content_length) if content_length > 0 else b"" def _read_json_body(self) -> dict[str, Any]: raw = self._read_body_bytes() or b"{}" try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValueError("invalid JSON request body") from exc if not isinstance(payload, dict): raise ValueError("request body must be a JSON object") return payload def _send_json(self, status: int, payload: dict[str, Any]) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type, X-API-Key, anthropic-version, anthropic-beta") self.send_header("Access-Control-Allow-Methods", "GET, HEAD, POST, OPTIONS") self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _models_payload(self) -> dict[str, Any]: created_at = int(time.time()) return { "object": "list", "data": [ { "id": model_id, "object": "model", "created": created_at, "owned_by": "newapi-tool-bridge", } for model_id in self.config.list_public_model_ids() ], } def _copy_upstream_response(self, status: int, body: bytes, headers: dict[str, str]) -> None: self.send_response(status) self.send_header("Content-Type", headers.get("Content-Type", "application/octet-stream")) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", headers.get("Cache-Control", "no-store")) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _start_sse_response(self) -> None: self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Connection", "close") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type, X-API-Key, anthropic-version, anthropic-beta") self.end_headers() def _write_sse_event(self, payload: str) -> None: if self.command == "HEAD": return self.wfile.write(f"data: {payload}\n\n".encode("utf-8")) self.wfile.flush() def _write_named_sse_event(self, event_name: str, payload: dict[str, Any]) -> None: if self.command == "HEAD": return serialized = json.dumps(payload, ensure_ascii=False) self.wfile.write(f"event: {event_name}\n".encode("utf-8")) self.wfile.write(f"data: {serialized}\n\n".encode("utf-8")) self.wfile.flush() def _build_upstream_headers(self, body: bytes | None = None) -> dict[str, str]: headers = {"Content-Type": self.headers.get("Content-Type", "application/json")} if body is not None: headers["Content-Length"] = str(len(body)) if self.config.upstream_auth_header: headers["Authorization"] = self.config.upstream_auth_header elif self.headers.get("Authorization"): headers["Authorization"] = self.headers["Authorization"] elif self.headers.get("X-API-Key"): api_key = self.headers["X-API-Key"].strip() if api_key: headers["Authorization"] = api_key if api_key.lower().startswith("bearer ") else f"Bearer {api_key}" if self.headers.get("Accept"): headers["Accept"] = self.headers["Accept"] return headers def _proxy_upstream_raw(self, method: str, path: str, body: bytes | None = None) -> tuple[int, bytes, dict[str, str]]: req = request.Request( f"{self.config.upstream_base_url}{path}", data=body, headers=self._build_upstream_headers(body), method=method, ) try: with request.urlopen(req, timeout=self.config.upstream_timeout_seconds) as upstream_resp: return upstream_resp.status, upstream_resp.read(), dict(upstream_resp.headers.items()) except error.HTTPError as exc: return exc.code, exc.read(), dict(exc.headers.items()) def _call_upstream_chat(self, payload: dict[str, Any]) -> dict[str, Any]: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") print( f"[upstream-size] model={payload.get('model')} bytes={len(body)} " f"messages={len(payload.get('messages') or [])}" ) req = request.Request( f"{self.config.upstream_base_url}/v1/chat/completions", data=body, headers=self._build_upstream_headers(body), method="POST", ) try: with request.urlopen(req, timeout=self.config.upstream_timeout_seconds) as upstream_resp: response_body = upstream_resp.read() except error.HTTPError as exc: raise UpstreamHTTPError(exc.code, exc.read(), dict(exc.headers.items())) from exc parsed = json.loads(response_body.decode("utf-8")) if not isinstance(parsed, dict): raise ValueError("upstream returned invalid JSON") return parsed def _call_upstream_chat_for_requested_model(self, payload: dict[str, Any], requested_model: str) -> dict[str, Any]: upstream_model = self.config.resolve_upstream_model(requested_model) upstream_payload = dict(payload) upstream_payload["model"] = upstream_model parsed = self._call_upstream_chat(upstream_payload) if requested_model != upstream_model: parsed["model"] = requested_model return parsed def _proxy_native_tool_request(self, payload: dict[str, Any], requested_model: str) -> None: upstream_model = self.config.resolve_upstream_model(requested_model) upstream_payload = dict(payload) upstream_payload["model"] = upstream_model body = json.dumps(upstream_payload, ensure_ascii=False).encode("utf-8") status, response_body, headers = self._proxy_upstream_raw("POST", self.path, body) if status != HTTPStatus.OK or bool(payload.get("stream")): self._copy_upstream_response(status, response_body, headers) return try: parsed = json.loads(response_body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): self._copy_upstream_response(status, response_body, headers) return if isinstance(parsed, dict): if requested_model != upstream_model: parsed["model"] = requested_model _scrub_chat_completion(parsed) self._send_json(status, parsed) return self._copy_upstream_response(status, response_body, headers) def _build_transformed_payload(self, payload: dict[str, Any]) -> tuple[dict[str, Any], str, bool]: requested_model = str(payload.get("model") or "").strip() if not requested_model: raise ValueError("model is required") upstream_model = self.config.resolve_upstream_model(requested_model) normalized_messages = _normalize_messages(payload.get("messages")) tools = payload.get("tools") if isinstance(payload.get("tools"), list) else [] tool_choice = payload.get("tool_choice", "auto") parallel_tool_calls = bool(payload.get("parallel_tool_calls", True)) stream = bool(payload.get("stream")) upstream_payload = dict(payload) upstream_payload["model"] = upstream_model upstream_payload["stream"] = False upstream_payload["messages"] = normalized_messages upstream_payload.pop("tools", None) upstream_payload.pop("tool_choice", None) upstream_payload.pop("parallel_tool_calls", None) upstream_payload.pop("stream_options", None) upstream_payload.pop("functions", None) upstream_payload.pop("function_call", None) if tools: upstream_payload["messages"] = normalized_messages + [ { "role": "system", "content": _build_tool_instruction( tools, tool_choice, parallel_tool_calls, self.config.tool_prompt_preamble, ), } ] upstream_payload.update(self.config.upstream_extra_body) return upstream_payload, requested_model, stream def _response_payload(self, upstream_response: dict[str, Any], requested_model: str, had_tools: bool) -> dict[str, Any]: choices = upstream_response.get("choices") if not isinstance(choices, list) or not choices: raise ValueError("upstream returned no choices") message = choices[0].get("message") if not isinstance(message, dict): raise ValueError("upstream returned invalid message") raw_content = _serialize_content(message.get("content")) parsed_output = _parse_virtual_tool_output(raw_content) if had_tools else None response_message: dict[str, Any] finish_reason = "stop" if parsed_output and parsed_output.tool_calls: finish_reason = "tool_calls" response_message = { "role": "assistant", "content": _strip_upstream_artifacts(parsed_output.assistant_response), "tool_calls": [ { "id": f"call_{uuid.uuid4().hex[:24]}", "type": "function", "function": { "name": tool_call.name, "arguments": json.dumps(tool_call.arguments, ensure_ascii=False, separators=(",", ":")), }, } for tool_call in parsed_output.tool_calls ], } else: response_message = { "role": "assistant", "content": _strip_upstream_artifacts(parsed_output.assistant_response if parsed_output and parsed_output.assistant_response is not None else raw_content), } payload = { "id": str(upstream_response.get("id") or f"chatcmpl-{uuid.uuid4().hex}"), "object": "chat.completion", "created": int(upstream_response.get("created") or time.time()), "model": requested_model, "choices": [{"index": 0, "message": response_message, "finish_reason": finish_reason}], } if isinstance(upstream_response.get("usage"), dict): payload["usage"] = upstream_response["usage"] return payload def _send_streamed_completion(self, payload: dict[str, Any]) -> None: self._start_sse_response() choice = payload["choices"][0] message = choice["message"] base = { "id": payload["id"], "object": "chat.completion.chunk", "created": payload["created"], "model": payload["model"], } self._write_sse_event(json.dumps({**base, "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}]}, ensure_ascii=False)) if isinstance(message.get("content"), str) and message["content"]: self._write_sse_event(json.dumps({**base, "choices": [{"index": 0, "delta": {"content": message["content"]}, "finish_reason": None}]}, ensure_ascii=False)) if isinstance(message.get("tool_calls"), list) and message["tool_calls"]: self._write_sse_event( json.dumps( { **base, "choices": [ { "index": 0, "delta": { "tool_calls": [ { "index": index, "id": tool_call["id"], "type": "function", "function": tool_call["function"], } for index, tool_call in enumerate(message["tool_calls"]) ] }, "finish_reason": None, } ], }, ensure_ascii=False, ) ) self._write_sse_event(json.dumps({**base, "choices": [{"index": 0, "delta": {}, "finish_reason": choice["finish_reason"]}]}, ensure_ascii=False)) self._write_sse_event("[DONE]") def _send_streamed_anthropic_message(self, payload: dict[str, Any]) -> None: self._start_sse_response() content_blocks = payload.get("content") if isinstance(payload.get("content"), list) else [] self._write_named_sse_event( "message_start", { "type": "message_start", "message": { "id": payload["id"], "type": "message", "role": "assistant", "content": [], "model": payload["model"], "stop_reason": None, "stop_sequence": None, "usage": { "input_tokens": payload.get("usage", {}).get("input_tokens", 0), "output_tokens": 0, }, }, }, ) for index, block in enumerate(content_blocks): block_type = str(block.get("type") or "").strip() if block_type == "thinking": self._write_named_sse_event( "content_block_start", { "type": "content_block_start", "index": index, "content_block": { "type": "thinking", "thinking": "", "signature": "", }, }, ) if isinstance(block.get("thinking"), str) and block["thinking"]: self._write_named_sse_event( "content_block_delta", { "type": "content_block_delta", "index": index, "delta": { "type": "thinking_delta", "thinking": block["thinking"], }, }, ) if isinstance(block.get("signature"), str) and block["signature"]: self._write_named_sse_event( "content_block_delta", { "type": "content_block_delta", "index": index, "delta": { "type": "signature_delta", "signature": block["signature"], }, }, ) self._write_named_sse_event("content_block_stop", {"type": "content_block_stop", "index": index}) continue if block_type == "tool_use": self._write_named_sse_event( "content_block_start", { "type": "content_block_start", "index": index, "content_block": { "type": "tool_use", "id": block.get("id"), "name": block.get("name"), "input": {}, }, }, ) tool_input = json.dumps(block.get("input") or {}, ensure_ascii=False, separators=(",", ":")) if tool_input: self._write_named_sse_event( "content_block_delta", { "type": "content_block_delta", "index": index, "delta": { "type": "input_json_delta", "partial_json": tool_input, }, }, ) self._write_named_sse_event("content_block_stop", {"type": "content_block_stop", "index": index}) continue self._write_named_sse_event( "content_block_start", { "type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}, }, ) text = str(block.get("text") or "") if text: self._write_named_sse_event( "content_block_delta", { "type": "content_block_delta", "index": index, "delta": { "type": "text_delta", "text": text, }, }, ) self._write_named_sse_event("content_block_stop", {"type": "content_block_stop", "index": index}) self._write_named_sse_event( "message_delta", { "type": "message_delta", "delta": { "stop_reason": payload.get("stop_reason"), "stop_sequence": payload.get("stop_sequence"), }, "usage": { "output_tokens": payload.get("usage", {}).get("output_tokens", 0), }, }, ) self._write_named_sse_event("message_stop", {"type": "message_stop"}) def _send_streamed_responses(self, payload: dict[str, Any]) -> None: # Upstream is non-streaming; replay the finished Responses object as the # canonical OpenAI Responses SSE event sequence. self._start_sse_response() sequence = 0 def emit(event_name: str, data: dict[str, Any]) -> None: nonlocal sequence self._write_named_sse_event(event_name, {**data, "sequence_number": sequence}) sequence += 1 output_items = payload.get("output") if isinstance(payload.get("output"), list) else [] skeleton = {**payload, "status": "in_progress", "output": [], "usage": None} emit("response.created", {"type": "response.created", "response": skeleton}) emit("response.in_progress", {"type": "response.in_progress", "response": skeleton}) for output_index, item in enumerate(output_items): item_type = str(item.get("type") or "").strip() item_id = item.get("id") if item_type == "function_call": emit( "response.output_item.added", {"type": "response.output_item.added", "output_index": output_index, "item": {**item, "arguments": "", "status": "in_progress"}}, ) arguments = str(item.get("arguments") or "") if arguments: emit( "response.function_call_arguments.delta", {"type": "response.function_call_arguments.delta", "item_id": item_id, "output_index": output_index, "delta": arguments}, ) emit( "response.function_call_arguments.done", {"type": "response.function_call_arguments.done", "item_id": item_id, "output_index": output_index, "arguments": arguments}, ) emit( "response.output_item.done", {"type": "response.output_item.done", "output_index": output_index, "item": {**item, "status": "completed"}}, ) continue emit( "response.output_item.added", {"type": "response.output_item.added", "output_index": output_index, "item": {**item, "content": [], "status": "in_progress"}}, ) content_parts = item.get("content") if isinstance(item.get("content"), list) else [] for content_index, part in enumerate(content_parts): text = str(part.get("text") or "") emit( "response.content_part.added", {"type": "response.content_part.added", "item_id": item_id, "output_index": output_index, "content_index": content_index, "part": {"type": "output_text", "text": "", "annotations": []}}, ) if text: emit( "response.output_text.delta", {"type": "response.output_text.delta", "item_id": item_id, "output_index": output_index, "content_index": content_index, "delta": text}, ) emit( "response.output_text.done", {"type": "response.output_text.done", "item_id": item_id, "output_index": output_index, "content_index": content_index, "text": text}, ) emit( "response.content_part.done", {"type": "response.content_part.done", "item_id": item_id, "output_index": output_index, "content_index": content_index, "part": {"type": "output_text", "text": text, "annotations": []}}, ) emit( "response.output_item.done", {"type": "response.output_item.done", "output_index": output_index, "item": {**item, "status": "completed"}}, ) emit("response.completed", {"type": "response.completed", "response": {**payload, "status": "completed"}}) def _send_anthropic_error(self, status: int, message: str, error_type: str = "invalid_request_error") -> None: self._send_json( status, { "type": "error", "error": { "type": error_type, "message": message, }, }, ) def do_OPTIONS(self) -> None: self.send_response(HTTPStatus.NO_CONTENT) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type, X-API-Key, anthropic-version, anthropic-beta") self.send_header("Access-Control-Allow-Methods", "GET, HEAD, POST, OPTIONS") self.end_headers() def do_GET(self) -> None: if self._request_path() == "/health": self._send_json(HTTPStatus.OK, {"ok": True}) return if self._request_path() == "/v1/models" and self.config.list_public_model_ids(): self._send_json(HTTPStatus.OK, self._models_payload()) return status, body, headers = self._proxy_upstream_raw("GET", self.path) self._copy_upstream_response(status, body, headers) def do_HEAD(self) -> None: if self._request_path() == "/health": self._send_json(HTTPStatus.OK, {"ok": True}) return if self._request_path() == "/v1/models" and self.config.list_public_model_ids(): self._send_json(HTTPStatus.OK, self._models_payload()) return status, body, headers = self._proxy_upstream_raw("HEAD", self.path) self._copy_upstream_response(status, body, headers) def do_POST(self) -> None: try: request_path = self._request_path() if request_path == "/v1/messages": payload = self._read_json_body() openai_payload = _build_openai_payload_from_anthropic(payload) requested_model = str(openai_payload.get("model") or "").strip() if not requested_model: raise ValueError("model is required") stream_requested = bool(payload.get("stream")) upstream_openai_payload = dict(openai_payload) upstream_openai_payload["stream"] = False has_tools = isinstance(openai_payload.get("tools"), list) and bool(openai_payload["tools"]) has_tool_history = _has_tool_history(openai_payload.get("messages")) if (has_tools or has_tool_history) and self.config.should_passthrough_native_tools(requested_model): upstream_response = self._call_upstream_chat_for_requested_model(upstream_openai_payload, requested_model) elif not has_tools and not has_tool_history: upstream_response = self._call_upstream_chat_for_requested_model(upstream_openai_payload, requested_model) else: upstream_payload, _, _ = self._build_transformed_payload(upstream_openai_payload) upstream_response = self._call_upstream_chat(upstream_payload) if requested_model != str(upstream_response.get("model") or "").strip(): upstream_response["model"] = requested_model anthropic_payload = _anthropic_message_from_openai_response( upstream_response, requested_model, had_tools=(has_tools or has_tool_history), ) if stream_requested: self._send_streamed_anthropic_message(anthropic_payload) return self._send_json(HTTPStatus.OK, anthropic_payload) return if request_path == "/v1/responses": payload = self._read_json_body() openai_payload = _build_openai_payload_from_responses(payload) requested_model = str(openai_payload.get("model") or "").strip() if not requested_model: raise ValueError("model is required") stream_requested = bool(payload.get("stream")) upstream_openai_payload = dict(openai_payload) upstream_openai_payload["stream"] = False has_tools = isinstance(openai_payload.get("tools"), list) and bool(openai_payload["tools"]) has_tool_history = _has_tool_history(openai_payload.get("messages")) if (has_tools or has_tool_history) and self.config.should_passthrough_native_tools(requested_model): upstream_response = self._call_upstream_chat_for_requested_model(upstream_openai_payload, requested_model) elif not has_tools and not has_tool_history: upstream_response = self._call_upstream_chat_for_requested_model(upstream_openai_payload, requested_model) else: upstream_payload, _, _ = self._build_transformed_payload(upstream_openai_payload) upstream_response = self._call_upstream_chat(upstream_payload) if requested_model != str(upstream_response.get("model") or "").strip(): upstream_response["model"] = requested_model responses_payload = _responses_object_from_openai_response( upstream_response, requested_model, had_tools=(has_tools or has_tool_history), ) if stream_requested: self._send_streamed_responses(responses_payload) return self._send_json(HTTPStatus.OK, responses_payload) return if request_path != "/v1/chat/completions": body = self._read_body_bytes() status, response_body, headers = self._proxy_upstream_raw("POST", self.path, body) self._copy_upstream_response(status, response_body, headers) return payload = self._read_json_body() requested_model = str(payload.get("model") or "").strip() if not requested_model: raise ValueError("model is required") has_tools = isinstance(payload.get("tools"), list) and bool(payload["tools"]) if (has_tools or _has_tool_history(payload.get("messages"))) and self.config.should_passthrough_native_tools(requested_model): self._proxy_native_tool_request(payload, requested_model) return if not has_tools and not _has_tool_history(payload.get("messages")): self._proxy_native_tool_request(payload, requested_model) return upstream_payload, requested_model, stream = self._build_transformed_payload(payload) upstream_response = self._call_upstream_chat(upstream_payload) response_payload = self._response_payload(upstream_response, requested_model, has_tools) if stream: self._send_streamed_completion(response_payload) return self._send_json(HTTPStatus.OK, response_payload) except ConfigError as exc: if self._request_path() == "/v1/messages": self._send_anthropic_error(HTTPStatus.BAD_REQUEST, str(exc)) return self._send_json(HTTPStatus.BAD_REQUEST, {"error": {"message": str(exc), "type": "invalid_request_error"}}) except ValueError as exc: if self._request_path() == "/v1/messages": self._send_anthropic_error(HTTPStatus.BAD_REQUEST, str(exc)) return self._send_json(HTTPStatus.BAD_REQUEST, {"error": {"message": str(exc), "type": "invalid_request_error"}}) except UpstreamHTTPError as exc: if self._request_path() == "/v1/messages": message = f"upstream returned {exc.status_code}" try: parsed_error = json.loads(exc.body.decode("utf-8")) if isinstance(parsed_error, dict): error_payload = parsed_error.get("error") if isinstance(error_payload, dict) and isinstance(error_payload.get("message"), str): message = error_payload["message"] except (UnicodeDecodeError, json.JSONDecodeError): pass self._send_anthropic_error(exc.status_code, message, "api_error") return self._copy_upstream_response(exc.status_code, exc.body, exc.headers) except Exception as exc: if self._request_path() == "/v1/messages": self._send_anthropic_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc), "api_error") return self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": {"message": str(exc), "type": "server_error"}}) def create_server(config: AppConfig) -> ThreadingHTTPServer: return BridgeHTTPServer((config.host, config.port), ToolBridgeHandler, config) def serve(config: AppConfig) -> None: server = create_server(config) print( f"newapi-tool-bridge listening on http://{server.server_address[0]}:{server.server_address[1]}" ) server.serve_forever() if __name__ == "__main__": serve(AppConfig.from_env())