feat(translation): response and stream paths, async engine, and the flag-gated completion() seam

Phase 3 of the anthropic port (M3):

- ir.py gains the response IR (ChatResponse, ResponseUsage, finish reasons)
  and stream-event union; providers/anthropic parse_response/parse_stream
  reproduce v1 exactly for the v2-reachable surface (tool-name reverse
  mapping recomputed from the request, json_tool_call rewrite, usage cache
  folding, redacted thinking) and fail loudly on block types a v2-sent
  request cannot trigger
- inbound/openai_chat serialize_response/serialize_stream mirror v1's
  ModelResponse / CustomStreamWrapper shapes chunk-for-chunk (role on first
  chunk, tool-index counting, thinking provider fields, finish chunk)
- engine/: pipeline.py (prepare -> send split so the fallback decision
  happens before any I/O; async-first, sync is one asyncio.run wrapper at
  the seam), http.py (injected HttpPort + ExecuteError values), stream.py
  (the ONE accumulator: provider lines -> IR events -> chunk bodies)
- litellm/translation_seam.py (outside the package) adapts deps from litellm
  ambient state, ModelResponse/ModelResponseStream envelopes, and owns the
  completion() fork; litellm.translation_v2_providers allowlist global
  seeded from LITELLM_TRANSLATION_V2_PROVIDERS (off by default, yaml-
  configurable via the litellm_settings setattr fallback)
- main.py anthropic branch forks to the seam; streaming and modify_params
  traffic stay on v1 (documented follow-ups); once sent, provider errors
  raise the v1 exception contract, never a silent re-send
- differential gates: responses (4 full-cycle shapes incl. sanitized tool
  names and json_tool_call), streams (3 SSE replays vs the real
  CustomStreamWrapper), seam tests over respx (serve, fallback, 429
  contract, async path); DIFFERENTIAL_REPORT.md committed as the merge
  artifact: 53 rows, 0 divergent

litellm/__init__.py

