Add cloud provider support via LiteLLM backend

Enables Headroom proxy to work with AWS Bedrock, Google Vertex AI,
Azure OpenAI, and 100+ other providers via LiteLLM.

Usage:
headroom proxy --backend bedrock --region us-west-2
headroom proxy --backend vertex_ai --region us-central1
headroom proxy --backend azure --region eastus

Features:
- Automatic format translation (Anthropic API <-> provider APIs)
- Streaming support with proper SSE event translation
- Uses existing cloud credentials (AWS, GCP, Azure)
- Shorthand: --backend bedrock (expands to litellm-bedrock)
- Stats tracking per provider

README.md

@@ -213,6 +213,20 @@ ANTHROPIC_BASE_URL=http://localhost:8787 claude
 OPENAI_BASE_URL=http://localhost:8787/v1 cursor
 ```
 
+**Using AWS Bedrock, Google Vertex, or Azure?** Route through Headroom:
+
+```bash
+# AWS Bedrock (uses your AWS credentials)
+headroom proxy --backend bedrock --region us-west-2
+ANTHROPIC_BASE_URL=http://localhost:8787 claude
+
+# Google Vertex AI
+headroom proxy --backend vertex_ai --region us-central1
+
+# Azure OpenAI
+headroom proxy --backend azure --region eastus
+```
+
 ### Option 2: LangChain Integration
 
 ```bash

headroom/backends/__init__.py

@@ -0,0 +1,19 @@
+"""Headroom Backends - API translation layers for different LLM providers.
+
+Backends handle the translation between the proxy's canonical format
+(Anthropic Messages API) and provider-specific APIs.
+
+Uses LiteLLM for broad provider support:
+- bedrock: AWS Bedrock (Claude, Cohere, Mistral, etc.)
+- vertex_ai: Google Vertex AI (Claude, Gemini, etc.)
+- azure: Azure OpenAI (GPT-4, etc.)
+- And 100+ more providers...
+
+Usage:
+    headroom proxy --backend litellm-bedrock --region us-west-2
+"""
+
+from .base import Backend, BackendResponse, StreamEvent
+from .litellm import LiteLLMBackend
+
+__all__ = ["Backend", "BackendResponse", "StreamEvent", "LiteLLMBackend"]

headroom/backends/base.py

