anthropic proxy

gistfile1.txt

#!/usr/bin/env python3
"""
Anthropic-to-OpenAI Proxy for Claude Code CLI.

Accepts native Anthropic API requests (/v1/messages) and translates them
to OpenAI chat completion format for OpenAI-compatible backends that host
Anthropic models (e.g., Azure AI, Bedrock gateways, MSKONG).

Fixes the gaps in LiteLLM's experimental pass-through adapter:
- Flattens system message arrays to strings
- Strips cache_control hints
- Drops unsupported params (thinking, output_config, effort)
- Proper Anthropic tool → OpenAI function conversion
- Streaming OpenAI responses → Anthropic SSE events
"""

import asyncio
import json
import logging
import os
import sys
import time
import uuid
from typing import Any

import httpx
import tiktoken
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
import uvicorn

# tiktoken encoder — cl100k_base is close to Anthropic's tokenizer
_enc = tiktoken.get_encoding("cl100k_base")


def count_tokens(text: str) -> int:
    """Count tokens using tiktoken."""
    if not text:
        return 0
    return len(_enc.encode(text))


def count_messages_tokens(messages: list[dict]) -> int:
    """Estimate token count for an OpenAI messages array."""
    total = 0
    for msg in messages:
        total += 4  # role + message overhead
        content = msg.get("content", "")
        if isinstance(content, str):
            total += count_tokens(content)
        elif isinstance(content, list):
            for part in content:
                if isinstance(part, dict) and "text" in part:
                    total += count_tokens(part["text"])
        if "tool_calls" in msg:
            for tc in msg["tool_calls"]:
                func = tc.get("function", {})
                total += count_tokens(func.get("name", ""))
                total += count_tokens(func.get("arguments", ""))
    return total


def count_tools_tokens(tools: list[dict]) -> int:
    """Estimate token count for tool definitions."""
    # Tool definitions are verbose — count the JSON representation
    return count_tokens(json.dumps(tools))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("proxy")

# Suppress noisy loggers
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

UPSTREAM_BASE = os.environ.get(
    "UPSTREAM_BASE_URL", "https://mskongai.use.ucdp.net/api"
)
UPSTREAM_API_KEY = os.environ.get("MSKONG_API_KEY", "")

# model_name sent by Claude CLI → actual model name on the backend
MODEL_MAP: dict[str, str] = {}

# Loaded from config file if present
CONFIG_PATH = os.path.expanduser("~/dotfiles/anthropic-proxy.json")

app = FastAPI()


def load_config():
    global MODEL_MAP, UPSTREAM_BASE, UPSTREAM_API_KEY
    if os.path.exists(CONFIG_PATH):
        with open(CONFIG_PATH) as f:
            cfg = json.load(f)
        MODEL_MAP = cfg.get("model_map", MODEL_MAP)
        if "upstream_base_url" in cfg:
            UPSTREAM_BASE = cfg["upstream_base_url"]
        if "upstream_api_key_env" in cfg:
            UPSTREAM_API_KEY = os.environ.get(cfg["upstream_api_key_env"], UPSTREAM_API_KEY)


load_config()


# ---------------------------------------------------------------------------
# Request conversion: Anthropic → OpenAI
# ---------------------------------------------------------------------------


def flatten_system(system: Any) -> str:
    """Convert Anthropic system (string or array of content blocks) to a plain string."""
    if isinstance(system, str):
        return system
    if isinstance(system, list):
        parts = []
        for block in system:
            if isinstance(block, dict):
                parts.append(block.get("text", ""))
            elif isinstance(block, str):
                parts.append(block)
        return "\n".join(parts)
    return str(system)