@@ -234,6 +234,13 @@ use_chat_completions_url_for_anthropic_messages: bool = bool(
 route_all_chat_openai_to_responses: bool = (
     os.getenv("LITELLM_ROUTE_ALL_CHAT_OPENAI_TO_RESPONSES", "false").lower() == "true"
 )  # When True, routes all OpenAI /chat/completions requests through the Responses API bridge
+translation_v2_providers: List[str] = [
+    entry.strip()
+    for entry in os.getenv("LITELLM_TRANSLATION_V2_PROVIDERS", "").split(",")
+    if entry.strip()
+]  # Per-provider opt-in allowlist for the translation v2 rewrite (litellm/translation).
+# Off by default; e.g. "anthropic" or "anthropic,bedrock_converse". Any request whose
+# shape is outside v2's proven surface transparently falls back to v1.
 # When True, Gemini/Vertex Live setup is deferred until client `session.update`.
 # Default False preserves historical behavior (auto-send setup on connect).
 gemini_live_defer_setup: bool = (
@@ -1356,16 +1363,6 @@ from .interactions.agents.main import (
     alist_versions as alist_agent_versions,
     list_versions as list_agent_versions,
 )
-from .skills.main import (
-    create_skill,
-    acreate_skill,
-    list_skills,
-    alist_skills,
-    get_skill,
-    aget_skill,
-    delete_skill,
-    adelete_skill,
-)
 from .containers.main import *
 from .ocr.main import *
 from .rag.main import *

litellm/main.py

@@ -2949,6 +2949,26 @@ def completion(  # type: ignore # noqa: PLR0915
                     "LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX is set, skipping /v1/messages suffix"
                 )
 
+            if litellm.translation_v2_providers:
+                from litellm import translation_seam
+
+                _v2_response = translation_seam.try_completion_v2(
+                    model=model,
+                    messages=messages,
+                    optional_param_args=optional_param_args,
+                    non_default_params=non_default_params,
+                    api_key=api_key,
+                    api_base=api_base,
+                    timeout=timeout,
+                    stream=stream,
+                    acompletion=acompletion,
+                    logging_obj=logging,
+                    model_response=model_response,
+                    request_drop_params=kwargs.get("drop_params"),
+                )
+                if _v2_response is not None:
+                    return _v2_response
+
             response = anthropic_chat_completions.completion(
                 model=model,
                 messages=messages,

litellm/translation/deps.py

@@ -9,8 +9,8 @@ free of upward imports.
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Callable, Optional
 
 
 @dataclass(frozen=True)
@@ -34,9 +34,10 @@ class TranslationDeps:
       the contract.
     """
 
-    max_tokens_for_model: Callable[[str], Optional[int]]
+    max_tokens_for_model: Callable[[str], int | None]
     supports_capability: Callable[[str, str], bool]
-    capability_flag: Callable[[str, str], Optional[bool]]
+    capability_flag: Callable[[str, str], bool | None]
+    count_response_tokens: Callable[[str], int]
     drop_params: bool
     drop_params_global: bool
     modify_params: bool

litellm/translation/engine/http.py

@@ -0,0 +1,78 @@
+"""The injected HTTP port: the only I/O boundary in the package.
+
+The package never creates a client; the seam passes an object satisfying
+``HttpPort`` (functional core, imperative shell). A non-2xx response is a
+value (``ProviderHttpError``) the seam converts into the provider exception
+contract, never an exception inside the package.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Literal, Protocol
+
+from expression import case, tag, tagged_union
+from typing_extensions import assert_never
+
+from ..errors import TranslationError
+from ..ir import PlainJson
+
+
+@dataclass(frozen=True)
+class Endpoint:
+    """Where and how to reach the provider; built by the seam (URL suffixing,
+    auth headers, beta headers are v1-side concerns)."""
+
+    url: str
+    headers: Mapping[str, str]
+    timeout_seconds: float
+
+
+@dataclass(frozen=True)
+class HttpResponse:
+    status_code: int
+    body: PlainJson
+    text: str
+    headers: Mapping[str, str]
+
+
+class HttpPort(Protocol):
+    async def post_json(
+        self, endpoint: Endpoint, body: Mapping[str, PlainJson]
+    ) -> HttpResponse: ...
+
+
+@dataclass(frozen=True)
+class ProviderHttpError:
+    """A non-2xx provider response; carries everything the seam needs to
+    raise the same exception v1 would."""
+
+    status_code: int
+    text: str
+    headers: Mapping[str, str]
+
+
+@tagged_union(frozen=True)
+class ExecuteError:
+    tag: Literal["translation", "provider_http"] = tag()
+
+    translation: TranslationError = case()
+    provider_http: ProviderHttpError = case()
+
+    @staticmethod
+    def of_translation(value: TranslationError) -> ExecuteError:
+        return ExecuteError(translation=value)
+
+    @staticmethod
+    def of_provider_http(value: ProviderHttpError) -> ExecuteError:
+        return ExecuteError(provider_http=value)
+
+    @property
+    def summary(self) -> str:
+        match self.tag:
+            case "translation":
+                return self.translation.summary
+            case "provider_http":
+                return f"provider returned HTTP {self.provider_http.status_code}"
+        assert_never(self.tag)

litellm/translation/engine/pipeline.py

@@ -1,26 +1,36 @@
-"""Composition and the public translate entry point.
+"""Async-first composition and the public translation entry points.
 
-``translate_chat_request`` composes the inbound parse with the provider
-serialize under injected deps; the whole pipeline returns one ``Result`` and
-never raises. Any error means "outside v2's proven surface" and the dispatch
-seam falls back to v1, so no request ever loses a feature silently.
+``translate_chat_request`` is the pure request transform (parse -> serialize)
+the differential gate runs. ``execute_chat_request`` is the full async
+pipeline: translate, send through the injected HTTP port (popping the
+``json_mode`` transform-seam marker exactly like v1's HTTP handler), parse
+the provider response, and serialize the outbound body. Everything returns
+one ``Result`` and never raises; sync callers get the one wrapper the seam
+provides (v1's completion() already runs on an executor thread).
 """
 
 from __future__ import annotations
 
 from collections.abc import Callable, Mapping
+from dataclasses import dataclass
 from types import MappingProxyType
 
-from expression import Error, Result
+from expression import Error, Ok, Result
 
 from ..deps import TranslationDeps
 from ..dispatch import Provider
 from ..errors import TranslateResult, TranslationError
 from ..inbound.openai_chat import parse_request
-from ..ir import Body, ChatRequest
+from ..inbound.openai_chat.response import serialize_response
+from ..ir import Body, ChatRequest, ChatResponse, PlainJson
 from ..providers.anthropic import serialize_request
+from ..providers.anthropic.response import parse_response
+from .http import Endpoint, ExecuteError, HttpPort, ProviderHttpError
 
 _Serializer = Callable[[ChatRequest, TranslationDeps], Result[Body, TranslationError]]
+_ResponseParser = Callable[
+    [PlainJson, ChatRequest], Result[ChatResponse, TranslationError]
+]
 
 _SERIALIZERS: Mapping[Provider, _Serializer] = MappingProxyType(
     {
@@ -28,6 +38,12 @@ _SERIALIZERS: Mapping[Provider, _Serializer] = MappingProxyType(
     }
 )
 
+_RESPONSE_PARSERS: Mapping[Provider, _ResponseParser] = MappingProxyType(
+    {
+        "anthropic": parse_response,
+    }
+)
+
 
 def translate_chat_request(
     raw: Mapping[str, object], provider: Provider, deps: TranslationDeps
@@ -40,3 +56,105 @@ def translate_chat_request(
             )
         )
     return parse_request(raw).bind(lambda request: serializer(request, deps))
+
+
+def translate_chat_response(
+    raw_response: PlainJson,
+    request: ChatRequest,
+    provider: Provider,
+    deps: TranslationDeps,
+) -> TranslateResult:
+    parser = _RESPONSE_PARSERS.get(provider)
+    if parser is None:
+        return Error(
+            TranslationError.of_unsupported(
+                f"provider {provider!r} has no v2 response parser yet"
+            )
+        )
+    return parser(raw_response, request).map(
+        lambda response: serialize_response(response, deps)
+    )
+
+
+@dataclass(frozen=True)
+class PreparedRequest:
+    """A request that passed the fail-closed translation; from here on the
+    seam is committed to v2 (an HTTP or response failure surfaces as the
+    provider error contract, never a silent re-send through v1)."""
+
+    request: ChatRequest
+    body: Body
+
+
+def prepare_chat_request(
+    raw: Mapping[str, object], provider: Provider, deps: TranslationDeps
+) -> Result[PreparedRequest, TranslationError]:
+    serializer = _SERIALIZERS.get(provider)
+    if serializer is None or provider not in _RESPONSE_PARSERS:
+        return Error(
+            TranslationError.of_unsupported(
+                f"provider {provider!r} is not fully ported to v2 yet"
+            )
+        )
+    match parse_request(raw):
+        case Result(tag="ok", ok=request):
+            pass
+        case Result(error=parse_err):
+            return Error(parse_err)
+    return serializer(request, deps).map(
+        lambda body: PreparedRequest(request=request, body=body)
+    )
+
+
+def wire_body(prepared: PreparedRequest) -> Body:
+    """v1's HTTP handler pops json_mode from optional_params before the wire;
+    the marker exists only at the transform seam."""
+    return {key: value for key, value in prepared.body.items() if key != "json_mode"}
+
+
+async def send_prepared(
+    prepared: PreparedRequest,
+    provider: Provider,
+    deps: TranslationDeps,
+    http: HttpPort,
+    endpoint: Endpoint,
+) -> Result[Body, ExecuteError]:
+    parser = _RESPONSE_PARSERS.get(provider)
+    if parser is None:
+        return Error(
+            ExecuteError.of_translation(
+                TranslationError.of_unsupported(
+                    f"provider {provider!r} has no v2 response parser yet"
+                )
+            )
+        )
+    response = await http.post_json(endpoint, wire_body(prepared))
+    if response.status_code < 200 or response.status_code >= 300:
+        return Error(
+            ExecuteError.of_provider_http(
+                ProviderHttpError(
+                    status_code=response.status_code,
+                    text=response.text,
+                    headers=response.headers,
+                )
+            )
+        )
+    match parser(response.body, prepared.request):
+        case Result(tag="ok", ok=chat_response):
+            return Ok(serialize_response(chat_response, deps))
+        case Result(error=response_err):
+            return Error(ExecuteError.of_translation(response_err))
+
+
+async def execute_chat_request(
+    raw: Mapping[str, object],
+    provider: Provider,
+    deps: TranslationDeps,
+    http: HttpPort,
+    endpoint: Endpoint,
+) -> Result[Body, ExecuteError]:
+    match prepare_chat_request(raw, provider, deps):
+        case Result(tag="ok", ok=prepared):
+            return await send_prepared(prepared, provider, deps, http, endpoint)
+        case Result(error=err):
+            return Error(ExecuteError.of_translation(err))

litellm/translation/engine/stream.py

@@ -0,0 +1,57 @@
+"""The ONE stream accumulator: provider wire lines -> IR events -> chunks.
+
+Async-first: ``chunk_stream`` is the production shape (an async line stream
+in, OpenAI chunk bodies out); ``fold_lines`` is the synchronous fold over a
+recorded stream that the differential tests replay. Both compose an injected
+provider line parser with the inbound chunk fold, so adding a provider or an
+inbound schema never touches this file.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable, Iterable
+
+from expression import Error, Ok, Result
+from expression.collections import Block
+
+from ..errors import TranslationError
+from ..inbound.openai_chat.stream import StreamState, initial_state, step
+from ..ir import Body, StreamEvent
+
+ParseLine = Callable[[str], Result[StreamEvent | None, TranslationError]]
+
+
+def fold_lines(
+    lines: Iterable[str], parse_line: ParseLine
+) -> Result[Block[Body], TranslationError]:
+    state: StreamState = initial_state()
+    chunks: list[Body] = []
+    for line in lines:
+        match parse_line(line):
+            case Result(tag="ok", ok=event):
+                if event is None:
+                    continue
+                state, emitted = step(state, event)
+                chunks.extend(emitted)  # nosemgrep: translation-no-mutation
+            case Result(error=err):
+                return Error(err)
+    return Ok(Block.of_seq(chunks))
+
+
+async def chunk_stream(
+    lines: AsyncIterator[str], parse_line: ParseLine
+) -> AsyncIterator[Result[Body, TranslationError]]:
+    """Async-first form: yields one Result per outbound chunk; the first
+    error ends the stream (the seam surfaces it as a provider error)."""
+    state: StreamState = initial_state()
+    async for line in lines:
+        match parse_line(line):
+            case Result(tag="ok", ok=event):
+                if event is None:
+                    continue
+                state, emitted = step(state, event)
+                for chunk in emitted:
+                    yield Ok(chunk)
+            case Result(error=err):
+                yield Error(err)
+                return

litellm/translation/inbound/openai_chat/response.py

@@ -0,0 +1,143 @@
+"""IR ``ChatResponse`` -> OpenAI chat-completion response body.
+
+Emits the plain dict the seam feeds into ``ModelResponse`` (which owns the
+ambient envelope: chatcmpl id, created timestamp). Field shapes mirror what
+v1's ``transform_parsed_response`` builds, including the usage detail
+wrappers and the always-present ``provider_specific_fields`` keys.
+"""
+
+from __future__ import annotations
+
+import json
+
+from expression import Option
+from expression.collections import Block
+
+from ...deps import TranslationDeps
+from ...ir import Body, ChatResponse, ContentBlock, PlainJson, ResponseUsage
+
+
+def serialize_response(response: ChatResponse, deps: TranslationDeps) -> Body:
+    text = "".join(block.text.text for block in response.content if block.tag == "text")
+    tool_calls = _tool_calls(response.content)
+    thinking_blocks = _thinking_blocks(response.content)
+    reasoning: str | None = None
+    if thinking_blocks is not None:
+        reasoning = "".join(
+            block.thinking.thinking
+            for block in response.content
+            if block.tag == "thinking"
+        )
+    message: dict[str, PlainJson]
+    if response.synthesized_json_content:
+        # v1's json-mode replacement is a bare Message(content=...): no
+        # provider fields, no reasoning, no thinking blocks.
+        message = {"role": "assistant", "content": text or None}
+    else:
+        message = {
+            "role": "assistant",
+            "content": text or None,
+            "tool_calls": tool_calls,
+            "reasoning_content": reasoning,
+            "thinking_blocks": thinking_blocks,
+            "provider_specific_fields": {
+                "citations": None,
+                "thinking_blocks": thinking_blocks,
+            },
+        }
+    return {
+        "object": "chat.completion",
+        "model": response.model,
+        "choices": [
+            {
+                "index": 0,
+                "finish_reason": response.finish,
+                "message": message,
+            }
+        ],
+        "usage": _usage_json(response.usage, reasoning, deps),
+    }
+
+
+def _tool_calls(content: Block[ContentBlock]) -> PlainJson:
+    calls: list[PlainJson] = [
+        {
+            "id": block.tool_use.id,
+            "type": "function",
+            "function": {
+                "name": block.tool_use.name,
+                "arguments": json.dumps(block.tool_use.arguments.value),
+            },
+            "index": index,
+        }
+        for index, block in enumerate(content)
+        if block.tag == "tool_use"
+    ]
+    return calls or None
+
+
+def _thinking_blocks(content: Block[ContentBlock]) -> PlainJson:
+    blocks = [
+        _thinking_block_json(block)
+        for block in content
+        if block.tag in ("thinking", "redacted_thinking")
+    ]
+    return blocks or None
+
+
+def _thinking_block_json(block: ContentBlock) -> PlainJson:
+    if block.tag == "redacted_thinking":
+        return {"type": "redacted_thinking", "data": block.redacted_thinking.data}
+    thinking = block.thinking
+    base: dict[str, PlainJson] = {"type": "thinking", "thinking": thinking.thinking}
+    match thinking.signature:
+        case Option(tag="some", some=signature):
+            return {**base, "signature": signature}
+        case _:
+            return base
+
+
+def _usage_json(
+    usage: ResponseUsage, reasoning: str | None, deps: TranslationDeps
+) -> PlainJson:
+    """v1 ``calculate_usage``: cache tokens fold into prompt_tokens, the raw
+    input count lands in details.text_tokens, and reasoning tokens are
+    estimated by the injected token counter, capped at completion tokens."""
+    prompt_tokens = (
+        usage.input_tokens
+        + usage.cache_creation_input_tokens
+        + usage.cache_read_input_tokens
+    )
+    completion_tokens = usage.output_tokens
+    estimated = deps.count_response_tokens(reasoning) if reasoning else 0
+    reasoning_tokens = min(estimated, completion_tokens)
+    creation_details: PlainJson = None
+    match usage.cache_creation:
+        case Option(tag="some", some=details):
+            creation_details = {
+                "ephemeral_5m_input_tokens": details.five_minute.default_value(None),
+                "ephemeral_1h_input_tokens": details.one_hour.default_value(None),
+            }
+        case _:
+            creation_details = None
+    return {
+        "prompt_tokens": prompt_tokens,
+        "completion_tokens": completion_tokens,
+        "total_tokens": prompt_tokens + completion_tokens,
+        "prompt_tokens_details": {
+            "cached_tokens": usage.cache_read_input_tokens,
+            "cache_creation_tokens": usage.cache_creation_input_tokens,
+            "cache_creation_token_details": creation_details,
+            "text_tokens": usage.input_tokens,
+        },
+        "completion_tokens_details": {
+            "reasoning_tokens": max(0, reasoning_tokens),
+            "text_tokens": (
+                completion_tokens - reasoning_tokens
+                if reasoning_tokens > 0
+                else completion_tokens
+            ),
+        },
+        "cache_creation_input_tokens": usage.cache_creation_input_tokens,
+        "cache_read_input_tokens": usage.cache_read_input_tokens,
+    }

litellm/translation/inbound/openai_chat/stream.py

@@ -0,0 +1,150 @@
+"""IR stream events -> OpenAI chat-completion chunk bodies.
+
+A pure fold: ``step(state, event)`` returns the next state plus zero or more
+chunk bodies. The shapes mirror what v1's CustomStreamWrapper emits for the
+anthropic iterator: the first content-bearing chunk carries
+``role: "assistant"``, tool chunks carry ``content: ""`` beside the tool
+delta, tool indices count tool blocks (not content blocks), thinking deltas
+ride ``reasoning_content`` + ``thinking_blocks`` + provider fields, and the
+finish chunk is an empty delta with the mapped finish reason. The seam wraps
+each body in ``ModelResponseStream`` (ambient id/created).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from typing_extensions import assert_never
+
+from ...ir import Body, PlainJson, StreamEvent
+
+
+@dataclass(frozen=True)
+class StreamState:
+    model: str
+    sent_role: bool
+    tool_index: int
+
+
+def initial_state() -> StreamState:
+    return StreamState(model="", sent_role=False, tool_index=-1)
+
+
+_StepResult = tuple[StreamState, tuple[Body, ...]]
+
+
+def step(state: StreamState, event: StreamEvent) -> _StepResult:
+    match event.tag:
+        case "start":
+            return (
+                StreamState(
+                    model=event.start.model,
+                    sent_role=state.sent_role,
+                    tool_index=state.tool_index,
+                ),
+                (),
+            )
+        case "text_delta":
+            return _emit(state, {"content": event.text_delta.text})
+        case "tool_use_start":
+            started = StreamState(
+                model=state.model,
+                sent_role=state.sent_role,
+                tool_index=state.tool_index + 1,
+            )
+            return _emit(
+                started,
+                {
+                    "content": "",
+                    "tool_calls": [
+                        {
+                            "id": event.tool_use_start.id,
+                            "type": "function",
+                            "function": {
+                                "name": event.tool_use_start.name,
+                                "arguments": "",
+                            },
+                            "index": started.tool_index,
+                        }
+                    ],
+                },
+            )
+        case "tool_args_delta":
+            return _emit(
+                state,
+                {
+                    "content": "",
+                    "tool_calls": [
+                        {
+                            "id": None,
+                            "type": "function",
+                            "function": {
+                                "name": None,
+                                "arguments": event.tool_args_delta.partial_json,
+                            },
+                            "index": max(state.tool_index, 0),
+                        }
+                    ],
+                },
+            )
+        case "thinking_delta":
+            return _emit(
+                state,
+                _thinking_delta_body(event.thinking_delta.thinking, signature=""),
+            )
+        case "signature_delta":
+            return _emit(
+                state,
+                _thinking_delta_body("", signature=event.signature_delta.signature),
+            )
+        case "finish":
+            chunk: Body = {
+                "model": state.model,
+                "object": "chat.completion.chunk",
+                "choices": [
+                    {
+                        "index": 0,
+                        "delta": {},
+                        "finish_reason": event.finish.finish,
+                    }
+                ],
+            }
+            return state, (chunk,)
+        case "stop":
+            return state, ()
+    assert_never(event.tag)
+
+
+def _thinking_delta_body(thinking: str, signature: str) -> dict[str, PlainJson]:
+    block: PlainJson = {
+        "type": "thinking",
+        "thinking": thinking,
+        "signature": signature,
+    }
+    return {
+        "content": "",
+        "reasoning_content": thinking,
+        "thinking_blocks": [block],
+        "provider_specific_fields": {"thinking_blocks": [block]},
+    }
+
+
+def _emit(state: StreamState, delta: dict[str, PlainJson]) -> _StepResult:
+    role: str | None = None if state.sent_role else "assistant"
+    # v1's wrapper always sets provider_specific_fields (None when absent) on
+    # content-bearing deltas; the finish chunk's empty delta never has it.
+    body: Body = {
+        "model": state.model,
+        "object": "chat.completion.chunk",
+        "choices": [
+            {
+                "index": 0,
+                "delta": {"role": role, "provider_specific_fields": None, **delta},
+                "finish_reason": None,
+            }
+        ],
+    }
+    next_state = StreamState(
+        model=state.model, sent_role=True, tool_index=state.tool_index
+    )
+    return next_state, (body,)

litellm/translation/ir.py

@@ -357,3 +357,161 @@ def has_tool_blocks(messages: Block[Message]) -> bool:
         for message in messages
         for block in message.content
     )
+
+
+# --------------------------------------------------------------------------
+# Response IR: what a provider's parse_response yields and an inbound
+# serialize_response consumes. Content reuses ContentBlock (text, tool_use,
+# thinking, redacted_thinking are the cases a v2-sent request can produce;
+# anything else fails parse loudly because the fail-closed request surface
+# cannot trigger it).
+# --------------------------------------------------------------------------
+
+FinishReason = Literal["stop", "length", "tool_calls", "content_filter"]
+
+
+@dataclass(frozen=True)
+class CacheCreationDetails:
+    five_minute: Option[int]
+    one_hour: Option[int]
+
+
+@dataclass(frozen=True)
+class ResponseUsage:
+    """Provider-reported token counts, provider-neutral."""
+
+    input_tokens: int
+    output_tokens: int
+    cache_creation_input_tokens: int
+    cache_read_input_tokens: int
+    cache_creation: Option[CacheCreationDetails]
+
+
+@dataclass(frozen=True)
+class ChatResponse:
+    id: str
+    model: str
+    content: Block[ContentBlock]
+    finish: FinishReason
+    usage: ResponseUsage
+    synthesized_json_content: bool
+    """True when the provider rewrote a forced json_tool_call into plain
+    content (v1 then emits a bare message: no provider fields, no thinking)."""
+
+
+# --------------------------------------------------------------------------
+# Stream IR: provider stream parsers map wire events onto these; the inbound
+# stream serializer folds them into outbound chunks. One event per provider
+# wire event that carries information; keep-alives map to nothing.
+# --------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class StreamStart:
+    id: str
+    model: str
+    usage: ResponseUsage
+
+
+@dataclass(frozen=True)
+class TextDelta:
+    index: int
+    text: str
+
+
+@dataclass(frozen=True)
+class ToolUseStart:
+    index: int
+    id: str
+    name: str
+
+
+@dataclass(frozen=True)
+class ToolArgsDelta:
+    index: int
+    partial_json: str
+
+
+@dataclass(frozen=True)
+class ThinkingDelta:
+    index: int
+    thinking: str
+
+
+@dataclass(frozen=True)
+class SignatureDelta:
+    index: int
+    signature: str
+
+
+@dataclass(frozen=True)
+class StreamFinish:
+    """Anthropic message_delta: the final stop reason plus output usage."""
+
+    finish: FinishReason
+    output_tokens: int
+
+
+@tagged_union(frozen=True)
+class StreamEvent:
+    tag: Literal[
+        "start",
+        "text_delta",
+        "tool_use_start",
+        "tool_args_delta",
+        "thinking_delta",
+        "signature_delta",
+        "finish",
+        "stop",
+    ] = tag()
+
+    start: StreamStart = case()
+    text_delta: TextDelta = case()
+    tool_use_start: ToolUseStart = case()
+    tool_args_delta: ToolArgsDelta = case()
+    thinking_delta: ThinkingDelta = case()
+    signature_delta: SignatureDelta = case()
+    finish: StreamFinish = case()
+    stop: Unit = case()
+
+    @staticmethod
+    def of_start(value: StreamStart) -> StreamEvent:
+        return _stream_start(value)
+
+    @staticmethod
+    def of_text_delta(value: TextDelta) -> StreamEvent:
+        return _stream_text_delta(value)
+
+    @staticmethod
+    def of_tool_use_start(value: ToolUseStart) -> StreamEvent:
+        return _stream_tool_use_start(value)
+
+    @staticmethod
+    def of_tool_args_delta(value: ToolArgsDelta) -> StreamEvent:
+        return _stream_tool_args_delta(value)
+
+    @staticmethod
+    def of_thinking_delta(value: ThinkingDelta) -> StreamEvent:
+        return _stream_thinking_delta(value)
+
+    @staticmethod
+    def of_signature_delta(value: SignatureDelta) -> StreamEvent:
+        return _stream_signature_delta(value)
+
+    @staticmethod
+    def of_finish(value: StreamFinish) -> StreamEvent:
+        return _stream_finish(value)
+
+    @staticmethod
+    def of_stop() -> StreamEvent:
+        return _stream_stop(UNIT)
+
+
+_stream_start = _case_maker(StreamEvent, "start")
+_stream_text_delta = _case_maker(StreamEvent, "text_delta")
+_stream_tool_use_start = _case_maker(StreamEvent, "tool_use_start")
+_stream_tool_args_delta = _case_maker(StreamEvent, "tool_args_delta")
+_stream_thinking_delta = _case_maker(StreamEvent, "thinking_delta")
+_stream_signature_delta = _case_maker(StreamEvent, "signature_delta")
+_stream_finish = _case_maker(StreamEvent, "finish")
+_stream_stop = _case_maker(StreamEvent, "stop")

litellm/translation/providers/anthropic/response.py

@@ -0,0 +1,224 @@
+"""Anthropic ``/v1/messages`` response JSON -> IR ``ChatResponse``.
+
+The envelope is lenient (unknown metadata keys are ignored, exactly like v1)
+but content blocks are fail-closed: a block type the v2 request surface
+cannot trigger (server tools, citations, compaction) is a loud error value,
+never silently dropped. The json_tool_call rewrite (structured outputs on the
+json-tool strategy) happens here because the tool is an anthropic-side
+artifact the caller never asked for by name.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from types import MappingProxyType
+
+from expression import Error, Nothing, Ok, Option, Result, Some
+from expression.collections import Block
+
+from ...errors import BoundaryError, TranslationError
+from ...ir import (
+    CacheCreationDetails,
+    ChatRequest,
+    ChatResponse,
+    ContentBlock,
+    FinishReason,
+    JsonBlob,
+    PlainJson,
+    RedactedThinking,
+    ResponseUsage,
+    Text,
+    Thinking,
+    ToolUse,
+)
+from . import params as p
+from .tools import request_name_maps
+
+_ParseResult = Result[ChatResponse, TranslationError]
+
+# v1 map_finish_reason, anthropic rows; unknown reasons default to "stop"
+# with a warning in v1, so the same default applies here.
+FINISH_MAP: Mapping[str, FinishReason] = MappingProxyType(
+    {
+        "stop_sequence": "stop",
+        "end_turn": "stop",
+        "max_tokens": "length",
+        "tool_use": "tool_calls",
+        "refusal": "content_filter",
+        "compaction": "length",
+    }
+)
+
+
+def parse_response(raw: PlainJson, request: ChatRequest) -> _ParseResult:
+    if not isinstance(raw, dict):
+        return Error(_boundary("response body is not a JSON object"))
+    if "error" in raw:
+        return Error(_boundary(f"provider error payload: {raw['error']!r}"))
+    content = raw.get("content")
+    if not isinstance(content, list):
+        return Error(_boundary("response 'content' is not an array"))
+    blocks: list[ContentBlock] = []
+    for block in content:
+        parsed = _parse_block(block, request)
+        if isinstance(parsed, TranslationError):
+            return Error(parsed)
+        blocks.append(parsed)  # nosemgrep: translation-no-mutation
+    stop_reason = raw.get("stop_reason")
+    finish = (
+        FINISH_MAP.get(stop_reason, "stop") if isinstance(stop_reason, str) else "stop"
+    )
+    usage = parse_usage(raw.get("usage"))
+    if isinstance(usage, TranslationError):
+        return Error(usage)
+    model = raw.get("model")
+    response_id = raw.get("id")
+    response = ChatResponse(
+        id=response_id if isinstance(response_id, str) else "",
+        model=model if isinstance(model, str) else request.model,
+        content=Block.of_seq(blocks),
+        finish=finish,
+        usage=usage,
+        synthesized_json_content=False,
+    )
+    return Ok(_resolve_json_tool(response, request))
+
+
+def _boundary(reason: str) -> TranslationError:
+    return TranslationError.of_boundary(BoundaryError.of(Block.of_seq([reason])))
+
+
+def _parse_block(
+    block: PlainJson, request: ChatRequest
+) -> ContentBlock | TranslationError:
+    if not isinstance(block, dict):
+        return _boundary("response content block is not an object")
+    if block.get("citations") is not None:
+        return TranslationError.of_unsupported(
+            "response citations blocks; unreachable for v2-sent requests"
+        )
+    kind = block.get("type")
+    if kind == "text":
+        text = block.get("text")
+        if not isinstance(text, str):
+            return _boundary("text block is missing 'text'")
+        return ContentBlock.of_text(Text(text=text, cache=Nothing))
+    if kind == "tool_use":
+        return _parse_tool_use(block, request)
+    if kind == "thinking":
+        thinking = block.get("thinking")
+        signature = block.get("signature")
+        return ContentBlock.of_thinking(
+            Thinking(
+                thinking=thinking if isinstance(thinking, str) else "",
+                signature=Some(signature) if isinstance(signature, str) else Nothing,
+                cache=Nothing,
+            )
+        )
+    if kind == "redacted_thinking":
+        data = block.get("data")
+        return ContentBlock.of_redacted_thinking(
+            RedactedThinking(data=data if isinstance(data, str) else "")
+        )
+    return TranslationError.of_unsupported(
+        f"response content block type {kind!r}; unreachable for v2-sent requests"
+    )
+
+
+def _parse_tool_use(
+    block: dict[str, PlainJson], request: ChatRequest
+) -> ContentBlock | TranslationError:
+    if "caller" in block:
+        return TranslationError.of_unsupported(
+            "programmatic tool calling (caller); unreachable for v2-sent requests"
+        )
+    identifier = block.get("id")
+    name = block.get("name")
+    if not isinstance(identifier, str) or not isinstance(name, str):
+        return _boundary("tool_use block is missing 'id'/'name'")
+    _, reverse = request_name_maps(request.tools)
+    return ContentBlock.of_tool_use(
+        ToolUse(
+            id=identifier,
+            name=reverse.get(name, name),
+            arguments=JsonBlob(value=block.get("input")),
+            cache=Nothing,
+        )
+    )
+
+
+def parse_usage(raw: PlainJson) -> ResponseUsage | TranslationError:
+    if not isinstance(raw, dict):
+        return _boundary("response 'usage' is not an object")
+    creation = raw.get("cache_creation")
+    details: Option[CacheCreationDetails] = Nothing
+    if isinstance(creation, dict):
+        five = creation.get("ephemeral_5m_input_tokens")
+        hour = creation.get("ephemeral_1h_input_tokens")
+        details = Some(
+            CacheCreationDetails(
+                five_minute=Some(five) if isinstance(five, int) else Nothing,
+                one_hour=Some(hour) if isinstance(hour, int) else Nothing,
+            )
+        )
+    return ResponseUsage(
+        input_tokens=_int_of(raw.get("input_tokens")),
+        output_tokens=_int_of(raw.get("output_tokens")),
+        cache_creation_input_tokens=_int_of(raw.get("cache_creation_input_tokens")),
+        cache_read_input_tokens=_int_of(raw.get("cache_read_input_tokens")),
+        cache_creation=details,
+    )
+
+
+def _int_of(value: PlainJson) -> int:
+    """v1 tolerates explicit nulls and non-numerics in usage counts."""
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, (int, float)):
+        return int(value)
+    return 0
+
+
+def _uses_json_tool(request: ChatRequest) -> bool:
+    match request.response_format:
+        case Option(tag="some", some=response_format):
+            return response_format.tag == "json_schema" and not p.uses_output_format(
+                request.model
+            )
+        case _:
+            return False
+
+
+def _resolve_json_tool(response: ChatResponse, request: ChatRequest) -> ChatResponse:
+    """v1 ``_resolve_json_mode_non_streaming``: when structured outputs ride
+    the json_tool_call strategy, the forced tool call comes back as plain
+    content and the stop reason is rewritten to a clean stop. The mixed
+    user-tools case is unreachable (the request side refuses it)."""
+    if not _uses_json_tool(request):
+        return response
+    tool_uses = [
+        block.tool_use for block in response.content if block.tag == "tool_use"
+    ]
+    json_calls = [tool for tool in tool_uses if tool.name == "json_tool_call"]
+    if not json_calls or len(json_calls) != len(tool_uses):
+        return response
+    arguments = json_calls[0].arguments.value
+    if arguments is None:
+        return response
+    payload: PlainJson = arguments
+    if isinstance(arguments, dict) and arguments.get("values") is not None:
+        payload = arguments["values"]
+    # v1 replaces the whole message with the JSON content (any text or
+    # thinking blocks in the raw response are discarded with it).
+    kept = Block.of_seq(
+        [ContentBlock.of_text(Text(text=json.dumps(payload), cache=Nothing))]
+    )
+    return ChatResponse(
+        id=response.id,
+        model=response.model,
+        content=kept,
+        finish="stop",
+        usage=response.usage,
+        synthesized_json_content=True,
+    )

litellm/translation/providers/anthropic/stream.py

@@ -0,0 +1,200 @@
+"""Anthropic SSE stream -> IR stream events.
+
+One SSE ``data:`` payload maps to at most one ``StreamEvent``; keep-alives
+and block-stop bookkeeping map to none. Event types the v2 request surface
+cannot trigger (server tools, citations) are loud error values. Tool names
+are reverse-mapped through the per-request map, mirroring the non-streaming
+path.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+
+from expression import Error, Ok, Result
+from expression.collections import Block
+
+from ...errors import BoundaryError, TranslationError
+from ...ir import (
+    ChatRequest,
+    FinishReason,
+    PlainJson,
+    SignatureDelta,
+    StreamEvent,
+    StreamFinish,
+    StreamStart,
+    TextDelta,
+    ThinkingDelta,
+    ToolArgsDelta,
+    ToolUseStart,
+)
+from .response import FINISH_MAP, parse_usage
+from .tools import request_name_maps
+
+_EventResult = Result[StreamEvent | None, TranslationError]
+
+
+def reverse_names(request: ChatRequest) -> Mapping[str, str]:
+    _, reverse = request_name_maps(request.tools)
+    return reverse
+
+
+def parse_sse_line(line: str, reverse: Mapping[str, str]) -> _EventResult:
+    """One raw SSE line -> at most one event. Non-data lines are framing."""
+    stripped = line.strip()
+    if not stripped.startswith("data:"):
+        return Ok(None)
+    payload = stripped[len("data:") :].strip()
+    try:
+        event: PlainJson = json.loads(payload)
+    except ValueError:
+        return Error(_boundary(f"stream payload is not JSON: {payload[:120]!r}"))
+    return parse_event(event, reverse)
+
+
+def parse_event(event: PlainJson, reverse: Mapping[str, str]) -> _EventResult:
+    if not isinstance(event, dict):
+        return Error(_boundary("stream event is not an object"))
+    kind = event.get("type")
+    if kind in ("ping", "content_block_stop"):
+        return Ok(None)
+    if kind == "message_start":
+        return _start_event(event)
+    if kind == "content_block_start":
+        return _block_start_event(event, reverse)
+    if kind == "content_block_delta":
+        return _delta_event(event)
+    if kind == "message_delta":
+        return _finish_event(event)
+    if kind == "message_stop":
+        return Ok(StreamEvent.of_stop())
+    if kind == "error":
+        return Error(_boundary(f"provider stream error: {event.get('error')!r}"))
+    return Error(
+        TranslationError.of_unsupported(
+            f"stream event type {kind!r}; unreachable for v2-sent requests"
+        )
+    )
+
+
+def _boundary(reason: str) -> TranslationError:
+    return TranslationError.of_boundary(BoundaryError.of(Block.of_seq([reason])))
+
+
+def _start_event(event: dict[str, PlainJson]) -> _EventResult:
+    message = event.get("message")
+    if not isinstance(message, dict):
+        return Error(_boundary("message_start is missing 'message'"))
+    usage = parse_usage(message.get("usage") or {})
+    if isinstance(usage, TranslationError):
+        return Error(usage)
+    identifier = message.get("id")
+    model = message.get("model")
+    return Ok(
+        StreamEvent.of_start(
+            StreamStart(
+                id=identifier if isinstance(identifier, str) else "",
+                model=model if isinstance(model, str) else "",
+                usage=usage,
+            )
+        )
+    )
+
+
+def _block_start_event(
+    event: dict[str, PlainJson], reverse: Mapping[str, str]
+) -> _EventResult:
+    block = event.get("content_block")
+    index = event.get("index")
+    if not isinstance(block, dict) or not isinstance(index, int):
+        return Error(_boundary("content_block_start is malformed"))
+    kind = block.get("type")
+    if kind == "text":
+        text = block.get("text")
+        if isinstance(text, str) and text:
+            return Ok(StreamEvent.of_text_delta(TextDelta(index=index, text=text)))
+        return Ok(None)
+    if kind == "thinking":
+        return Ok(None)  # thinking starts empty; deltas carry the content
+    if kind == "tool_use":
+        identifier = block.get("id")
+        name = block.get("name")
+        if not isinstance(identifier, str) or not isinstance(name, str):
+            return Error(_boundary("tool_use block start is missing 'id'/'name'"))
+        return Ok(
+            StreamEvent.of_tool_use_start(
+                ToolUseStart(index=index, id=identifier, name=reverse.get(name, name))
+            )
+        )
+    return Error(
+        TranslationError.of_unsupported(
+            f"stream content block type {kind!r}; unreachable for v2-sent requests"
+        )
+    )
+
+
+def _delta_event(event: dict[str, PlainJson]) -> _EventResult:
+    delta = event.get("delta")
+    index = event.get("index")
+    if not isinstance(delta, dict) or not isinstance(index, int):
+        return Error(_boundary("content_block_delta is malformed"))
+    kind = delta.get("type")
+    if kind == "text_delta":
+        text = delta.get("text")
+        return Ok(
+            StreamEvent.of_text_delta(
+                TextDelta(index=index, text=text if isinstance(text, str) else "")
+            )
+        )
+    if kind == "input_json_delta":
+        partial = delta.get("partial_json")
+        return Ok(
+            StreamEvent.of_tool_args_delta(
+                ToolArgsDelta(
+                    index=index,
+                    partial_json=partial if isinstance(partial, str) else "",
+                )
+            )
+        )
+    if kind == "thinking_delta":
+        thinking = delta.get("thinking")
+        return Ok(
+            StreamEvent.of_thinking_delta(
+                ThinkingDelta(
+                    index=index,
+                    thinking=thinking if isinstance(thinking, str) else "",
+                )
+            )
+        )
+    if kind == "signature_delta":
+        signature = delta.get("signature")
+        return Ok(
+            StreamEvent.of_signature_delta(
+                SignatureDelta(
+                    index=index,
+                    signature=signature if isinstance(signature, str) else "",
+                )
+            )
+        )
+    return Error(
+        TranslationError.of_unsupported(
+            f"stream delta type {kind!r}; unreachable for v2-sent requests"
+        )
+    )
+
+
+def _finish_event(event: dict[str, PlainJson]) -> _EventResult:
+    delta = event.get("delta")
+    usage = event.get("usage")
+    stop_reason = delta.get("stop_reason") if isinstance(delta, dict) else None
+    finish: FinishReason = (
+        FINISH_MAP.get(stop_reason, "stop") if isinstance(stop_reason, str) else "stop"
+    )
+    output_tokens = 0
+    if isinstance(usage, dict):
+        raw = usage.get("output_tokens")
+        output_tokens = int(raw) if isinstance(raw, (int, float)) else 0
+    return Ok(
+        StreamEvent.of_finish(StreamFinish(finish=finish, output_tokens=output_tokens))
+    )

litellm/translation_seam.py

@@ -0,0 +1,356 @@
+"""The v1-side adapter for translation v2 (lives OUTSIDE litellm/translation).
+
+This module is the one place that speaks both languages: it reads litellm
+ambient state (model map helpers, drop_params/modify_params, the per-provider
+allowlist flag) into the package's injected ``TranslationDeps`` value, adapts
+v2's plain response bodies onto ``ModelResponse``, and owns the
+``completion()`` fork. The translation package never imports the v1 stack;
+this seam imports both, in the allowed direction.
+
+Flag precedent (dossier section 7): ``litellm.translation_v2_providers`` is a
+module global seeded from the ``LITELLM_TRANSLATION_V2_PROVIDERS``
+comma-separated env var, and the proxy's ``litellm_settings`` generic setattr
+fallback makes it yaml-configurable with zero extra plumbing. Off by default.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, FrozenSet, Optional, cast
+
+import litellm
+from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+from litellm.types.utils import (
+    CacheCreationTokenDetails,
+    CompletionTokensDetailsWrapper,
+    ModelResponse,
+    PromptTokensDetailsWrapper,
+    Usage,
+)
+
+from litellm.translation import TranslationDeps
+from litellm.translation.ir import Body
+
+
+def enabled_providers() -> FrozenSet[str]:
+    """The per-provider opt-in allowlist, read at call time so proxy yaml and
+    runtime changes apply without restart."""
+    configured = getattr(litellm, "translation_v2_providers", None) or []
+    return frozenset(str(entry).strip() for entry in configured if str(entry).strip())
+
+
+def _max_tokens_for_model(model: str) -> Optional[int]:
+    try:
+        return litellm.utils.get_max_tokens(model)
+    except Exception:
+        return None
+
+
+def _count_response_tokens(text: str) -> int:
+    from litellm.utils import token_counter
+
+    return token_counter(text=text, count_response_tokens=True)
+
+
+def build_translation_deps(
+    request_drop_params: Optional[bool] = None,
+) -> TranslationDeps:
+    drop_params_global = litellm.drop_params is True
+    return TranslationDeps(
+        max_tokens_for_model=_max_tokens_for_model,
+        supports_capability=AnthropicModelInfo._supports_model_capability,
+        capability_flag=AnthropicModelInfo._get_model_capability,
+        count_response_tokens=_count_response_tokens,
+        drop_params=drop_params_global or request_drop_params is True,
+        drop_params_global=drop_params_global,
+        modify_params=litellm.modify_params is True,
+    )
+
+
+def to_model_response(
+    body: Body, model_response: Optional[ModelResponse] = None
+) -> ModelResponse:
+    """Adapt a v2 response body onto litellm's ModelResponse envelope.
+
+    Mirrors v1's ``transform_parsed_response`` assembly exactly (assign into
+    the pre-allocated response's first choice, stamp created/model, setattr a
+    real ``Usage``), so the serialized shape is identical to v1's. The
+    envelope (chatcmpl id, created timestamp) stays litellm-ambient.
+    """
+    import time
+
+    from litellm.types.utils import Message
+
+    response = model_response if model_response is not None else ModelResponse()
+    choices = body.get("choices")
+    first = choices[0] if isinstance(choices, list) and choices else {}
+    message_payload = first.get("message") if isinstance(first, dict) else {}
+    if isinstance(message_payload, dict):
+        response.choices[0].message = Message(**cast(Dict[str, Any], message_payload))
+    finish = first.get("finish_reason") if isinstance(first, dict) else None
+    response.choices[0].finish_reason = cast(
+        Any, finish if isinstance(finish, str) else "stop"
+    )
+    usage_payload = body.get("usage")
+    usage = _build_usage(usage_payload) if isinstance(usage_payload, dict) else Usage()
+    setattr(response, "usage", usage)
+    response.created = int(time.time())
+    model = body.get("model")
+    if isinstance(model, str):
+        response.model = model
+    return response
+
+
+def _build_usage(payload: dict) -> Usage:
+    """Construct Usage with the exact kwarg set v1's calculate_usage passes,
+    because nested (extra-attribute) serialization only includes explicitly
+    set fields; a kwargs-splat would drop the always-set None fields v1 emits
+    (server_tool_use, inference_geo, speed, wrapper audio/image/video)."""
+    prompt_details = payload.get("prompt_tokens_details") or {}
+    completion_details = payload.get("completion_tokens_details") or {}
+    creation_details = prompt_details.get("cache_creation_token_details")
+    wrapper_details = (
+        CacheCreationTokenDetails(
+            ephemeral_5m_input_tokens=creation_details.get("ephemeral_5m_input_tokens"),
+            ephemeral_1h_input_tokens=creation_details.get("ephemeral_1h_input_tokens"),
+        )
+        if isinstance(creation_details, dict)
+        else None
+    )
+    return Usage(
+        prompt_tokens=payload.get("prompt_tokens"),
+        completion_tokens=payload.get("completion_tokens"),
+        total_tokens=payload.get("total_tokens"),
+        prompt_tokens_details=PromptTokensDetailsWrapper(
+            cached_tokens=prompt_details.get("cached_tokens"),
+            cache_creation_tokens=prompt_details.get("cache_creation_tokens"),
+            cache_creation_token_details=wrapper_details,
+            text_tokens=prompt_details.get("text_tokens"),
+        ),
+        cache_creation_input_tokens=payload.get("cache_creation_input_tokens"),
+        cache_read_input_tokens=payload.get("cache_read_input_tokens"),
+        completion_tokens_details=CompletionTokensDetailsWrapper(
+            reasoning_tokens=completion_details.get("reasoning_tokens"),
+            text_tokens=completion_details.get("text_tokens"),
+        ),
+        server_tool_use=None,
+        inference_geo=None,
+        speed=None,
+    )
+
+
+def to_model_response_stream(body: Body, stream_id: str):
+    """Adapt one v2 chunk body onto ModelResponseStream; the stream id is
+    minted once per stream by the caller (v1 reuses one id for every chunk).
+    system_fingerprint/citations are set explicitly because v1's wrapper
+    always sets them and extra-field serialization only dumps set fields."""
+    from litellm.types.utils import ModelResponseStream
+
+    chunk = ModelResponseStream(
+        id=stream_id, system_fingerprint=None, **cast(Dict[str, Any], body)
+    )
+    choices = body.get("choices")
+    first = choices[0] if isinstance(choices, list) and choices else {}
+    if isinstance(first, dict) and first.get("finish_reason") is None:
+        # v1 sets citations only on content chunks, never the finish chunk.
+        setattr(chunk, "citations", None)
+    return chunk
+
+
+class HttpxJsonPort:
+    """HttpPort over a fresh httpx.AsyncClient per request.
+
+    A fresh client avoids binding litellm's cached async clients to the
+    short-lived event loop the sync wrapper creates. Pooled-client reuse on
+    the async path is a follow-up optimization; the flag-gated M3 traffic
+    does not need it.
+    """
+
+    async def post_json(self, endpoint, body):  # noqa: ANN001, ANN201
+        import httpx
+
+        from litellm.translation.engine.http import HttpResponse
+
+        async with httpx.AsyncClient(timeout=endpoint.timeout_seconds) as client:
+            raw = await client.post(
+                endpoint.url, headers=dict(endpoint.headers), json=dict(body)
+            )
+        try:
+            payload = raw.json()
+        except ValueError:
+            payload = None
+        return HttpResponse(
+            status_code=raw.status_code,
+            body=payload,
+            text=raw.text,
+            headers=dict(raw.headers),
+        )
+
+
+def try_completion_v2(
+    *,
+    model: str,
+    messages: list,
+    optional_param_args: dict,
+    non_default_params: dict,
+    api_key: Optional[str],
+    api_base: Optional[str],
+    timeout: Optional[float],
+    stream: Optional[bool],
+    acompletion: Optional[bool],
+    logging_obj,
+    model_response: ModelResponse,
+    request_drop_params: Optional[bool],
+):
+    """The completion() fork for translation v2 (anthropic, non-streaming).
+
+    Returns None to stay on v1: flag off, streaming (the v2 stream seam is a
+    follow-up), modify_params behaviors, or any request shape outside v2's
+    proven surface (the fail-closed translate). Once the request is sent,
+    failures raise the provider error contract exactly like v1; there is no
+    silent re-send.
+    """
+    from litellm.translation import route
+    from litellm.translation.engine.pipeline import prepare_chat_request
+
+    if "anthropic" not in enabled_providers():
+        return None
+    if stream is True or litellm.modify_params is True:
+        return None
+    decision = route(
+        schema="openai_chat",
+        provider="anthropic",
+        enabled_providers=frozenset({"anthropic"}),  # checked above
+        body_touching=False,
+    )
+    if decision.tag != "v2":
+        return None
+    raw_body = _raw_openai_body(
+        model, messages, optional_param_args, non_default_params
+    )
+    deps = build_translation_deps(request_drop_params=request_drop_params)
+    prepared_result = prepare_chat_request(raw_body, "anthropic", deps)
+    if prepared_result.is_error():
+        litellm.verbose_logger.debug(
+            "translation v2 fallback to v1: %s", prepared_result.error.summary
+        )
+        return None
+    prepared = prepared_result.ok
+    coroutine = _send_v2(
+        prepared=prepared,
+        deps=deps,
+        model=model,
+        messages=messages,
+        api_key=api_key,
+        api_base=api_base,
+        timeout=timeout,
+        logging_obj=logging_obj,
+        model_response=model_response,
+    )
+    if acompletion is True:
+        return coroutine
+    import asyncio
+
+    return asyncio.run(coroutine)
+
+
+_BODY_FIELDS = (
+    "temperature",
+    "top_p",
+    "max_tokens",
+    "max_completion_tokens",
+    "stop",
+    "stream",
+    "tools",
+    "tool_choice",
+    "parallel_tool_calls",
+    "response_format",
+    "user",
+    "reasoning_effort",
+    "thinking",
+)
+
+
+def _raw_openai_body(
+    model: str, messages: list, optional_param_args: dict, non_default_params: dict
+) -> dict:
+    """Rebuild the caller's OpenAI-shape body from completion()'s pre-mapping
+    locals. Provider-specific extras (non_default_params, e.g. top_k) ride
+    along verbatim: anything v2 does not account for becomes a typed
+    unsupported error and falls back to v1 -- the fail-closed allowlist."""
+    named = {
+        key: optional_param_args.get(key)
+        for key in _BODY_FIELDS
+        if optional_param_args.get(key) is not None
+    }
+    return {"model": model, "messages": messages, **named, **non_default_params}
+
+
+async def _send_v2(
+    *,
+    prepared,
+    deps: TranslationDeps,
+    model: str,
+    messages: list,
+    api_key: Optional[str],
+    api_base: Optional[str],
+    timeout: Optional[float],
+    logging_obj,
+    model_response: ModelResponse,
+) -> ModelResponse:
+    from litellm.llms.anthropic.common_utils import (
+        AnthropicError,
+        process_anthropic_headers,
+    )
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    from litellm.translation.engine.http import Endpoint
+    from litellm.translation.engine.pipeline import send_prepared, wire_body
+
+    config = AnthropicConfig()
+    headers = config.validate_environment(
+        headers={},
+        model=model,
+        messages=messages,
+        optional_params=dict(prepared.body),
+        litellm_params={},
+        api_key=api_key,
+        api_base=api_base,
+    )
+    endpoint = Endpoint(
+        url=api_base or "https://api.anthropic.com/v1/messages",
+        headers=headers,
+        timeout_seconds=float(timeout) if timeout else 600.0,
+    )
+    logging_obj.pre_call(
+        input=messages,
+        api_key=api_key,
+        additional_args={
+            "complete_input_dict": wire_body(prepared),
+            "api_base": endpoint.url,
+            "headers": dict(headers),
+        },
+    )
+    result = await send_prepared(prepared, "anthropic", deps, HttpxJsonPort(), endpoint)
+    if result.is_error():
+        error = result.error
+        if error.tag == "provider_http":
+            logging_obj.post_call(
+                input=messages,
+                api_key=api_key,
+                original_response=error.provider_http.text,
+            )
+            raise AnthropicError(
+                status_code=error.provider_http.status_code,
+                message=error.provider_http.text,
+            )
+        raise AnthropicError(status_code=500, message=error.summary)
+    body = result.ok
+    logging_obj.post_call(
+        input=messages,
+        api_key=api_key,
+        original_response=json.dumps(body, default=str),
+    )
+    response = to_model_response(body, model_response)
+    response._hidden_params["additional_headers"] = process_anthropic_headers({})
+    return response

tests/test_litellm/translation/DIFFERENTIAL_REPORT.md

@@ -0,0 +1,77 @@
+# Translation v2 differential report (anthropic)
+
+v1 and v2 run over the same corpus; every row must be IDENTICAL for
+the anthropic flag to turn on. Regenerate with:
+`python -m tests.test_litellm.translation.generate_differential_report`
+
+- commit: 23724c4392
+
+## Request bodies (v1 map_openai_params + transform_request vs v2)
+
+- IDENTICAL: assistant_text_and_tool_call
+- IDENTICAL: cache_control_everywhere
+- IDENTICAL: cache_control_on_string_message
+- IDENTICAL: cache_control_on_tool_result
+- IDENTICAL: current_model_default_max_tokens
+- IDENTICAL: duplicate_tool_call_ids_dedupe
+- IDENTICAL: empty_user_content_placeholder
+- IDENTICAL: final_assistant_text_rstripped
+- IDENTICAL: https_image_url
+- IDENTICAL: image_data_uri
+- IDENTICAL: image_format_override
+- IDENTICAL: max_completion_tokens
+- IDENTICAL: multiturn_stop_stream
+- IDENTICAL: no_max_tokens_legacy_model
+- IDENTICAL: parallel_tool_calls_false
+- IDENTICAL: parallel_tool_results_merge
+- IDENTICAL: parallel_with_string_none_choice
+- IDENTICAL: parallel_with_tool_choice
+- IDENTICAL: pydantic_style_schema_filtered
+- IDENTICAL: reasoning_effort_high_no_max_tokens
+- IDENTICAL: reasoning_effort_low
+- IDENTICAL: reasoning_effort_none
+- IDENTICAL: response_format_json_object_current
+- IDENTICAL: response_format_json_object_legacy_noop
+- IDENTICAL: response_format_json_schema_current
+- IDENTICAL: response_format_json_schema_legacy_tool
+- IDENTICAL: response_format_with_thinking_no_forced_choice
+- IDENTICAL: stop_as_string
+- IDENTICAL: stop_whitespace_kept_without_drop_params
+- IDENTICAL: system_and_sampling
+- IDENTICAL: system_as_array
+- IDENTICAL: temperature_int_stays_int
+- IDENTICAL: text
+- IDENTICAL: thinking_explicit
+- IDENTICAL: thinking_history_blocks
+- IDENTICAL: thinking_no_max_tokens_bumps
+- IDENTICAL: tool_call_roundtrip
+- IDENTICAL: tool_choice_dict_forms
+- IDENTICAL: tool_choice_none
+- IDENTICAL: tool_choice_required
+- IDENTICAL: tool_choice_specific
+- IDENTICAL: tool_name_sanitization_with_history
+- IDENTICAL: tool_result_parts_not_placeholdered
+- IDENTICAL: tool_schema_missing_parameters
+- IDENTICAL: tool_schema_type_coerced
+- IDENTICAL: tool_use_id_sanitized
+- IDENTICAL: tool_without_description
+- IDENTICAL: tools_auto
+- IDENTICAL: top_k
+- IDENTICAL: user_email_skipped
+- IDENTICAL: user_metadata
+- IDENTICAL: whitespace_text_part_placeholder
+
+## Responses (v1 transform_response vs v2)
+
+- IDENTICAL: json_tool
+- IDENTICAL: text
+- IDENTICAL: thinking
+- IDENTICAL: tools
+
+## Streams (v1 CustomStreamWrapper replay vs v2 engine/stream)
+
+- IDENTICAL: text
+- IDENTICAL: thinking
+- IDENTICAL: tools
+
+Result: 0 divergent rows. Shapes outside the corpus fall back to v1 (fail-closed), so this table is the complete flag-on surface.

tests/test_litellm/translation/conftest.py

@@ -13,7 +13,7 @@ import pytest
 os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True")
 
 from litellm.llms.anthropic.common_utils import AnthropicModelInfo  # noqa: E402
-from litellm.utils import get_max_tokens  # noqa: E402
+from litellm.utils import get_max_tokens, token_counter  # noqa: E402
 
 from litellm.translation import TranslationDeps  # noqa: E402
 
@@ -34,6 +34,9 @@ def build_real_deps(
         max_tokens_for_model=_max_tokens_for_model,
         supports_capability=AnthropicModelInfo._supports_model_capability,
         capability_flag=AnthropicModelInfo._get_model_capability,
+        count_response_tokens=lambda text: token_counter(
+            text=text, count_response_tokens=True
+        ),
         drop_params=drop_params,
         drop_params_global=drop_params_global,
         modify_params=modify_params,

tests/test_litellm/translation/generate_differential_report.py

@@ -0,0 +1,83 @@
+"""Regenerate DIFFERENTIAL_REPORT.md: the v1-vs-v2 parity merge artifact.
+
+Run:  python -m tests.test_litellm.translation.generate_differential_report
+"""
+
+import json
+import pathlib
+import subprocess
+import sys
+import time
+import uuid
+
+_HERE = pathlib.Path(__file__).parent
+
+
+def main() -> None:
+    from . import test_differential_anthropic_request as req
+    from . import test_differential_anthropic_response as resp
+    from . import test_differential_anthropic_stream as stream
+
+    counter = iter(range(1, 1_000_000))
+    uuid.uuid4 = lambda: uuid.UUID(int=next(counter))  # type: ignore[assignment]
+    time.time = lambda: resp.FROZEN_TIME  # type: ignore[assignment]
+
+    lines = [
+        "# Translation v2 differential report (anthropic)",
+        "",
+        "v1 and v2 run over the same corpus; every row must be IDENTICAL for",
+        "the anthropic flag to turn on. Regenerate with:",
+        "`python -m tests.test_litellm.translation.generate_differential_report`",
+        "",
+        f"- commit: {_git_sha()}",
+        "",
+        "## Request bodies (v1 map_openai_params + transform_request vs v2)",
+        "",
+    ]
+    failures = 0
+    for name in sorted(req.CORPUS):
+        same = req._norm(req._v2_body(req.CORPUS[name])) == req._norm(
+            req._v1_body(req.CORPUS[name])
+        )
+        failures += 0 if same else 1
+        lines.append(f"- {'IDENTICAL' if same else 'DIVERGENT'}: {name}")
+    lines += ["", "## Responses (v1 transform_response vs v2)", ""]
+    for name in sorted(resp._REQUESTS):
+        same = resp._norm(resp._v2_model_response(name)) == resp._norm(
+            resp._v1_model_response(name)
+        )
+        failures += 0 if same else 1
+        lines.append(f"- {'IDENTICAL' if same else 'DIVERGENT'}: {name}")
+    lines += ["", "## Streams (v1 CustomStreamWrapper replay vs v2 engine/stream)", ""]
+    for name in sorted(stream.STREAMS):
+        same = stream._norm(stream._v2_chunks(stream.STREAMS[name])) == stream._norm(
+            stream._v1_chunks(stream.STREAMS[name])
+        )
+        failures += 0 if same else 1
+        lines.append(f"- {'IDENTICAL' if same else 'DIVERGENT'}: {name}")
+    lines += [
+        "",
+        f"Result: {failures} divergent rows."
+        " Shapes outside the corpus fall back to v1 (fail-closed), so this"
+        " table is the complete flag-on surface.",
+        "",
+    ]
+    (_HERE / "DIFFERENTIAL_REPORT.md").write_text("\n".join(lines))
+    print("\n".join(lines))
+    sys.exit(1 if failures else 0)
+
+
+def _git_sha() -> str:
+    try:
+        return subprocess.run(
+            ["git", "rev-parse", "--short", "HEAD"],
+            capture_output=True,
+            text=True,
+            check=True,
+        ).stdout.strip()
+    except Exception:
+        return "unknown"
+
+
+if __name__ == "__main__":
+    main()

tests/test_litellm/translation/test_differential_anthropic_response.py

@@ -0,0 +1,238 @@
+"""Differential parity for the response path: v1 transform_response vs v2.
+
+Each case is a full cycle: the OpenAI request goes through both request
+translators (so per-request state like the tool-name reverse map flows the
+same way), then a recorded anthropic response JSON goes through v1's
+``transform_response`` and v2's ``parse_response`` -> ``serialize_response``
+-> ``ModelResponse(**body)``, and the two ``ModelResponse`` dumps must be
+identical. uuid/time are frozen because both sides mint ambient ids.
+"""
+
+import copy
+import json
+
+import httpx
+import pytest
+
+import litellm
+from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+from litellm.types.utils import ModelResponse
+
+from litellm.translation_seam import build_translation_deps, to_model_response
+
+from litellm.translation.inbound.openai_chat import parse_request
+from litellm.translation.inbound.openai_chat.response import serialize_response
+from litellm.translation.providers.anthropic.response import parse_response
+
+MODEL = "claude-sonnet-4-5"
+
+_REQUESTS = {
+    "text": {
+        "model": MODEL,
+        "max_tokens": 64,
+        "messages": [{"role": "user", "content": "hi"}],
+    },
+    "tools": {
+        "model": MODEL,
+        "max_tokens": 64,
+        "tools": [
+            {
+                "type": "function",
+                "function": {
+                    "name": "mcp.server/get_weather",
+                    "parameters": {"type": "object", "properties": {}},
+                },
+            }
+        ],
+        "messages": [{"role": "user", "content": "weather in Paris"}],
+    },
+    "thinking": {
+        "model": "claude-3-7-sonnet-20250219",
+        "max_tokens": 2048,
+        "thinking": {"type": "enabled", "budget_tokens": 1024},
+        "messages": [{"role": "user", "content": "think"}],
+    },
+    "json_tool": {
+        "model": "claude-3-5-haiku-20241022",
+        "max_tokens": 256,
+        "messages": [{"role": "user", "content": "capital of France?"}],
+        "response_format": {
+            "type": "json_schema",
+            "json_schema": {
+                "name": "answer",
+                "schema": {
+                    "type": "object",
+                    "properties": {"capital": {"type": "string"}},
+                },
+            },
+        },
+    },
+}
+
+_RESPONSES = {
+    "text": {
+        "id": "msg_01XYZ",
+        "type": "message",
+        "role": "assistant",
+        "model": "claude-sonnet-4-5",
+        "content": [{"type": "text", "text": "Hello there."}],
+        "stop_reason": "end_turn",
+        "stop_sequence": None,
+        "usage": {
+            "input_tokens": 12,
+            "output_tokens": 6,
+            "cache_creation_input_tokens": 0,
+            "cache_read_input_tokens": 0,
+            "cache_creation": {
+                "ephemeral_5m_input_tokens": 0,
+                "ephemeral_1h_input_tokens": 0,
+            },
+            "service_tier": "standard",
+        },
+    },
+    "tools": {
+        "id": "msg_01ABC",
+        "type": "message",
+        "role": "assistant",
+        "model": "claude-sonnet-4-5",
+        "content": [
+            {"type": "text", "text": "Checking."},
+            {
+                "type": "tool_use",
+                "id": "toolu_01",
+                "name": "mcp_server_get_weather",
+                "input": {"city": "Paris"},
+            },
+        ],
+        "stop_reason": "tool_use",
+        "stop_sequence": None,
+        "usage": {
+            "input_tokens": 30,
+            "output_tokens": 20,
+            "cache_creation_input_tokens": 4,
+            "cache_read_input_tokens": 8,
+            "cache_creation": {
+                "ephemeral_5m_input_tokens": 4,
+                "ephemeral_1h_input_tokens": 0,
+            },
+        },
+    },
+    "thinking": {
+        "id": "msg_01THINK",
+        "type": "message",
+        "role": "assistant",
+        "model": "claude-3-7-sonnet-20250219",
+        "content": [
+            {
+                "type": "thinking",
+                "thinking": "The capital of France is Paris.",
+                "signature": "sig==",
+            },
+            {"type": "text", "text": "Paris."},
+        ],
+        "stop_reason": "end_turn",
+        "stop_sequence": None,
+        "usage": {"input_tokens": 40, "output_tokens": 30},
+    },
+    "json_tool": {
+        "id": "msg_01JSON",
+        "type": "message",
+        "role": "assistant",
+        "model": "claude-3-5-haiku-20241022",
+        "content": [
+            {
+                "type": "tool_use",
+                "id": "toolu_02",
+                "name": "json_tool_call",
+                "input": {"capital": "Paris"},
+            }
+        ],
+        "stop_reason": "tool_use",
+        "stop_sequence": None,
+        "usage": {"input_tokens": 25, "output_tokens": 12},
+    },
+}
+
+FROZEN_TIME = 1718064000.0
+
+
+@pytest.fixture(autouse=True)
+def _deterministic_ambient(monkeypatch):
+    import time
+    import uuid
+
+    counter = iter(range(1, 10_000))
+    monkeypatch.setattr(
+        uuid,
+        "uuid4",
+        lambda: uuid.UUID(int=next(counter)),
+    )
+    monkeypatch.setattr(time, "time", lambda: FROZEN_TIME)
+
+
+def _v1_model_response(name: str) -> dict:
+    request = copy.deepcopy(_REQUESTS[name])
+    config = AnthropicConfig()
+    model = request["model"]
+    params = {
+        key: value for key, value in request.items() if key not in ("model", "messages")
+    }
+    optional = config.map_openai_params(
+        copy.deepcopy(params), {}, model, drop_params=False
+    )
+    litellm_params: dict = {}
+    config.transform_request(
+        model, copy.deepcopy(request["messages"]), optional, litellm_params, {}
+    )
+    json_mode = optional.pop("json_mode", False)
+    raw = httpx.Response(
+        status_code=200,
+        json=_RESPONSES[name],
+        request=httpx.Request("POST", "https://api.anthropic.com/v1/messages"),
+    )
+    logging = litellm.litellm_core_utils.litellm_logging.Logging(
+        model=model,
+        messages=copy.deepcopy(request["messages"]),
+        stream=False,
+        call_type="completion",
+        start_time=None,
+        litellm_call_id="char-test",
+        function_id="char-test",
+    )
+    result = config.transform_response(
+        model=model,
+        raw_response=raw,
+        model_response=ModelResponse(),
+        logging_obj=logging,
+        request_data={},
+        messages=copy.deepcopy(request["messages"]),
+        optional_params=optional,
+        litellm_params=litellm_params,
+        encoding=None,
+        json_mode=json_mode,
+    )
+    return result.model_dump()
+
+
+def _v2_model_response(name: str) -> dict:
+    request = copy.deepcopy(_REQUESTS[name])
+    deps = build_translation_deps()
+    parsed = parse_request(request)
+    assert parsed.is_ok(), parsed.error.summary
+    response = parse_response(copy.deepcopy(_RESPONSES[name]), parsed.ok)
+    assert response.is_ok(), response.error.summary
+    body = serialize_response(response.ok, deps)
+    return to_model_response(body).model_dump()
+
+
+def _norm(payload: dict) -> str:
+    # The chatcmpl id is ambient (each side mints its own from the frozen
+    # uuid counter in a different order); everything else must match.
+    return json.dumps({**payload, "id": "chatcmpl-X"}, sort_keys=True, default=str)
+
+
+@pytest.mark.parametrize("name", sorted(_REQUESTS))
+def test_v2_response_matches_v1(name: str) -> None:
+    v1 = _v1_model_response(name)
+    v2 = _v2_model_response(name)
+    assert _norm(v2) == _norm(v1)

tests/test_litellm/translation/test_differential_anthropic_stream.py

@@ -0,0 +1,245 @@
+"""Differential parity for streaming: v1's full CustomStreamWrapper replay vs
+v2's engine/stream fold over the same recorded SSE lines.
+
+The replay method mirrors the characterization corpus
+(tests/translation_characterization/_seams.py): raw SSE lines through the
+real ``ModelResponseIterator`` inside ``CustomStreamWrapper`` on the v1 side;
+``engine.stream.fold_lines`` plus the ``ModelResponseStream`` seam adapter on
+the v2 side. uuid/time are frozen; stream ids are ambient and normalized.
+"""
+
+import json
+
+import pytest
+
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+
+from litellm.translation.engine.stream import fold_lines
+from litellm.translation.inbound.openai_chat import parse_request
+from litellm.translation.providers.anthropic.stream import parse_sse_line, reverse_names
+from litellm.translation_seam import to_model_response_stream
+
+MODEL = "claude-sonnet-4-5"
+
+FROZEN_TIME = 1718064000.0
+
+
+def _sse(events: list) -> list:
+    lines = []
+    for event in events:
+        lines.append(f"event: {event['type']}")
+        lines.append("data: " + json.dumps(event))
+        lines.append("")
+    return lines
+
+
+_MESSAGE_START = {
+    "type": "message_start",
+    "message": {
+        "id": "msg_stream_01",
+        "type": "message",
+        "role": "assistant",
+        "model": MODEL,
+        "content": [],
+        "stop_reason": None,
+        "stop_sequence": None,
+        "usage": {"input_tokens": 12, "output_tokens": 1},
+    },
+}
+
+STREAMS = {
+    "text": _sse(
+        [
+            _MESSAGE_START,
+            {"type": "ping"},
+            {
+                "type": "content_block_start",
+                "index": 0,
+                "content_block": {"type": "text", "text": ""},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 0,
+                "delta": {"type": "text_delta", "text": "Paris is the"},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 0,
+                "delta": {"type": "text_delta", "text": " capital."},
+            },
+            {"type": "content_block_stop", "index": 0},
+            {
+                "type": "message_delta",
+                "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+                "usage": {"output_tokens": 9},
+            },
+            {"type": "message_stop"},
+        ]
+    ),
+    "tools": _sse(
+        [
+            _MESSAGE_START,
+            {
+                "type": "content_block_start",
+                "index": 0,
+                "content_block": {"type": "text", "text": ""},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 0,
+                "delta": {"type": "text_delta", "text": "Checking."},
+            },
+            {"type": "content_block_stop", "index": 0},
+            {
+                "type": "content_block_start",
+                "index": 1,
+                "content_block": {
+                    "type": "tool_use",
+                    "id": "toolu_s1",
+                    "name": "mcp_server_get_weather",
+                    "input": {},
+                },
+            },
+            {
+                "type": "content_block_delta",
+                "index": 1,
+                "delta": {"type": "input_json_delta", "partial_json": '{"ci'},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 1,
+                "delta": {"type": "input_json_delta", "partial_json": 'ty": "Paris"}'},
+            },
+            {"type": "content_block_stop", "index": 1},
+            {
+                "type": "message_delta",
+                "delta": {"stop_reason": "tool_use", "stop_sequence": None},
+                "usage": {"output_tokens": 30},
+            },
+            {"type": "message_stop"},
+        ]
+    ),
+    "thinking": _sse(
+        [
+            _MESSAGE_START,
+            {
+                "type": "content_block_start",
+                "index": 0,
+                "content_block": {"type": "thinking", "thinking": "", "signature": ""},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 0,
+                "delta": {"type": "thinking_delta", "thinking": "France. Capital."},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 0,
+                "delta": {"type": "signature_delta", "signature": "sig=="},
+            },
+            {"type": "content_block_stop", "index": 0},
+            {
+                "type": "content_block_start",
+                "index": 1,
+                "content_block": {"type": "text", "text": ""},
+            },
+            {
+                "type": "content_block_delta",
+                "index": 1,
+                "delta": {"type": "text_delta", "text": "Paris."},
+            },
+            {"type": "content_block_stop", "index": 1},
+            {
+                "type": "message_delta",
+                "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+                "usage": {"output_tokens": 20},
+            },
+            {"type": "message_stop"},
+        ]
+    ),
+}
+
+_REQUEST = {
+    "model": MODEL,
+    "max_tokens": 64,
+    "tools": [
+        {
+            "type": "function",
+            "function": {
+                "name": "mcp.server/get_weather",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        }
+    ],
+    "messages": [{"role": "user", "content": "weather in Paris"}],
+}
+
+
+@pytest.fixture(autouse=True)
+def _deterministic_ambient(monkeypatch):
+    import time
+    import uuid
+
+    counter = iter(range(1, 10_000))
+    monkeypatch.setattr(uuid, "uuid4", lambda: uuid.UUID(int=next(counter)))
+    monkeypatch.setattr(time, "time", lambda: FROZEN_TIME)
+
+
+def _v1_chunks(lines: list) -> list:
+    from litellm.llms.anthropic.chat.handler import ModelResponseIterator
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    # Production passes the per-request tool-name reverse map into the
+    # iterator (built by transform_request); replicate that wiring.
+    config = AnthropicConfig()
+    params = {k: v for k, v in _REQUEST.items() if k not in ("model", "messages")}
+    optional = config.map_openai_params(dict(params), {}, MODEL, drop_params=False)
+    litellm_params: dict = {}
+    config.transform_request(MODEL, [dict(m) for m in _REQUEST["messages"]], optional, litellm_params, {})
+    reverse_map = litellm_params.get("_anthropic_tool_name_map") or {}
+
+    iterator = ModelResponseIterator(
+        streaming_response=iter(lines),
+        sync_stream=True,
+        tool_name_reverse_map=reverse_map,
+    )
+    wrapper = CustomStreamWrapper(
+        completion_stream=iterator,
+        model=MODEL,
+        custom_llm_provider="anthropic",
+        logging_obj=Logging(
+            model=MODEL,
+            messages=[{"role": "user", "content": "stream"}],
+            stream=True,
+            call_type="completion",
+            start_time=None,
+            litellm_call_id="diff-stream",
+            function_id="diff-stream",
+        ),
+    )
+    return [chunk.model_dump() for chunk in wrapper]
+
+
+def _v2_chunks(lines: list) -> list:
+    parsed = parse_request(dict(_REQUEST))
+    assert parsed.is_ok(), parsed.error.summary
+    reverse = reverse_names(parsed.ok)
+    folded = fold_lines(lines, lambda line: parse_sse_line(line, reverse))
+    assert folded.is_ok(), folded.error.summary
+    return [
+        to_model_response_stream(chunk, "chatcmpl-X").model_dump()
+        for chunk in folded.ok
+    ]
+
+
+def _norm(chunks: list) -> str:
+    return json.dumps(
+        [{**chunk, "id": "chatcmpl-X"} for chunk in chunks], sort_keys=True, default=str
+    )
+
+
+@pytest.mark.parametrize("name", sorted(STREAMS))
+def test_v2_stream_matches_v1(name: str) -> None:
+    lines = STREAMS[name]
+    assert _norm(_v2_chunks(lines)) == _norm(_v1_chunks(lines))

tests/test_litellm/translation/test_seam.py

@@ -0,0 +1,139 @@
+"""The completion() fork: flag-gated v2 execution with fail-closed fallback.
+
+Uses respx to intercept the anthropic endpoint, so these run the REAL
+pipeline (translate -> http port -> response translate -> ModelResponse)
+end to end without network.
+"""
+
+import json
+
+import pytest
+import respx
+from httpx import Response
+
+import litellm
+
+_URL = "https://api.anthropic.com/v1/messages"
+
+_PROVIDER_RESPONSE = {
+    "id": "msg_seam_01",
+    "type": "message",
+    "role": "assistant",
+    "model": "claude-sonnet-4-5",
+    "content": [{"type": "text", "text": "Hello from v2."}],
+    "stop_reason": "end_turn",
+    "stop_sequence": None,
+    "usage": {"input_tokens": 10, "output_tokens": 5},
+}
+
+
+@pytest.fixture(autouse=True)
+def _flag(monkeypatch):
+    monkeypatch.setattr(litellm, "translation_v2_providers", ["anthropic"])
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+
+
+@respx.mock
+def test_flag_on_serves_through_v2() -> None:
+    route = respx.post(_URL).mock(return_value=Response(200, json=_PROVIDER_RESPONSE))
+    response = litellm.completion(
+        model="anthropic/claude-sonnet-4-5",
+        max_tokens=32,
+        messages=[{"role": "user", "content": "hi"}],
+    )
+    assert route.called
+    sent = json.loads(route.calls.last.request.content)
+    assert sent["model"] == "claude-sonnet-4-5"
+    assert "json_mode" not in sent
+    assert response.choices[0].message.content == "Hello from v2."
+    assert response.usage.prompt_tokens == 10
+    assert response.usage.completion_tokens == 5
+
+
+@respx.mock
+def test_unsupported_shape_falls_back_to_v1() -> None:
+    route = respx.post(_URL).mock(return_value=Response(200, json=_PROVIDER_RESPONSE))
+    response = litellm.completion(
+        model="anthropic/claude-sonnet-4-5",
+        max_tokens=32,
+        messages=[
+            {"role": "user", "content": "hi"},
+            # assistant prefix is a v1 feature outside v2's surface: the
+            # request must be served (by v1), never rejected or dropped.
+            {"role": "assistant", "content": "Hel", "prefix": True},
+        ],
+    )
+    assert route.called
+    assert response.choices[0].message.content == "Hello from v2."
+
+
+@respx.mock
+def test_provider_error_raises_contract_not_fallback() -> None:
+    route = respx.post(_URL).mock(
+        return_value=Response(429, json={"error": {"message": "rate limited"}})
+    )
+    with pytest.raises(litellm.exceptions.RateLimitError):
+        litellm.completion(
+            model="anthropic/claude-sonnet-4-5",
+            max_tokens=32,
+            messages=[{"role": "user", "content": "hi"}],
+            num_retries=0,
+        )
+    assert route.call_count == 1  # sent once: no silent re-send through v1
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_acompletion_path_returns_awaitable() -> None:
+    respx.post(_URL).mock(return_value=Response(200, json=_PROVIDER_RESPONSE))
+    response = await litellm.acompletion(
+        model="anthropic/claude-sonnet-4-5",
+        max_tokens=32,
+        messages=[{"role": "user", "content": "hi"}],
+    )
+    assert response.choices[0].message.content == "Hello from v2."
+
+
+def test_flag_off_returns_none_from_seam() -> None:
+    from litellm.translation_seam import try_completion_v2
+
+    litellm.translation_v2_providers = []
+    assert (
+        try_completion_v2(
+            model="claude-sonnet-4-5",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_param_args={},
+            non_default_params={},
+            api_key=None,
+            api_base=None,
+            timeout=None,
+            stream=None,
+            acompletion=None,
+            logging_obj=None,
+            model_response=None,
+            request_drop_params=None,
+        )
+        is None
+    )
+
+
+def test_streaming_stays_on_v1() -> None:
+    from litellm.translation_seam import try_completion_v2
+
+    assert (
+        try_completion_v2(
+            model="claude-sonnet-4-5",
+            messages=[{"role": "user", "content": "hi"}],
+            optional_param_args={},
+            non_default_params={},
+            api_key=None,
+            api_base=None,
+            timeout=None,
+            stream=True,
+            acompletion=None,
+            logging_obj=None,
+            model_response=None,
+            request_drop_params=None,
+        )
+        is None
+    )