@@ -0,0 +1,122 @@
+"""Base backend interface for Headroom.
+
+Backends translate between the canonical Anthropic Messages API format
+(used by the proxy) and provider-specific APIs.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class BackendResponse:
+    """Standardized response from a backend."""
+
+    # Response body (Anthropic Messages API format)
+    body: dict[str, Any]
+
+    # HTTP status code
+    status_code: int = 200
+
+    # Response headers to forward
+    headers: dict[str, str] = field(default_factory=dict)
+
+    # Error message if any
+    error: str | None = None
+
+
+@dataclass
+class StreamEvent:
+    """A single event from a streaming response."""
+
+    # Event type (message_start, content_block_delta, etc.)
+    event_type: str
+
+    # Event data (Anthropic SSE format)
+    data: dict[str, Any]
+
+    # Raw SSE line to forward (if available)
+    raw_sse: str | None = None
+
+
+class Backend(ABC):
+    """Abstract base class for LLM API backends.
+
+    Backends are responsible for:
+    - Translating requests from Anthropic format to provider format
+    - Making API calls to the provider
+    - Translating responses back to Anthropic format
+    - Handling streaming
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Backend name (e.g., 'anthropic', 'bedrock')."""
+        ...
+
+    @abstractmethod
+    async def send_message(
+        self,
+        body: dict[str, Any],
+        headers: dict[str, str],
+    ) -> BackendResponse:
+        """Send a non-streaming message request.
+
+        Args:
+            body: Request body in Anthropic Messages API format.
+            headers: Request headers (may include API keys, etc.).
+
+        Returns:
+            BackendResponse with body in Anthropic Messages API format.
+        """
+        ...
+
+    @abstractmethod
+    def stream_message(
+        self,
+        body: dict[str, Any],
+        headers: dict[str, str],
+    ) -> AsyncIterator[StreamEvent]:
+        """Stream a message request.
+
+        Args:
+            body: Request body in Anthropic Messages API format.
+            headers: Request headers.
+
+        Yields:
+            StreamEvent objects in Anthropic SSE format.
+        """
+        ...
+
+    @abstractmethod
+    def map_model_id(self, anthropic_model: str) -> str:
+        """Map Anthropic model ID to provider model ID.
+
+        Args:
+            anthropic_model: Model ID in Anthropic format (e.g., 'claude-3-opus-20240229').
+
+        Returns:
+            Model ID in provider format.
+        """
+        ...
+
+    @abstractmethod
+    def supports_model(self, model: str) -> bool:
+        """Check if this backend supports a model.
+
+        Args:
+            model: Model ID (can be Anthropic or provider format).
+
+        Returns:
+            True if model is supported.
+        """
+        ...
+
+    async def close(self) -> None:  # noqa: B027
+        """Clean up resources (e.g., close HTTP clients)."""
+        pass

headroom/backends/litellm.py

@@ -0,0 +1,421 @@
+"""LiteLLM-based backend for Headroom.
+
+Uses LiteLLM to support 100+ providers with minimal code:
+- AWS Bedrock: model="bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0"
+- Azure OpenAI: model="azure/gpt-4"
+- Google Vertex: model="vertex_ai/claude-3-5-sonnet"
+- And many more...
+
+LiteLLM handles all the auth and format translation internally.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from collections.abc import AsyncIterator
+from typing import Any
+
+from .base import Backend, BackendResponse, StreamEvent
+
+logger = logging.getLogger(__name__)
+
+try:
+    import litellm
+    from litellm import acompletion
+
+    LITELLM_AVAILABLE = True
+except ImportError:
+    LITELLM_AVAILABLE = False
+    litellm = None  # type: ignore
+    acompletion = None  # type: ignore
+
+
+# Model mapping: Anthropic model IDs -> LiteLLM model strings
+BEDROCK_MODEL_MAP = {
+    # Claude 4.5
+    "claude-opus-4-5-20251101": "bedrock/anthropic.claude-opus-4-5-20251101-v1:0",
+    "claude-sonnet-4-5-20250929": "bedrock/anthropic.claude-sonnet-4-5-20250929-v1:0",
+    # Claude 4
+    "claude-opus-4-20250514": "bedrock/anthropic.claude-opus-4-20250514-v1:0",
+    "claude-sonnet-4-20250514": "bedrock/anthropic.claude-sonnet-4-20250514-v1:0",
+    # Claude 3.7
+    "claude-3-7-sonnet-20250219": "bedrock/anthropic.claude-3-7-sonnet-20250219-v1:0",
+    # Claude 3.5
+    "claude-3-5-sonnet-20241022": "bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0",
+    "claude-3-5-sonnet-20240620": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+    "claude-3-5-haiku-20241022": "bedrock/anthropic.claude-3-5-haiku-20241022-v1:0",
+    # Claude 3
+    "claude-3-opus-20240229": "bedrock/anthropic.claude-3-opus-20240229-v1:0",
+    "claude-3-sonnet-20240229": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0",
+    "claude-3-haiku-20240307": "bedrock/anthropic.claude-3-haiku-20240307-v1:0",
+}
+
+VERTEX_MODEL_MAP = {
+    "claude-3-5-sonnet-20241022": "vertex_ai/claude-3-5-sonnet-v2@20241022",
+    "claude-3-5-sonnet-20240620": "vertex_ai/claude-3-5-sonnet@20240620",
+    "claude-3-opus-20240229": "vertex_ai/claude-3-opus@20240229",
+    "claude-3-sonnet-20240229": "vertex_ai/claude-3-sonnet@20240229",
+    "claude-3-haiku-20240307": "vertex_ai/claude-3-haiku@20240307",
+}
+
+
+class LiteLLMBackend(Backend):
+    """Backend using LiteLLM for multi-provider support.
+
+    Supports any provider LiteLLM supports:
+    - bedrock: AWS Bedrock (uses AWS credentials)
+    - vertex_ai: Google Vertex AI (uses GCP credentials)
+    - azure: Azure OpenAI (uses Azure credentials)
+    - And 100+ more...
+    """
+
+    def __init__(
+        self,
+        provider: str = "bedrock",
+        region: str | None = None,
+        **kwargs: Any,
+    ):
+        """Initialize LiteLLM backend.
+
+        Args:
+            provider: LiteLLM provider prefix (bedrock, vertex_ai, azure, etc.)
+            region: Cloud region (provider-specific)
+            **kwargs: Additional provider-specific config
+        """
+        if not LITELLM_AVAILABLE:
+            raise ImportError(
+                "litellm is required for LiteLLMBackend. Install with: pip install litellm"
+            )
+
+        self.provider = provider
+        self.region = region
+        self.kwargs = kwargs
+
+        # Select model map based on provider
+        if provider == "bedrock":
+            self._model_map = BEDROCK_MODEL_MAP
+            # Set AWS region for litellm
+            if region:
+                litellm.set_verbose = False  # Reduce noise
+        elif provider == "vertex_ai":
+            self._model_map = VERTEX_MODEL_MAP
+        else:
+            self._model_map = {}
+
+        logger.info(f"LiteLLM backend initialized (provider={provider})")
+
+    @property
+    def name(self) -> str:
+        return f"litellm-{self.provider}"
+
+    def map_model_id(self, anthropic_model: str) -> str:
+        """Map Anthropic model ID to LiteLLM model string."""
+        # Check direct mapping
+        if anthropic_model in self._model_map:
+            return self._model_map[anthropic_model]
+
+        # If already has provider prefix, use as-is
+        if "/" in anthropic_model:
+            return anthropic_model
+
+        # Fallback: construct provider/model format
+        return f"{self.provider}/{anthropic_model}"
+
+    def supports_model(self, model: str) -> bool:
+        """Check if model is supported."""
+        return "claude" in model.lower() or model in self._model_map
+
+    def _convert_messages_for_litellm(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Convert Anthropic message format to LiteLLM/OpenAI format.
+
+        LiteLLM expects OpenAI-style messages but handles most Anthropic
+        content blocks automatically.
+        """
+        converted = []
+        for msg in messages:
+            role = msg.get("role", "user")
+            content = msg.get("content", "")
+
+            # Handle string content directly
+            if isinstance(content, str):
+                converted.append({"role": role, "content": content})
+                continue
+
+            # Handle content blocks (Anthropic style)
+            if isinstance(content, list):
+                # Check if it's simple text blocks only
+                text_parts = []
+                has_complex_content = False
+
+                for block in content:
+                    if isinstance(block, dict):
+                        if block.get("type") == "text":
+                            text_parts.append(block.get("text", ""))
+                        elif block.get("type") in ("tool_use", "tool_result", "image"):
+                            has_complex_content = True
+                            break
+
+                if not has_complex_content and text_parts:
+                    # Simple text - join into single string
+                    converted.append({"role": role, "content": "\n".join(text_parts)})
+                else:
+                    # Complex content - pass through (LiteLLM handles it)
+                    converted.append({"role": role, "content": content})
+
+        return converted
+
+    def _to_anthropic_response(
+        self,
+        litellm_response: Any,
+        original_model: str,
+    ) -> dict[str, Any]:
+        """Convert LiteLLM/OpenAI response to Anthropic format."""
+        msg_id = f"msg_{uuid.uuid4().hex[:24]}"
+
+        # Extract content from OpenAI format
+        choice = litellm_response.choices[0]
+        message = choice.message
+
+        # Build Anthropic content blocks
+        content = []
+        if message.content:
+            content.append({"type": "text", "text": message.content})
+
+        # Handle tool calls if present
+        if hasattr(message, "tool_calls") and message.tool_calls:
+            for tc in message.tool_calls:
+                content.append(
+                    {
+                        "type": "tool_use",
+                        "id": tc.id,
+                        "name": tc.function.name,
+                        "input": tc.function.arguments,
+                    }
+                )
+
+        # Map stop reason
+        stop_reason_map = {
+            "stop": "end_turn",
+            "length": "max_tokens",
+            "tool_calls": "tool_use",
+            "content_filter": "end_turn",
+        }
+        stop_reason = stop_reason_map.get(choice.finish_reason, "end_turn")
+
+        # Build usage
+        usage = {
+            "input_tokens": getattr(litellm_response.usage, "prompt_tokens", 0),
+            "output_tokens": getattr(litellm_response.usage, "completion_tokens", 0),
+        }
+
+        return {
+            "id": msg_id,
+            "type": "message",
+            "role": "assistant",
+            "content": content,
+            "model": original_model,
+            "stop_reason": stop_reason,
+            "stop_sequence": None,
+            "usage": usage,
+        }
+
+    async def send_message(
+        self,
+        body: dict[str, Any],
+        headers: dict[str, str],
+    ) -> BackendResponse:
+        """Send message via LiteLLM."""
+        original_model = body.get("model", "claude-3-5-sonnet-20241022")
+        litellm_model = self.map_model_id(original_model)
+
+        try:
+            # Convert messages
+            messages = self._convert_messages_for_litellm(body.get("messages", []))
+
+            # Build kwargs for litellm
+            kwargs: dict[str, Any] = {
+                "model": litellm_model,
+                "messages": messages,
+            }
+
+            # Optional parameters
+            if "max_tokens" in body:
+                kwargs["max_tokens"] = body["max_tokens"]
+            if "temperature" in body:
+                kwargs["temperature"] = body["temperature"]
+            if "top_p" in body:
+                kwargs["top_p"] = body["top_p"]
+            if "stop_sequences" in body:
+                kwargs["stop"] = body["stop_sequences"]
+
+            # System prompt (Anthropic puts it in body, OpenAI in messages)
+            if "system" in body:
+                system = body["system"]
+                if isinstance(system, str):
+                    kwargs["messages"].insert(0, {"role": "system", "content": system})
+                elif isinstance(system, list):
+                    # Anthropic list format
+                    system_text = " ".join(
+                        s.get("text", "") if isinstance(s, dict) else str(s) for s in system
+                    )
+                    kwargs["messages"].insert(0, {"role": "system", "content": system_text})
+
+            # AWS region for Bedrock
+            if self.provider == "bedrock" and self.region:
+                kwargs["aws_region_name"] = self.region
+
+            logger.debug(f"LiteLLM request: model={litellm_model}")
+
+            # Make the call
+            response = await acompletion(**kwargs)
+
+            # Convert to Anthropic format
+            anthropic_response = self._to_anthropic_response(response, original_model)
+
+            return BackendResponse(
+                body=anthropic_response,
+                status_code=200,
+                headers={"content-type": "application/json"},
+            )
+
+        except Exception as e:
+            logger.error(f"LiteLLM error: {e}")
+
+            # Map to Anthropic error format
+            error_type = "api_error"
+            status_code = 500
+
+            error_str = str(e).lower()
+            if "authentication" in error_str or "credentials" in error_str:
+                error_type = "authentication_error"
+                status_code = 401
+            elif "rate" in error_str or "limit" in error_str:
+                error_type = "rate_limit_error"
+                status_code = 429
+            elif "not found" in error_str:
+                error_type = "not_found_error"
+                status_code = 404
+
+            return BackendResponse(
+                body={
+                    "type": "error",
+                    "error": {"type": error_type, "message": str(e)},
+                },
+                status_code=status_code,
+                error=str(e),
+            )
+
+    async def stream_message(
+        self,
+        body: dict[str, Any],
+        headers: dict[str, str],
+    ) -> AsyncIterator[StreamEvent]:
+        """Stream message via LiteLLM."""
+        original_model = body.get("model", "claude-3-5-sonnet-20241022")
+        litellm_model = self.map_model_id(original_model)
+
+        try:
+            messages = self._convert_messages_for_litellm(body.get("messages", []))
+
+            kwargs: dict[str, Any] = {
+                "model": litellm_model,
+                "messages": messages,
+                "stream": True,
+            }
+
+            if "max_tokens" in body:
+                kwargs["max_tokens"] = body["max_tokens"]
+            if "temperature" in body:
+                kwargs["temperature"] = body["temperature"]
+            if "system" in body:
+                system = body["system"]
+                if isinstance(system, str):
+                    kwargs["messages"].insert(0, {"role": "system", "content": system})
+
+            if self.provider == "bedrock" and self.region:
+                kwargs["aws_region_name"] = self.region
+
+            msg_id = f"msg_{uuid.uuid4().hex[:24]}"
+
+            # Emit message_start
+            yield StreamEvent(
+                event_type="message_start",
+                data={
+                    "type": "message_start",
+                    "message": {
+                        "id": msg_id,
+                        "type": "message",
+                        "role": "assistant",
+                        "content": [],
+                        "model": original_model,
+                        "stop_reason": None,
+                        "stop_sequence": None,
+                        "usage": {"input_tokens": 0, "output_tokens": 0},
+                    },
+                },
+            )
+
+            # Emit content_block_start
+            yield StreamEvent(
+                event_type="content_block_start",
+                data={
+                    "type": "content_block_start",
+                    "index": 0,
+                    "content_block": {"type": "text", "text": ""},
+                },
+            )
+
+            # Stream content
+            response = await acompletion(**kwargs)
+            output_tokens = 0
+
+            async for chunk in response:
+                if hasattr(chunk, "choices") and chunk.choices:
+                    delta = chunk.choices[0].delta
+                    if hasattr(delta, "content") and delta.content:
+                        yield StreamEvent(
+                            event_type="content_block_delta",
+                            data={
+                                "type": "content_block_delta",
+                                "index": 0,
+                                "delta": {"type": "text_delta", "text": delta.content},
+                            },
+                        )
+                        output_tokens += 1  # Rough estimate
+
+            # Emit content_block_stop
+            yield StreamEvent(
+                event_type="content_block_stop",
+                data={"type": "content_block_stop", "index": 0},
+            )
+
+            # Emit message_delta with stop reason
+            yield StreamEvent(
+                event_type="message_delta",
+                data={
+                    "type": "message_delta",
+                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+                    "usage": {"output_tokens": output_tokens},
+                },
+            )
+
+            # Emit message_stop
+            yield StreamEvent(
+                event_type="message_stop",
+                data={"type": "message_stop"},
+            )
+
+        except Exception as e:
+            logger.error(f"LiteLLM streaming error: {e}")
+            yield StreamEvent(
+                event_type="error",
+                data={
+                    "type": "error",
+                    "error": {"type": "api_error", "message": str(e)},
+                },
+            )
+
+    async def close(self) -> None:  # noqa: B027
+        """Clean up (no-op for LiteLLM)."""
+        pass

headroom/cli.py

@@ -78,12 +78,26 @@ def cmd_proxy(args: argparse.Namespace) -> int:
         memory_inject_tools=not args.no_memory_tools,
         memory_inject_context=not args.no_memory_context,
         memory_top_k=args.memory_top_k,
+        # Backend (Anthropic direct, Bedrock, or LiteLLM)
+        backend=args.backend,
+        bedrock_region=args.bedrock_region or args.region,
+        bedrock_profile=args.bedrock_profile,
     )
 
     memory_status = "DISABLED"
     if config.memory_enabled:
         memory_status = f"ENABLED ({config.memory_backend})"
 
+    region = args.bedrock_region or args.region
+    backend_status = "Anthropic (direct API)"
+    if config.backend != "anthropic":
+        # Normalize: "bedrock" -> "litellm-bedrock"
+        backend = config.backend
+        if not backend.startswith("litellm-"):
+            backend = f"litellm-{backend}"
+        provider = backend.replace("litellm-", "")
+        backend_status = f"{provider.upper()} via LiteLLM (region={region})"
+
     print(f"""
 ╔═══════════════════════════════════════════════════════════════════════╗
 ║                         HEADROOM PROXY                                 ║
@@ -93,12 +107,14 @@ def cmd_proxy(args: argparse.Namespace) -> int:
 Starting proxy server...
 
   URL:          http://{config.host}:{config.port}
+  Backend:      {backend_status}
   Optimization: {"ENABLED" if config.optimize else "DISABLED"}
   Caching:      {"ENABLED" if config.cache_enabled else "DISABLED"}
   Rate Limit:   {"ENABLED" if config.rate_limit_enabled else "DISABLED"}
   Memory:       {memory_status}
 
 Usage with Claude Code:
+  {"unset CLAUDE_CODE_USE_BEDROCK  # Important!" if config.backend == "bedrock" else ""}
   ANTHROPIC_BASE_URL=http://{config.host}:{config.port} claude
 
 Usage with OpenAI-compatible clients:
@@ -544,6 +560,29 @@ Documentation: https://github.com/headroom-sdk/headroom
         default=10,
         help="Number of memories to inject as context (default: 10)",
     )