def convert_content_blocks(content: Any) -> Any:
    """Convert Anthropic content blocks to OpenAI format.

    Anthropic sends content as either a string or an array of typed blocks.
    OpenAI expects a string or an array of {type, text} / {type, image_url} objects.
    We strip cache_control and other Anthropic-specific fields.
    """
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return str(content)

    openai_parts = []
    for block in content:
        if not isinstance(block, dict):
            openai_parts.append({"type": "text", "text": str(block)})
            continue

        btype = block.get("type", "text")

        if btype == "text":
            openai_parts.append({"type": "text", "text": block.get("text", "")})

        elif btype == "image":
            # Anthropic: {type: "image", source: {type: "base64", media_type, data}}
            source = block.get("source", {})
            if source.get("type") == "base64":
                data_uri = f"data:{source.get('media_type', 'image/png')};base64,{source.get('data', '')}"
                openai_parts.append({
                    "type": "image_url",
                    "image_url": {"url": data_uri},
                })

        elif btype == "tool_use":
            # Tool use blocks in assistant messages — handled at message level
            pass

        elif btype == "tool_result":
            # Tool results — handled at message level
            pass

        else:
            # Unknown block type — pass text if available
            if "text" in block:
                openai_parts.append({"type": "text", "text": block["text"]})

    # If all blocks were text, simplify to a single string
    if len(openai_parts) == 1 and openai_parts[0]["type"] == "text":
        return openai_parts[0]["text"]
    if all(p["type"] == "text" for p in openai_parts):
        return "\n".join(p["text"] for p in openai_parts)

    return openai_parts if openai_parts else ""


def convert_messages(anthropic_messages: list[dict]) -> list[dict]:
    """Convert Anthropic messages array to OpenAI messages array."""
    openai_messages = []

    for msg in anthropic_messages:
        role = msg.get("role", "user")
        content = msg.get("content", "")

        if role == "user":
            # Check for tool_result blocks mixed with text
            if isinstance(content, list):
                tool_results = [b for b in content if isinstance(b, dict) and b.get("type") == "tool_result"]
                other_blocks = [b for b in content if not (isinstance(b, dict) and b.get("type") == "tool_result")]

                # Add tool result messages first
                for tr in tool_results:
                    tr_content = tr.get("content", "")
                    if isinstance(tr_content, list):
                        tr_content = convert_content_blocks(tr_content)
                    elif isinstance(tr_content, dict):
                        tr_content = tr_content.get("text", str(tr_content))
                    openai_messages.append({
                        "role": "tool",
                        "tool_call_id": tr.get("tool_use_id", ""),
                        "content": tr_content if isinstance(tr_content, str) else json.dumps(tr_content),
                    })

                # Add remaining content as user message
                if other_blocks:
                    openai_messages.append({
                        "role": "user",
                        "content": convert_content_blocks(other_blocks),
                    })
            else:
                openai_messages.append({
                    "role": "user",
                    "content": convert_content_blocks(content),
                })

        elif role == "assistant":
            if isinstance(content, list):
                # Separate text content from tool_use blocks
                text_blocks = [b for b in content if isinstance(b, dict) and b.get("type") == "text"]
                tool_blocks = [b for b in content if isinstance(b, dict) and b.get("type") == "tool_use"]
                # Also handle thinking blocks — just skip them
                # thinking_blocks are type "thinking" — we drop them

                text_content = ""
                if text_blocks:
                    text_content = "\n".join(b.get("text", "") for b in text_blocks)

                tool_calls = []
                for tb in tool_blocks:
                    tool_calls.append({
                        "id": tb.get("id", f"call_{uuid.uuid4().hex[:8]}"),
                        "type": "function",
                        "function": {
                            "name": tb.get("name", ""),
                            "arguments": json.dumps(tb.get("input", {})),
                        },
                    })

                assistant_msg: dict[str, Any] = {"role": "assistant"}
                if text_content:
                    assistant_msg["content"] = text_content
                else:
                    assistant_msg["content"] = None
                if tool_calls:
                    assistant_msg["tool_calls"] = tool_calls

                openai_messages.append(assistant_msg)
            else:
                openai_messages.append({
                    "role": "assistant",
                    "content": convert_content_blocks(content),
                })

        else:
            openai_messages.append({
                "role": role,
                "content": convert_content_blocks(content),
            })

    return openai_messages


