feat(proxy): add Amazon Transcribe SigV4 pass-through routes

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/_lazy_features.py

@@ -208,6 +208,7 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
             "/nvidia_nim/",
             "/openai/",
             "/openai_passthrough/",
+            "/transcribe",
             "/vertex-ai/",
             "/vertex_ai/",
             "/vllm/",

litellm/proxy/_lazy_openapi_snapshot.json

@@ -20373,6 +20373,77 @@
           ]
         }
       },
+      "/transcribe": {
+        "post": {
+          "description": "AWS-SDK-shaped pass-through for Amazon Transcribe: point the SDK's `endpoint_url`\nat `/transcribe` and the operation is read from the `X-Amz-Target` header, per the\nAWS JSON 1.1 protocol.\n\n[Docs](https://docs.litellm.ai/docs/pass_through/transcribe)",
+          "operationId": "transcribe_sdk_proxy_route_transcribe_post",
+          "responses": {
+            "200": {
+              "content": {
+                "application/json": {
+                  "schema": {}
+                }
+              },
+              "description": "Successful Response"
+            }
+          },
+          "security": [
+            {
+              "APIKeyHeader": []
+            }
+          ],
+          "summary": "Transcribe Sdk Proxy Route",
+          "tags": [
+            "llm_passthrough"
+          ]
+        }
+      },
+      "/transcribe/{operation}": {
+        "post": {
+          "description": "Pass-through for the Amazon Transcribe API, e.g. `POST /transcribe/StartTranscriptionJob`.\n\nThe request body is forwarded as-is to the AWS JSON 1.1 API and signed with SigV4\nusing the proxy's AWS credentials. Streaming transcription (`transcribestreaming`)\nuses a separate HTTP/2 event-stream protocol and is not served by this route.\n\n[Docs](https://docs.litellm.ai/docs/pass_through/transcribe)",
+          "operationId": "transcribe_proxy_route_transcribe__operation__post",
+          "parameters": [
+            {
+              "in": "path",
+              "name": "operation",
+              "required": true,
+              "schema": {
+                "title": "Operation",
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "content": {
+                "application/json": {
+                  "schema": {}
+                }
+              },
+              "description": "Successful Response"
+            },
+            "422": {
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "$ref": "#/components/schemas/HTTPValidationError"
+                  }
+                }
+              },
+              "description": "Validation Error"
+            }
+          },
+          "security": [
+            {
+              "APIKeyHeader": []
+            }
+          ],
+          "summary": "Transcribe Proxy Route",
+          "tags": [
+            "llm_passthrough"
+          ]
+        }
+      },
       "/vertex_ai/discovery/{endpoint}": {
         "delete": {
           "description": "Call any vertex discovery endpoint using the proxy.\n\nJust use `{PROXY_BASE_URL}/vertex_ai/discovery/{endpoint:path}`\n\nTarget url: `https://discoveryengine.googleapis.com`",

litellm/proxy/_types.py

@@ -468,6 +468,7 @@ class LiteLLMRoutes(enum.Enum):
     mapped_pass_through_routes = [
         "/bedrock",
         "/comprehendmedical",
+        "/transcribe",
         "/vertex-ai",
         "/vertex_ai",
         "/cohere",

litellm/proxy/middleware/billable_request_metrics_middleware.py

@@ -93,6 +93,7 @@ _LLM_ROUTE_EXACT: Final[tuple[str, ...]] = (
     "/interactions",  # Google Interactions create; /{id} reads and /cancel do not match
     "/v1beta/interactions",
     "/comprehendmedical",  # AWS-SDK-shaped passthrough: the operation rides in the X-Amz-Target header
+    "/transcribe",
 )
 
 # Provider passthrough prefixes (e.g. /bedrock/..., /vertex-ai/...) carry real

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -1200,7 +1200,7 @@ async def bedrock_proxy_route(
 COMPREHEND_MEDICAL_TARGET_PREFIX: Final = "ComprehendMedical_20181030"
 
 
-def _resolve_comprehend_medical_region() -> str | None:
+def _resolve_aws_passthrough_region() -> str | None:
     region_candidates: Final = (
         get_secret_str(secret_name="AWS_REGION_NAME"),
         get_secret_str(secret_name="AWS_REGION"),
@@ -1240,7 +1240,7 @@ async def comprehend_medical_proxy_route(
             ),
         )
 
-    aws_region_name: Final = _resolve_comprehend_medical_region()
+    aws_region_name: Final = _resolve_aws_passthrough_region()
     if aws_region_name is None:
         raise HTTPException(
             status_code=400,
@@ -1317,6 +1317,121 @@ async def comprehend_medical_sdk_proxy_route(
     )
 
 
+@router.post(
+    "/transcribe/{operation}",
+    tags=["Amazon Transcribe Pass-through", "pass-through"],  # mutable-ok: fastapi route tags must be a list
+)
+async def transcribe_proxy_route(
+    operation: str,
+    request: Request,
+    fastapi_response: Response,
+    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
+):
+    """
+    Pass-through for the Amazon Transcribe API, e.g. `POST /transcribe/StartTranscriptionJob`.
+
+    The request body is forwarded as-is to the AWS JSON 1.1 API and signed with SigV4
+    using the proxy's AWS credentials. Streaming transcription (`transcribestreaming`)
+    uses a separate HTTP/2 event-stream protocol and is not served by this route.
+
+    [Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
+    """
+    from .llm_provider_handlers.transcribe_passthrough_logging_handler import (
+        TRANSCRIBE_CUSTOM_LLM_PROVIDER,
+        TRANSCRIBE_TARGET_PREFIX,
+        transcribe_supported_operations,
+    )
+
+    if operation not in transcribe_supported_operations():
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Unsupported Amazon Transcribe operation: {operation}. "
+                f"Supported operations: {', '.join(sorted(transcribe_supported_operations()))}"
+            ),
+        )
+
+    aws_region_name: Final = _resolve_aws_passthrough_region()
+    if aws_region_name is None:
+        raise HTTPException(
+            status_code=400,
+            detail="AWS region not found. Set AWS_REGION_NAME in the proxy environment.",
+        )
+
+    try:
+        data: Final = await _json_request_body(request)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Request body must be valid JSON: {e}")
+
+    if not isinstance(data, dict):
+        raise HTTPException(status_code=400, detail="Request body must be a JSON object")
+    if "stream" in data:
+        raise HTTPException(status_code=400, detail="'stream' is not an Amazon Transcribe request member")
+
+    from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing, sign_aws_json_post
+
+    target_url: Final = f"https://transcribe.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}/"
+    prepped: Final = await run_aws_signing(
+        sign_aws_json_post,
+        get_credentials=partial(BaseAWSLLM().get_credentials, aws_region_name=aws_region_name),
+        service_name="transcribe",
+        aws_region_name=aws_region_name,
+        url=target_url,
+        body=json.dumps(data),
+        headers=MappingProxyType(
+            {
+                "Content-Type": "application/x-amz-json-1.1",
+                "X-Amz-Target": f"{TRANSCRIBE_TARGET_PREFIX}.{operation}",
+            }
+        ),
+    )
+
+    endpoint_func: Final = create_pass_through_route(
+        endpoint=operation,
+        target=str(prepped.url),
+        custom_headers=prepped.headers,
+        custom_llm_provider=TRANSCRIBE_CUSTOM_LLM_PROVIDER,
+    )
+    setattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY, data)
+    setattr(request.state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, prepped.body)
+    return await endpoint_func(request, fastapi_response, user_api_key_dict)
+
+
+@router.post(
+    "/transcribe",
+    tags=["Amazon Transcribe Pass-through", "pass-through"],  # mutable-ok: fastapi route tags must be a list
+)
+async def transcribe_sdk_proxy_route(
+    request: Request,
+    fastapi_response: Response,
+    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
+):
+    """
+    AWS-SDK-shaped pass-through for Amazon Transcribe: point the SDK's `endpoint_url`
+    at `/transcribe` and the operation is read from the `X-Amz-Target` header, per the
+    AWS JSON 1.1 protocol.
+
+    [Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
+    """
+    from .llm_provider_handlers.transcribe_passthrough_logging_handler import (
+        TRANSCRIBE_TARGET_PREFIX,
+    )
+
+    target_header: Final = request.headers.get("x-amz-target", "")
+    target_prefix, _, operation = target_header.partition(".")
+    if target_prefix != TRANSCRIBE_TARGET_PREFIX or not operation:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Expected an X-Amz-Target header of the form {TRANSCRIBE_TARGET_PREFIX}.<Operation>",
+        )
+    return await transcribe_proxy_route(
+        operation=operation,
+        request=request,
+        fastapi_response=fastapi_response,
+        user_api_key_dict=user_api_key_dict,
+    )
+
+
 def _resolve_vertex_model_from_router(
     model_id: str,
     llm_router: litellm.Router | None,

litellm/proxy/pass_through_endpoints/llm_provider_handlers/transcribe_passthrough_logging_handler.py

@@ -0,0 +1,90 @@
+from collections.abc import Mapping
+from datetime import datetime
+from functools import lru_cache
+from typing import Final
+
+import httpx
+
+from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.litellm_core_utils.litellm_logging import (
+    get_standard_logging_object_payload,
+)
+from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
+from litellm.types.utils import StandardPassThroughResponseObject
+
+TRANSCRIBE_TARGET_PREFIX: Final = "Transcribe"
+TRANSCRIBE_CUSTOM_LLM_PROVIDER: Final = "transcribe"
+
+
+@lru_cache(maxsize=1)
+def transcribe_supported_operations() -> frozenset[str]:
+    """
+    Operation names of the Amazon Transcribe JSON 1.1 API, read from the botocore
+    service model so the allowlist tracks the installed SDK instead of a hand-typed copy.
+    """
+    from botocore.session import get_session
+
+    return frozenset(get_session().get_service_model("transcribe").operation_names)
+
+
+class TranscribePassthroughLoggingHandler:
+    @staticmethod
+    def _operation_from_response(httpx_response: httpx.Response) -> str:
+        target: Final = httpx_response.request.headers.get("x-amz-target", "")
+        return target.split(".")[-1]
+
+    @staticmethod
+    def transcribe_passthrough_handler(
+        httpx_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        url_route: str,
+        result: str,
+        start_time: datetime,
+        end_time: datetime,
+        cache_hit: bool,
+        request_body: Mapping[str, object],
+        **kwargs: object,  # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler
+    ) -> PassThroughEndpointLoggingTypedDict:
+        """
+        Records model and provider for an Amazon Transcribe control-plane call. Transcribe
+        bills per second of audio once a job finishes, which no request or response on this
+        path carries, so response_cost is recorded as 0.0 rather than estimated.
+        """
+        try:
+            operation: Final = TranscribePassthroughLoggingHandler._operation_from_response(httpx_response)
+            model_name: Final = f"{TRANSCRIBE_CUSTOM_LLM_PROVIDER}/{operation}"
+
+            updated_kwargs: Final = {  # mutable-ok: the logging pipeline requires a plain kwargs dict
+                **kwargs,
+                "model": model_name,
+                "custom_llm_provider": TRANSCRIBE_CUSTOM_LLM_PROVIDER,
+                "response_cost": 0.0,
+            }
+            logging_obj.model_call_details.update(
+                model=model_name,
+                custom_llm_provider=TRANSCRIBE_CUSTOM_LLM_PROVIDER,
+                response_cost=0.0,
+            )
+
+            standard_logging_object: Final = get_standard_logging_object_payload(
+                kwargs=updated_kwargs,
+                init_response_obj=StandardPassThroughResponseObject(response=result),
+                start_time=start_time,
+                end_time=end_time,
+                logging_obj=logging_obj,
+                status="success",
+            )
+
+            handler_payload: Final[PassThroughEndpointLoggingTypedDict] = {
+                "result": StandardPassThroughResponseObject(response=result),
+                "kwargs": {**updated_kwargs, "standard_logging_object": standard_logging_object},
+            }
+        except Exception as e:  # noqa: BLE001  # logging must never fail the forwarded request
+            verbose_proxy_logger.exception("Error in Amazon Transcribe passthrough logging handler: %s", e)
+            fallback_payload: Final[PassThroughEndpointLoggingTypedDict] = {
+                "result": StandardPassThroughResponseObject(response=result),
+                "kwargs": kwargs,
+            }
+            return fallback_payload
+        return handler_payload

litellm/proxy/pass_through_endpoints/success_handler.py

@@ -27,6 +27,10 @@ from .llm_provider_handlers.cursor_passthrough_logging_handler import (
 from .llm_provider_handlers.gemini_passthrough_logging_handler import (
     GeminiPassthroughLoggingHandler,
 )
+from .llm_provider_handlers.transcribe_passthrough_logging_handler import (
+    TRANSCRIBE_CUSTOM_LLM_PROVIDER,
+    TranscribePassthroughLoggingHandler,
+)
 from .llm_provider_handlers.vertex_passthrough_logging_handler import (
     VertexPassthroughLoggingHandler,
 )
@@ -256,6 +260,20 @@ class PassThroughEndpointLogging:
             )
             standard_logging_response_object = comprehend_medical_handler_result["result"]  # rebind-ok: elif-chain
             kwargs = comprehend_medical_handler_result["kwargs"]  # rebind-ok: elif-chain contract
+        elif self.is_transcribe_route(custom_llm_provider):
+            transcribe_handler_result: Final = TranscribePassthroughLoggingHandler.transcribe_passthrough_handler(
+                httpx_response=httpx_response,
+                logging_obj=logging_obj,
+                url_route=url_route,
+                result=result,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=cache_hit,
+                request_body=request_body,
+                **kwargs,
+            )
+            standard_logging_response_object = transcribe_handler_result["result"]  # rebind-ok: elif-chain
+            kwargs = transcribe_handler_result["kwargs"]  # rebind-ok: elif-chain contract
         elif self.is_vertex_ai_live_route(url_route):
             from .llm_provider_handlers.vertex_ai_live_passthrough_logging_handler import (
                 VertexAILivePassthroughLoggingHandler,
@@ -389,6 +407,9 @@ class PassThroughEndpointLogging:
     def is_comprehend_medical_route(self, custom_llm_provider: str | None) -> bool:
         return custom_llm_provider == "comprehendmedical"
 
+    def is_transcribe_route(self, custom_llm_provider: str | None) -> bool:
+        return custom_llm_provider == TRANSCRIBE_CUSTOM_LLM_PROVIDER
+
     def is_langfuse_route(self, url_route: str):
         parsed_url: Final = urlparse(url_route)
         for route in self.TRACKED_LANGFUSE_ROUTES:

tests/pass_through_unit_tests/test_pass_through_unit_tests.py

@@ -411,6 +411,8 @@ async def test_pass_through_request_logging_failure_with_stream(
 PROTOCOL_CONSTRAINED_PASS_THROUGH_ROUTES = {
     "/comprehendmedical": {"POST"},
     "/comprehendmedical/{operation}": {"POST"},
+    "/transcribe": {"POST"},
+    "/transcribe/{operation}": {"POST"},
 }
 
 
@@ -419,8 +421,8 @@ def test_pass_through_routes_support_all_methods():
     A pass-through route fronts a whole provider API, so narrowing its method
     set turns a request the upstream would have accepted into a 405. The
     exceptions are providers whose wire protocol admits only one method: Amazon
-    Comprehend Medical speaks AWS JSON 1.1, which is POST-only, so there is no
-    other method to forward.
+    Comprehend Medical and Amazon Transcribe speak AWS JSON 1.1, which is
+    POST-only, so there is no other method to forward.
     """
     from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
         router as llm_router,

tests/test_litellm/proxy/middleware/test_billable_request_metrics_middleware.py

@@ -116,6 +116,8 @@ def test_is_pure_asgi_not_base_http_middleware():
         # Bare AWS-SDK-shaped route carries the operation in X-Amz-Target and writes SpendLogs
         ("/comprehendmedical", (BillableCategory.LLM, "/comprehendmedical")),
         ("/comprehendmedical/DetectEntitiesV2", (BillableCategory.LLM, "/comprehendmedical")),
+        ("/transcribe", (BillableCategory.LLM, "/transcribe")),
+        ("/transcribe/StartTranscriptionJob", (BillableCategory.LLM, "/transcribe")),
         ("/mcp", (BillableCategory.MCP, "/mcp")),
         ("/mcp/", (BillableCategory.MCP, "/mcp")),
         ("/mcp/tools/list", (BillableCategory.MCP, "/mcp")),

tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_transcribe_passthrough_logging_handler.py

@@ -0,0 +1,110 @@
+from datetime import datetime
+from unittest.mock import MagicMock
+
+import httpx
+
+from litellm.proxy.pass_through_endpoints.llm_provider_handlers.transcribe_passthrough_logging_handler import (
+    TranscribePassthroughLoggingHandler,
+    transcribe_supported_operations,
+)
+from litellm.proxy.pass_through_endpoints.success_handler import (
+    PassThroughEndpointLogging,
+)
+
+
+def _make_response(operation: str) -> httpx.Response:
+    request = httpx.Request(
+        "POST",
+        "https://transcribe.us-west-2.amazonaws.com/",
+        headers={"X-Amz-Target": f"Transcribe.{operation}"},
+    )
+    return httpx.Response(200, request=request, text='{"TranscriptionJob": {}}')
+
+
+def _make_logging_obj() -> MagicMock:
+    logging_obj = MagicMock()
+    logging_obj.litellm_call_id = "test-call-id"
+    logging_obj.model_call_details = {}
+    return logging_obj
+
+
+class TestTranscribeSupportedOperations:
+    def test_matches_the_installed_botocore_service_model(self):
+        from botocore.session import get_session
+
+        assert transcribe_supported_operations() == frozenset(
+            get_session().get_service_model("transcribe").operation_names
+        )
+        assert "StartTranscriptionJob" in transcribe_supported_operations()
+
+
+class TestTranscribePassthroughHandler:
+    def test_records_model_provider_and_zero_cost(self):
+        logging_obj = _make_logging_obj()
+        request_body = {"TranscriptionJobName": "litellm-job-1"}
+
+        handler_result = TranscribePassthroughLoggingHandler.transcribe_passthrough_handler(
+            httpx_response=_make_response("StartTranscriptionJob"),
+            logging_obj=logging_obj,
+            url_route="https://transcribe.us-west-2.amazonaws.com/",
+            result='{"TranscriptionJob": {}}',
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+            cache_hit=False,
+            request_body=request_body,
+        )
+
+        assert handler_result["result"] == {"response": '{"TranscriptionJob": {}}'}
+        assert handler_result["kwargs"]["model"] == "transcribe/StartTranscriptionJob"
+        assert handler_result["kwargs"]["custom_llm_provider"] == "transcribe"
+        assert handler_result["kwargs"]["response_cost"] == 0.0
+        assert "standard_logging_object" in handler_result["kwargs"]
+        assert logging_obj.model_call_details["model"] == "transcribe/StartTranscriptionJob"
+        assert logging_obj.model_call_details["custom_llm_provider"] == "transcribe"
+        assert logging_obj.model_call_details["response_cost"] == 0.0
+        assert request_body == {"TranscriptionJobName": "litellm-job-1"}
+
+
+class TestIsTranscribeRoute:
+    def test_matches_by_provider_tag(self):
+        assert PassThroughEndpointLogging().is_transcribe_route("transcribe")
+
+    def test_does_not_match_other_providers(self):
+        assert not PassThroughEndpointLogging().is_transcribe_route("comprehendmedical")
+
+    def test_dispatch_reaches_transcribe_handler(self):
+        logging_obj = _make_logging_obj()
+
+        normalized = PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
+            httpx_response=_make_response("GetTranscriptionJob"),
+            response_body={"TranscriptionJob": {}},
+            request_body={"TranscriptionJobName": "litellm-job-1"},
+            logging_obj=logging_obj,
+            url_route="https://transcribe.us-west-2.amazonaws.com/",
+            result='{"TranscriptionJob": {}}',
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+            cache_hit=False,
+            custom_llm_provider="transcribe",
+        )
+
+        assert normalized["kwargs"]["model"] == "transcribe/GetTranscriptionJob"
+        assert normalized["kwargs"]["response_cost"] == 0.0
+
+    def test_config_driven_passthrough_to_transcribe_host_is_not_claimed(self):
+        logging_obj = _make_logging_obj()
+
+        normalized = PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
+            httpx_response=_make_response("GetTranscriptionJob"),
+            response_body={"TranscriptionJob": {}},
+            request_body={"TranscriptionJobName": "litellm-job-1"},
+            logging_obj=logging_obj,
+            url_route="https://transcribe.us-west-2.amazonaws.com/",
+            result='{"TranscriptionJob": {}}',
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+            cache_hit=False,
+            custom_llm_provider=None,
+        )
+
+        assert normalized["kwargs"].get("model") != "transcribe/GetTranscriptionJob"

tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -5135,6 +5135,137 @@ class TestComprehendMedicalProxyRoute:
         assert exc_info.value.status_code == 400
 
 
+TRANSCRIBE_UPSTREAM = "https://transcribe.us-west-2.amazonaws.com/"
+
+
+@pytest.fixture
+def transcribe_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
+    from litellm.proxy.proxy_server import app
+
+    monkeypatch.setenv("AWS_REGION_NAME", "us-west-2")
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test-access-key")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test-secret-key")
+    monkeypatch.delenv("AWS_SESSION_TOKEN", raising=False)
+    monkeypatch.delenv("SERVER_ROOT_PATH", raising=False)
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    monkeypatch.setitem(app.dependency_overrides, user_api_key_auth, lambda: UserAPIKeyAuth(api_key="sk-virtual"))
+    yield TestClient(app)
+
+
+class TestTranscribeProxyRoute:
+    START_JOB_BODY: Final = MappingProxyType(
+        {
+            "TranscriptionJobName": "litellm-job-1",
+            "LanguageCode": "en-US",
+            "Media": {"MediaFileUri": "s3://bucket/audio.wav"},
+        }
+    )
+
+    def test_signs_and_forwards_start_transcription_job(self, transcribe_client: TestClient) -> None:
+        upstream_body = {"TranscriptionJob": {"TranscriptionJobName": "litellm-job-1", "TranscriptionJobStatus": "IN_PROGRESS"}}
+        with respx.mock(assert_all_called=True) as upstream:
+            route = upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(200, json=upstream_body))
+            response = transcribe_client.post(
+                "/transcribe/StartTranscriptionJob",
+                json=dict(self.START_JOB_BODY),
+                headers={"Authorization": "Bearer sk-virtual"},
+            )
+
+        assert (response.status_code, response.json()) == (200, upstream_body)
+        sent = route.calls.last.request
+        assert json.loads(sent.content) == dict(self.START_JOB_BODY)
+        assert sent.headers["x-amz-target"] == "Transcribe.StartTranscriptionJob"
+        assert sent.headers["content-type"] == "application/x-amz-json-1.1"
+        assert sent.headers["authorization"].startswith("AWS4-HMAC-SHA256 Credential=test-access-key/")
+        assert "/us-west-2/transcribe/aws4_request" in sent.headers["authorization"]
+        assert "x-amz-date" in sent.headers
+
+    def test_sdk_route_reads_operation_from_x_amz_target_and_resigns(self, transcribe_client: TestClient) -> None:
+        with respx.mock(assert_all_called=True) as upstream:
+            route = upstream.post(TRANSCRIBE_UPSTREAM).mock(
+                return_value=httpx.Response(200, json={"TranscriptionJob": {"TranscriptionJobStatus": "COMPLETED"}})
+            )
+            response = transcribe_client.post(
+                "/transcribe",
+                json={"TranscriptionJobName": "litellm-job-1"},
+                headers={
+                    "Authorization": "AWS4-HMAC-SHA256 Credential=sk-virtual/20260101/us-west-2/transcribe/aws4_request",
+                    "X-Amz-Target": "Transcribe.GetTranscriptionJob",
+                    "Content-Type": "application/x-amz-json-1.1",
+                },
+            )
+
+        assert (response.status_code, response.json()) == (200, {"TranscriptionJob": {"TranscriptionJobStatus": "COMPLETED"}})
+        sent = route.calls.last.request
+        assert sent.headers["x-amz-target"] == "Transcribe.GetTranscriptionJob"
+        assert "Credential=test-access-key/" in sent.headers["authorization"]
+        assert "sk-virtual" not in sent.headers["authorization"]
+
+    def test_upstream_error_status_and_body_are_returned(self, transcribe_client: TestClient) -> None:
+        aws_error = {"__type": "BadRequestException", "Message": "The requested job couldn't be found."}
+        with respx.mock(assert_all_called=True) as upstream:
+            upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(400, json=aws_error))
+            response = transcribe_client.post("/transcribe/GetTranscriptionJob", json={"TranscriptionJobName": "missing"})
+
+        assert (response.status_code, response.json()) == (400, aws_error)
+
+    @pytest.mark.parametrize(
+        "operation",
+        ["Start-Transcription-Job", "Transcribe.StartTranscriptionJob", "a" * 200, "starttranscriptionjob", "DetectEntitiesV2"],
+    )
+    def test_rejects_unsupported_operations_without_calling_aws(self, transcribe_client: TestClient, operation: str) -> None:
+        with respx.mock(assert_all_called=False) as upstream:
+            route = upstream.post(TRANSCRIBE_UPSTREAM)
+            response = transcribe_client.post(f"/transcribe/{operation}", json={})
+
+        assert response.status_code == 400
+        assert "Unsupported Amazon Transcribe operation" in response.json()["detail"]
+        assert not route.called
+
+    @pytest.mark.parametrize(
+        "raw_body",
+        ['{"MaxResults": 5, "stream": true}', '{"MaxResults": 5, "stream": false}', '["x"]', "not json"],
+    )
+    def test_rejects_bad_bodies_without_calling_aws(self, transcribe_client: TestClient, raw_body: str) -> None:
+        with respx.mock(assert_all_called=False) as upstream:
+            route = upstream.post(TRANSCRIBE_UPSTREAM)
+            response = transcribe_client.post(
+                "/transcribe/ListTranscriptionJobs", content=raw_body, headers={"Content-Type": "application/json"}
+            )
+
+        assert response.status_code == 400
+        assert not route.called
+
+    def test_missing_region_returns_400_without_calling_aws(
+        self, transcribe_client: TestClient, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        for name in ("AWS_REGION_NAME", "AWS_REGION", "AWS_DEFAULT_REGION"):
+            monkeypatch.delenv(name, raising=False)
+        with respx.mock(assert_all_called=False) as upstream:
+            route = upstream.post(TRANSCRIBE_UPSTREAM)
+            response = transcribe_client.post("/transcribe/ListTranscriptionJobs", json={})
+
+        assert response.status_code == 400
+        assert "AWS region" in response.json()["detail"]
+        assert not route.called
+
+    @pytest.mark.parametrize("target_header", ["", "Transcribe", "ComprehendMedical_20181030.DetectPHI", "Transcribe."])
+    def test_sdk_route_rejects_bad_x_amz_target(self, transcribe_client: TestClient, target_header: str) -> None:
+        with respx.mock(assert_all_called=False) as upstream:
+            route = upstream.post(TRANSCRIBE_UPSTREAM)
+            response = transcribe_client.post("/transcribe", json={}, headers={"X-Amz-Target": target_header})
+
+        assert response.status_code == 400
+        assert "X-Amz-Target" in response.json()["detail"]
+        assert not route.called
+
+    def test_transcribe_is_a_mapped_pass_through_route(self) -> None:
+        from litellm.proxy._types import LiteLLMRoutes
+
+        assert "/transcribe" in LiteLLMRoutes.mapped_pass_through_routes.value
+
+
 LIVE_RESOURCE_PATH = "projects/proj-db/locations/global/publishers/google/models/gemini-live-2.5-flash"
 
 

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -16437,6 +16437,56 @@ export interface paths {
         patch: operations["toolset_mcp_route_toolset__toolset_name__mcp_patch"];
         trace?: never;
     };
+    "/transcribe": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        /**
+         * Transcribe Sdk Proxy Route
+         * @description AWS-SDK-shaped pass-through for Amazon Transcribe: point the SDK's `endpoint_url`
+         *     at `/transcribe` and the operation is read from the `X-Amz-Target` header, per the
+         *     AWS JSON 1.1 protocol.
+         *
+         *     [Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
+         */
+        post: operations["transcribe_sdk_proxy_route_transcribe_post"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
+    "/transcribe/{operation}": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        /**
+         * Transcribe Proxy Route
+         * @description Pass-through for the Amazon Transcribe API, e.g. `POST /transcribe/StartTranscriptionJob`.
+         *
+         *     The request body is forwarded as-is to the AWS JSON 1.1 API and signed with SigV4
+         *     using the proxy's AWS credentials. Streaming transcription (`transcribestreaming`)
+         *     uses a separate HTTP/2 event-stream protocol and is not served by this route.
+         *
+         *     [Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
+         */
+        post: operations["transcribe_proxy_route_transcribe__operation__post"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/update/default_team_settings": {
         parameters: {
             query?: never;
@@ -61441,6 +61491,57 @@ export interface operations {
             };
         };
     };
+    transcribe_sdk_proxy_route_transcribe_post: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": unknown;
+                };
+            };
+        };
+    };
+    transcribe_proxy_route_transcribe__operation__post: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path: {
+                operation: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": unknown;
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     update_default_team_settings_update_default_team_settings_patch: {
         parameters: {
             query?: never;