fix(types): read upstream headers through a typed helper

litellm/llms/custom_httpx/http_handler.py

@@ -479,6 +479,11 @@ def _safe_get_response_text(response: httpx.Response) -> str:
         return ""
 
 
+def header_value(headers: Mapping[str, str], name: str) -> str | None:
+    """Read one header as ``str | None``; ``httpx.Headers.get`` itself is typed ``Any``."""
+    return headers.get(name)
+
+
 async def _safe_aread_response(response: httpx.Response, timeout: float | None = None) -> bytes:
     """Safely read async response body, falling back to empty bytes on errors."""
     try:

litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py

@@ -49,6 +49,7 @@ from litellm.litellm_core_utils.url_utils import async_safe_get
 from litellm.llms.custom_httpx.http_handler import (
     AsyncHTTPHandler,
     get_async_httpx_client,
+    header_value,
     httpxSpecialProvider,
 )
 from litellm.proxy._experimental.mcp_server.tool_registry import (
@@ -457,7 +458,7 @@ def _raise_for_upstream_failure(
     if response.status_code == 401 and relays_upstream_auth:
         raise MCPUpstreamAuthError(
             status_code=response.status_code,
-            www_authenticate=dict(response.headers).get("www-authenticate"),
+            www_authenticate=header_value(response.headers, "www-authenticate"),
             server_name=upstream,
         )
     raise MCPOpenApiUpstreamError(response.status_code, upstream)

litellm/rag/ingestion/gemini_ingestion.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Final, cast
 from litellm._logging import verbose_logger
 from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
+    header_value,
     httpxSpecialProvider,
 )
 from litellm.llms.gemini.common_utils import GeminiModelInfo
@@ -277,7 +278,7 @@ class GeminiRAGIngestion(BaseRAGIngestion):
             raise Exception(error_msg)
         verbose_logger.debug("Initiate resumable upload response: %s", response.headers)
         # Extract upload URL from response headers
-        upload_url: Final = dict(response.headers).get("x-goog-upload-url")
+        upload_url: Final = header_value(response.headers, "x-goog-upload-url")
         if not upload_url:
             raise Exception("No upload URL returned in response headers")