def convert_tools(anthropic_tools: list[dict]) -> list[dict]:
    """Convert Anthropic tool definitions to OpenAI function tool format."""
    openai_tools = []
    for tool in anthropic_tools:
        openai_tool: dict[str, Any] = {
            "type": "function",
            "function": {
                "name": tool.get("name", ""),
            },
        }
        if "description" in tool:
            openai_tool["function"]["description"] = tool["description"]
        if "input_schema" in tool:
            openai_tool["function"]["parameters"] = tool["input_schema"]
        openai_tools.append(openai_tool)
    return openai_tools


def convert_tool_choice(anthropic_tc: Any) -> Any:
    """Convert Anthropic tool_choice to OpenAI format."""
    if anthropic_tc is None:
        return None
    if isinstance(anthropic_tc, str):
        mapping = {"auto": "auto", "any": "required", "none": "none"}
        return mapping.get(anthropic_tc, "auto")
    if isinstance(anthropic_tc, dict):
        tc_type = anthropic_tc.get("type", "auto")
        if tc_type == "auto":
            return "auto"
        elif tc_type == "any":
            return "required"
        elif tc_type == "tool":
            return {"type": "function", "function": {"name": anthropic_tc.get("name", "")}}
    return None


def anthropic_to_openai(body: dict) -> dict:
    """Convert a full Anthropic /v1/messages request to OpenAI /v1/chat/completions."""
    openai_body: dict[str, Any] = {}

    # Model
    model = body.get("model", "")
    openai_body["model"] = MODEL_MAP.get(model, model)

    # Messages
    messages = convert_messages(body.get("messages", []))

    # System message — prepend as first message
    if "system" in body and body["system"]:
        system_text = flatten_system(body["system"])
        if system_text.strip():
            messages.insert(0, {"role": "system", "content": system_text})

    openai_body["messages"] = messages

    # Max tokens
    if "max_tokens" in body:
        openai_body["max_tokens"] = body["max_tokens"]

    # Temperature
    if "temperature" in body:
        openai_body["temperature"] = body["temperature"]

    # Top P
    if "top_p" in body:
        openai_body["top_p"] = body["top_p"]

    # Stop sequences
    if "stop_sequences" in body:
        openai_body["stop"] = body["stop_sequences"]

    # Stream
    openai_body["stream"] = body.get("stream", True)
    if openai_body["stream"]:
        openai_body["stream_options"] = {"include_usage": True}

    # Tools
    if "tools" in body and body["tools"]:
        openai_body["tools"] = convert_tools(body["tools"])

    # Tool choice
    if "tool_choice" in body:
        tc = convert_tool_choice(body["tool_choice"])
        if tc is not None:
            openai_body["tool_choice"] = tc

    # Metadata — pass user if present
    if "metadata" in body and isinstance(body["metadata"], dict):
        if "user_id" in body["metadata"]:
            openai_body["user"] = body["metadata"]["user_id"]

    # Explicitly drop Anthropic-only params:
    # thinking, output_config, cache_control, anthropic_version, etc.
    # These have no OpenAI equivalent.

    return openai_body


# ---------------------------------------------------------------------------
# Response conversion: OpenAI → Anthropic (streaming)
# ---------------------------------------------------------------------------


def make_message_start(model: str, msg_id: str) -> dict:
    return {
        "type": "message_start",
        "message": {
            "id": msg_id,
            "type": "message",
            "role": "assistant",
            "content": [],
            "model": model,
            "stop_reason": None,
            "stop_sequence": None,
            "usage": {"input_tokens": 0, "output_tokens": 0},
        },
    }


def make_content_block_start(index: int, block_type: str = "text", **kwargs) -> dict:
    block: dict[str, Any] = {"type": block_type}
    if block_type == "text":
        block["text"] = ""
    elif block_type == "tool_use":
        block["id"] = kwargs.get("id", "")
        block["name"] = kwargs.get("name", "")
        block["input"] = {}
    return {"type": "content_block_start", "index": index, "content_block": block}