+    # Backend configuration
+    proxy_parser.add_argument(
+        "--backend",
+        default="anthropic",
+        help=(
+            "API backend: 'anthropic' (direct), 'bedrock' (AWS), "
+            "or 'litellm-<provider>' (e.g., litellm-bedrock, litellm-vertex)"
+        ),
+    )
+    proxy_parser.add_argument(
+        "--region",
+        default="us-west-2",
+        help="Cloud region for Bedrock/Vertex/etc (default: us-west-2)",
+    )
+    proxy_parser.add_argument(
+        "--bedrock-region",
+        default=None,
+        help="(deprecated, use --region) AWS region for Bedrock",
+    )
+    proxy_parser.add_argument(
+        "--bedrock-profile",
+        help="AWS profile name for Bedrock (default: use default credentials)",
+    )
     proxy_parser.set_defaults(func=cmd_proxy)
 
     # Memory eval command

headroom/proxy/server.py

@@ -53,6 +53,8 @@ except ImportError:
 # Add parent to path for imports
 sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 
+from headroom.backends import LiteLLMBackend
+from headroom.backends.base import Backend
 from headroom.cache.compression_feedback import get_compression_feedback
 from headroom.cache.compression_store import get_compression_store
 from headroom.ccr import (
@@ -206,6 +208,12 @@ class ProxyConfig:
     openai_api_url: str | None = None  # Custom OpenAI API URL override
     gemini_api_url: str | None = None  # Custom Gemini API URL override
 
+    # Backend: "anthropic" (direct API), "bedrock" (AWS Bedrock), or "litellm-*" (via LiteLLM)
+    # LiteLLM backends: "litellm-bedrock", "litellm-vertex", "litellm-azure", etc.
+    backend: str = "anthropic"
+    bedrock_region: str = "us-west-2"  # AWS region for Bedrock/LiteLLM
+    bedrock_profile: str | None = None  # AWS profile (optional)
+
     # Optimization
     optimize: bool = True
     image_optimize: bool = True  # Compress images using trained ML router
@@ -981,6 +989,29 @@ class HeadroomProxy:
         # HTTP client
         self.http_client: httpx.AsyncClient | None = None
 
+        # Backend for Anthropic API (direct or via LiteLLM)
+        # Supports: "anthropic" (direct), "bedrock", "vertex", or "litellm-<provider>"
+        self.anthropic_backend: Backend | None = None
+        if config.backend != "anthropic":
+            # Normalize backend name: "bedrock" -> "litellm-bedrock"
+            backend = config.backend
+            if not backend.startswith("litellm-"):
+                backend = f"litellm-{backend}"
+            provider = backend.replace("litellm-", "")
+
+            try:
+                self.anthropic_backend = LiteLLMBackend(
+                    provider=provider,
+                    region=config.bedrock_region,
+                )
+                logger.info(
+                    f"LiteLLM backend enabled (provider={provider}, region={config.bedrock_region})"
+                )
+            except ImportError as e:
+                logger.warning(f"LiteLLM backend not available: {e}")
+            except Exception as e:
+                logger.error(f"Failed to initialize LiteLLM backend: {e}")
+
         # Request counter for IDs
         self._request_counter = 0
         self._request_counter_lock = asyncio.Lock()
@@ -1583,7 +1614,70 @@ class HeadroomProxy:
         if tools is not None:
             body["tools"] = tools
 
-        # Forward request
+        # Forward request - use Bedrock backend if configured, otherwise direct API
+        if self.anthropic_backend is not None:
+            # Route through Bedrock backend
+            try:
+                if stream:
+                    return await self._stream_response_bedrock(
+                        body,
+                        headers,
+                        "anthropic",
+                        model,
+                        request_id,
+                        original_tokens,
+                        optimized_tokens,
+                        tokens_saved,
+                        transforms_applied,
+                        tags,
+                        optimization_latency,
+                    )
+                else:
+                    backend_response = await self.anthropic_backend.send_message(body, headers)
+
+                    if backend_response.error:
+                        return JSONResponse(
+                            status_code=backend_response.status_code,
+                            content=backend_response.body,
+                        )
+
+                    # Track metrics
+                    total_latency = (time.time() - start_time) * 1000
+                    usage = backend_response.body.get("usage", {})
+                    output_tokens = usage.get("output_tokens", 0)
+
+                    await self.metrics.record_request(
+                        provider="bedrock",
+                        model=model,
+                        input_tokens=optimized_tokens,
+                        output_tokens=output_tokens,
+                        tokens_saved=tokens_saved,
+                        latency_ms=total_latency,
+                        cached=False,
+                    )
+
+                    if self.cost_tracker:
+                        cost_usd = self.cost_tracker.estimate_cost(
+                            model, optimized_tokens, output_tokens
+                        )
+                        if cost_usd:
+                            self.cost_tracker.record_cost(cost_usd)
+
+                    return JSONResponse(
+                        status_code=backend_response.status_code,
+                        content=backend_response.body,
+                    )
+            except Exception as e:
+                logger.error(f"[{request_id}] Bedrock backend error: {e}")
+                return JSONResponse(
+                    status_code=500,
+                    content={
+                        "type": "error",
+                        "error": {"type": "api_error", "message": str(e)},
+                    },
+                )
+
+        # Direct Anthropic API
         url = f"{self.ANTHROPIC_API_URL}/v1/messages"
 
         try:
@@ -3506,6 +3600,101 @@ class HeadroomProxy:
             media_type="text/event-stream",
         )
 
+    async def _stream_response_bedrock(
+        self,
+        body: dict,
+        headers: dict,
+        provider: str,
+        model: str,
+        request_id: str,
+        original_tokens: int,
+        optimized_tokens: int,
+        tokens_saved: int,
+        transforms_applied: list[str],
+        tags: dict[str, str],
+        optimization_latency: float,
+    ) -> StreamingResponse:
+        """Stream response from Bedrock backend with metrics tracking.
+
+        Translates Bedrock streaming events to Anthropic SSE format.
+        """
+        start_time = time.time()
+
+        # Mutable state for the generator
+        stream_state: dict[str, Any] = {
+            "input_tokens": 0,
+            "output_tokens": 0,
+        }
+
+        async def generate():
+            try:
+                assert self.anthropic_backend is not None
+
+                async for event in self.anthropic_backend.stream_message(body, headers):
+                    # Format as SSE
+                    if event.raw_sse:
+                        yield event.raw_sse.encode()
+                    else:
+                        sse_line = f"event: {event.event_type}\ndata: {json.dumps(event.data)}\n\n"
+                        yield sse_line.encode()
+
+                    # Track usage from message_start event
+                    if event.event_type == "message_start":
+                        msg = event.data.get("message", {})
+                        usage = msg.get("usage", {})
+                        if "input_tokens" in usage:
+                            stream_state["input_tokens"] = usage["input_tokens"]
+
+                    # Track output tokens from message_delta
+                    if event.event_type == "message_delta":
+                        usage = event.data.get("usage", {})
+                        if "output_tokens" in usage:
+                            stream_state["output_tokens"] = usage["output_tokens"]
+
+                    # Handle errors
+                    if event.event_type == "error":
+                        logger.error(f"[{request_id}] Bedrock stream error: {event.data}")
+
+            except Exception as e:
+                logger.error(f"[{request_id}] Bedrock streaming error: {e}")
+                error_event = {
+                    "type": "error",
+                    "error": {"type": "api_error", "message": str(e)},
+                }
+                yield f"event: error\ndata: {json.dumps(error_event)}\n\n".encode()
+
+            finally:
+                # Record metrics
+                total_latency = (time.time() - start_time) * 1000
+                output_tokens = stream_state["output_tokens"]
+
+                await self.metrics.record_request(
+                    provider="bedrock",
+                    model=model,
+                    input_tokens=optimized_tokens,
+                    output_tokens=output_tokens,
+                    tokens_saved=tokens_saved,
+                    latency_ms=total_latency,
+                    cached=False,
+                )
+
+                if self.cost_tracker:
+                    cost_usd = self.cost_tracker.estimate_cost(
+                        model, optimized_tokens, output_tokens
+                    )
+                    if cost_usd:
+                        self.cost_tracker.record_cost(cost_usd)
+
+                if tokens_saved > 0:
+                    logger.info(
+                        f"[{request_id}] Bedrock {model}: saved {tokens_saved:,} tokens (streaming)"
+                    )
+
+        return StreamingResponse(
+            generate(),
+            media_type="text/event-stream",
+        )
+
     async def handle_openai_chat(
         self,
         request: Request,
@@ -5948,6 +6137,17 @@ def run_server(
     pool_info = f"max={config.max_connections}, keepalive={config.max_keepalive_connections}"
     http2_status = "ENABLED" if config.http2 else "DISABLED"
 
+    # Backend status
+    if config.backend == "anthropic":
+        backend_status = "ANTHROPIC (direct API)"
+    else:
+        # Normalize: "bedrock" -> "litellm-bedrock"
+        backend = config.backend
+        if not backend.startswith("litellm-"):
+            backend = f"litellm-{backend}"
+        provider = backend.replace("litellm-", "")
+        backend_status = f"{provider.upper()} via LiteLLM (region={config.bedrock_region})"
+
     print(f"""
 ╔══════════════════════════════════════════════════════════════════════╗
 ║                      HEADROOM PROXY SERVER                           ║
@@ -5955,6 +6155,7 @@ def run_server(
 ║  Version: 1.0.0                                                      ║
 ║  Listening: http://{config.host}:{config.port:<5}                                      ║
 ║  Workers: {workers:<3}  Concurrency Limit: {limit_concurrency:<5}                          ║
+║  Backend: {backend_status:<59}║
 ╠══════════════════════════════════════════════════════════════════════╣
 ║  FEATURES:                                                           ║
 ║    Optimization:    {"ENABLED " if config.optimize else "DISABLED"}                                       ║
@@ -6045,6 +6246,23 @@ if __name__ == "__main__":
         "--openai-api-url", help=f"Custom OpenAI API URL (default: {HeadroomProxy.OPENAI_API_URL})"
     )
 
+    # Backend (anthropic direct or bedrock)
+    parser.add_argument(
+        "--backend",
+        choices=["anthropic", "bedrock"],
+        default="anthropic",
+        help="Backend for Anthropic API: 'anthropic' (direct) or 'bedrock' (AWS Bedrock)",
+    )
+    parser.add_argument(
+        "--bedrock-region",
+        default="us-west-2",
+        help="AWS region for Bedrock backend (default: us-west-2)",
+    )
+    parser.add_argument(
+        "--bedrock-profile",
+        help="AWS profile for Bedrock backend (default: use default credentials)",
+    )
+
     # Connection pool (scalability)
     parser.add_argument(
         "--max-connections",
@@ -6170,6 +6388,10 @@ if __name__ == "__main__":
         host=_get_env_str("HEADROOM_HOST", args.host),
         port=_get_env_int("HEADROOM_PORT", args.port),
         openai_api_url=_get_env_str("OPENAI_TARGET_API_URL", args.openai_api_url),
+        # Backend settings
+        backend=_get_env_str("HEADROOM_BACKEND", args.backend),  # type: ignore[arg-type]
+        bedrock_region=_get_env_str("HEADROOM_BEDROCK_REGION", args.bedrock_region),
+        bedrock_profile=args.bedrock_profile or os.environ.get("AWS_PROFILE"),
         optimize=optimize,
         min_tokens_to_crush=_get_env_int("HEADROOM_MIN_TOKENS", args.min_tokens),
         max_items_after_crush=_get_env_int("HEADROOM_MAX_ITEMS", args.max_items),