def make_content_block_delta(index: int, block_type: str = "text", **kwargs) -> dict:
    if block_type == "text":
        delta = {"type": "text_delta", "text": kwargs.get("text", "")}
    elif block_type == "tool_use":
        delta = {"type": "input_json_delta", "partial_json": kwargs.get("arguments", "")}
    else:
        delta = {"type": "text_delta", "text": ""}
    return {"type": "content_block_delta", "index": index, "delta": delta}


def make_content_block_stop(index: int) -> dict:
    return {"type": "content_block_stop", "index": index}


def make_message_delta(stop_reason: str = "end_turn", input_tokens: int = 0, output_tokens: int = 0) -> dict:
    return {
        "type": "message_delta",
        "delta": {"stop_reason": stop_reason},
        "usage": {"input_tokens": input_tokens, "output_tokens": output_tokens},
    }


def make_message_stop() -> dict:
    return {"type": "message_stop"}


def sse_event(event_type: str, data: dict) -> str:
    return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"


async def stream_openai_to_anthropic(
    response: httpx.Response, model: str, request_start: float = 0,
    estimated_input_tokens: int = 0,
) -> AsyncIterator[str]:
    """Convert an OpenAI streaming response to Anthropic SSE events."""
    msg_id = f"msg_{uuid.uuid4().hex[:24]}"

    # Send message_start
    yield sse_event("message_start", make_message_start(model, msg_id))

    content_index = 0
    current_text_started = False
    tool_call_states: dict[int, dict] = {}  # OpenAI tool_call index → state
    tool_content_indices: dict[int, int] = {}  # OpenAI index → Anthropic content index
    input_tokens = 0
    output_tokens = 0
    output_text_parts: list[str] = []  # collect output for token counting
    stop_reason = "end_turn"
    saw_content = False

    async for line in response.aiter_lines():
        if not line.startswith("data: "):
            continue
        data_str = line[6:].strip()
        if data_str == "[DONE]":
            break

        try:
            chunk = json.loads(data_str)
        except json.JSONDecodeError:
            continue

        # Extract usage from chunk if present (OpenAI sends this in a final
        # chunk with choices=[] when stream_options.include_usage is true)
        if "usage" in chunk and chunk["usage"]:
            usage = chunk["usage"]
            input_tokens = usage.get("prompt_tokens", input_tokens)
            output_tokens = usage.get("completion_tokens", output_tokens)

        choices = chunk.get("choices", [])
        if not choices:
            continue  # usage-only chunks have no choices — already extracted above

        choice = choices[0]
        delta = choice.get("delta", {})
        finish_reason = choice.get("finish_reason")

        # Text content
        text = delta.get("content")
        if text is not None:
            if not current_text_started:
                yield sse_event(
                    "content_block_start",
                    make_content_block_start(content_index, "text"),
                )
                current_text_started = True
                saw_content = True

            if text:
                output_text_parts.append(text)
                yield sse_event(
                    "content_block_delta",
                    make_content_block_delta(content_index, "text", text=text),
                )

        # Tool calls
        if "tool_calls" in delta:
            for tc in delta["tool_calls"]:
                tc_index = tc.get("index", 0)

                if tc_index not in tool_call_states:
                    # Close text block if open
                    if current_text_started:
                        yield sse_event(
                            "content_block_stop",
                            make_content_block_stop(content_index),
                        )
                        content_index += 1
                        current_text_started = False

                    # Start new tool_use block
                    tool_id = tc.get("id", f"toolu_{uuid.uuid4().hex[:12]}")
                    func = tc.get("function", {})
                    tool_name = func.get("name", "")
                    tool_call_states[tc_index] = {
                        "id": tool_id,
                        "name": tool_name,
                        "arguments": "",
                    }
                    tool_content_indices[tc_index] = content_index

                    yield sse_event(
                        "content_block_start",
                        make_content_block_start(
                            content_index, "tool_use", id=tool_id, name=tool_name
                        ),
                    )
                    saw_content = True
                    content_index += 1

                # Accumulate arguments
                func = tc.get("function", {})
                args_chunk = func.get("arguments", "")
                if args_chunk:
                    output_text_parts.append(args_chunk)
                    tool_call_states[tc_index]["arguments"] += args_chunk
                    yield sse_event(
                        "content_block_delta",
                        make_content_block_delta(
                            tool_content_indices[tc_index],
                            "tool_use",
                            arguments=args_chunk,
                        ),
                    )

        # Finish reason
        if finish_reason:
            reason_map = {
                "stop": "end_turn",
                "length": "max_tokens",
                "tool_calls": "tool_use",
                "content_filter": "end_turn",
            }
            stop_reason = reason_map.get(finish_reason, "end_turn")

    # Close any open blocks
    if current_text_started:
        yield sse_event("content_block_stop", make_content_block_stop(content_index))
    elif not saw_content:
        # If we never got any content, send an empty text block so the response is valid
        yield sse_event("content_block_start", make_content_block_start(0, "text"))
        yield sse_event("content_block_delta", make_content_block_delta(0, "text", text=""))
        yield sse_event("content_block_stop", make_content_block_stop(0))

    # Close tool blocks
    for tc_idx, anthropic_idx in tool_content_indices.items():
        yield sse_event("content_block_stop", make_content_block_stop(anthropic_idx))

    # Count output tokens with tiktoken if backend didn't report
    if not output_tokens:
        output_tokens = count_tokens("".join(output_text_parts))
    if not input_tokens:
        input_tokens = estimated_input_tokens

    elapsed = time.time() - request_start if request_start else 0
    log.info(
        f"← {model} | {input_tokens:,} in, {output_tokens:,} out, "
        f"stop={stop_reason}, {elapsed:.1f}s"
    )

    # Send message_delta and message_stop — inject counted tokens
    yield sse_event(
        "message_delta",
        make_message_delta(stop_reason, input_tokens, output_tokens),
    )
    yield sse_event("message_stop", make_message_stop())


# We need to use the proper async iterator type
from typing import AsyncIterator


# ---------------------------------------------------------------------------
# Non-streaming response conversion
# ---------------------------------------------------------------------------


def openai_response_to_anthropic(openai_resp: dict, model: str) -> dict:
    """Convert a non-streaming OpenAI response to Anthropic format."""
    msg_id = f"msg_{uuid.uuid4().hex[:24]}"
    content = []
    stop_reason = "end_turn"

    choices = openai_resp.get("choices", [])
    if choices:
        choice = choices[0]
        message = choice.get("message", {})
        finish_reason = choice.get("finish_reason", "stop")

        reason_map = {
            "stop": "end_turn",
            "length": "max_tokens",
            "tool_calls": "tool_use",
        }
        stop_reason = reason_map.get(finish_reason, "end_turn")

        if message.get("content"):
            content.append({"type": "text", "text": message["content"]})

        for tc in message.get("tool_calls", []):
            func = tc.get("function", {})
            try:
                tool_input = json.loads(func.get("arguments", "{}"))
            except json.JSONDecodeError:
                tool_input = {}
            content.append({
                "type": "tool_use",
                "id": tc.get("id", f"toolu_{uuid.uuid4().hex[:12]}"),
                "name": func.get("name", ""),
                "input": tool_input,
            })

    usage = openai_resp.get("usage", {})

    return {
        "id": msg_id,
        "type": "message",
        "role": "assistant",
        "content": content,
        "model": model,
        "stop_reason": stop_reason,
        "stop_sequence": None,
        "usage": {
            "input_tokens": usage.get("prompt_tokens", 0),
            "output_tokens": usage.get("completion_tokens", 0),
        },
    }


# ---------------------------------------------------------------------------
# HTTP endpoint
# ---------------------------------------------------------------------------


@app.post("/v1/messages")
async def anthropic_messages(request: Request):
    body = await request.json()
    original_model = body.get("model", "unknown")
    is_stream = body.get("stream", True)
    request_start = time.time()

    # Convert request
    openai_body = anthropic_to_openai(body)

    # Count tokens with tiktoken
    input_token_count = count_messages_tokens(openai_body.get("messages", []))
    if "tools" in openai_body:
        input_token_count += count_tools_tokens(openai_body["tools"])

    msg_count = len(body.get("messages", []))
    tool_count = len(body.get("tools", []))
    log.info(
        f"→ {original_model} | {msg_count} msgs, {tool_count} tools, "
        f"{input_token_count:,} input tokens, stream={is_stream}"
    )

    # Build headers for upstream
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {UPSTREAM_API_KEY}",
    }

    upstream_url = f"{UPSTREAM_BASE.rstrip('/')}/v1/chat/completions"

    if is_stream:
        # Streaming response
        client = httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=30.0))
        try:
            upstream_resp = await client.send(
                client.build_request(
                    "POST",
                    upstream_url,
                    json=openai_body,
                    headers=headers,
                ),
                stream=True,
            )

            if upstream_resp.status_code != 200:
                error_body = await upstream_resp.aread()
                await client.aclose()
                return Response(
                    content=json.dumps({
                        "type": "error",
                        "error": {
                            "type": "api_error",
                            "message": f"Upstream error {upstream_resp.status_code}: {error_body.decode()}",
                        },
                    }),
                    status_code=upstream_resp.status_code,
                    media_type="application/json",
                )

            # Determine model name for response
            resp_model = MODEL_MAP.get(original_model, original_model)

            async def generate():
                try:
                    async for event in stream_openai_to_anthropic(
                        upstream_resp, resp_model, request_start, input_token_count
                    ):
                        yield event
                finally:
                    await upstream_resp.aclose()
                    await client.aclose()

            return StreamingResponse(
                generate(),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "X-Accel-Buffering": "no",
                },
            )

        except Exception as e:
            await client.aclose()
            return Response(
                content=json.dumps({
                    "type": "error",
                    "error": {"type": "api_error", "message": str(e)},
                }),
                status_code=500,
                media_type="application/json",
            )

    else:
        # Non-streaming response
        async with httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=30.0)) as client:
            upstream_resp = await client.post(
                upstream_url, json=openai_body, headers=headers
            )

            if upstream_resp.status_code != 200:
                return Response(
                    content=json.dumps({
                        "type": "error",
                        "error": {
                            "type": "api_error",
                            "message": f"Upstream error {upstream_resp.status_code}: {upstream_resp.text}",
                        },
                    }),
                    status_code=upstream_resp.status_code,
                    media_type="application/json",
                )

            openai_data = upstream_resp.json()
            resp_model = MODEL_MAP.get(original_model, original_model)
            anthropic_resp = openai_response_to_anthropic(openai_data, resp_model)
            elapsed = time.time() - request_start
            u = anthropic_resp.get("usage", {})
            log.info(
                f"← {resp_model} | {u.get('input_tokens', 0)} in, "
                f"{u.get('output_tokens', 0)} out, {elapsed:.1f}s"
            )
            return Response(
                content=json.dumps(anthropic_resp),
                media_type="application/json",
            )


# Health check
@app.get("/health")
async def health():
    return {"status": "ok"}


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Anthropic-to-OpenAI proxy")
    parser.add_argument("--port", type=int, default=4000)
    parser.add_argument("--host", type=str, default="0.0.0.0")
    args = parser.parse_args()

    print(f"Starting Anthropic→OpenAI proxy on {args.host}:{args.port}")
    print(f"Upstream: {UPSTREAM_BASE}")
    print(f"Model map: {MODEL_MAP or '(passthrough)'}")

    uvicorn.run(app, host=args.host, port=args.port, access_log=False, log_level="warning")