[Fix] Resolve 7 mypy linting errors across 5 files

- realtime_api/main.py: Widen param types to accept both Dict and Pydantic models
- proxy/realtime_endpoints/endpoints.py: Widen return type to Union[..., Response]
- proxy/guardrails/guardrail_hooks/presidio.py: Add type:ignore[override] for bytes in streaming return
- proxy/_experimental/mcp_server/rest_endpoints.py: Annotate _oauth2_flow with Literal type
- proxy/management_endpoints/ui_sso.py: Add httpx import under TYPE_CHECKING, remove invalid timeout kwarg from AsyncHTTPHandler.get()

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

litellm/proxy/_experimental/mcp_server/rest_endpoints.py

@@ -1,6 +1,6 @@
 import importlib
 from datetime import datetime
-from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Union
+from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Set, Union
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
 
@@ -905,7 +905,7 @@ if MCP_AVAILABLE:
         try:
             client_id, client_secret, scopes = _extract_credentials(request)
 
-            _oauth2_flow = (
+            _oauth2_flow: Optional[Literal["client_credentials", "authorization_code"]] = (
                 "client_credentials"
                 if client_id and client_secret and request.token_url
                 else None

litellm/proxy/guardrails/guardrail_hooks/presidio.py

@@ -1234,9 +1234,13 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
         user_api_key_dict: UserAPIKeyAuth,
         response: Any,
         request_data: dict,
-    ) -> AsyncGenerator[Union[ModelResponseStream, bytes], None]:
+    ) -> AsyncGenerator[Union[ModelResponseStream, bytes], None]:  # type: ignore[override]
         """
         Process streaming response chunks to unmask PII tokens when needed.
+
+        Note: the return type includes `bytes` because Anthropic native SSE
+        streaming sends raw bytes chunks that pass through untransformed.
+        The base class declares ModelResponseStream only.
         """
         if self.apply_to_output:
             async for chunk in self._stream_apply_output_masking(

litellm/proxy/management_endpoints/ui_sso.py

@@ -17,6 +17,9 @@ import secrets
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Tuple, Union, cast
 
+if TYPE_CHECKING:
+    import httpx
+
 import jwt
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from fastapi.responses import RedirectResponse
@@ -2984,7 +2987,6 @@ class SSOAuthenticationHandler:
                         **additional_headers,
                         "Authorization": f"Bearer {access_token}",  # must not be overridden
                     },
-                    timeout=30.0,
                 )
                 if resp.status_code == 200:
                     try:

litellm/proxy/realtime_endpoints/endpoints.py

@@ -2,7 +2,7 @@
 
 import json
 import time
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Union
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
@@ -89,7 +89,7 @@ async def create_realtime_client_secret(
     request: Request,
     fastapi_response: Response,
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-) -> RealtimeClientSecretResponse:
+) -> Union[RealtimeClientSecretResponse, Response]:
     from litellm.proxy.proxy_server import (
         add_litellm_data_to_request,
         general_settings,

litellm/realtime_api/main.py

@@ -1,7 +1,7 @@
 """Abstraction function for OpenAI's realtime API"""
 
 import os
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional, Union, cast
 
 import litellm
 from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES, request_timeout
@@ -9,7 +9,12 @@ from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
 from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
 from litellm.secret_managers.main import get_secret_str
-from litellm.types.realtime import RealtimeClientSecretRequest, RealtimeQueryParams
+from litellm.types.realtime import (
+    RealtimeClientSecretRequest,
+    RealtimeExpiresAfter,
+    RealtimeQueryParams,
+    RealtimeSessionConfig,
+)
 from litellm.types.router import GenericLiteLLMParams
 from litellm.types.utils import LlmProviders
 from litellm.utils import ProviderConfigManager
@@ -93,8 +98,8 @@ def _get_realtime_http_provider_config(
 @wrapper_client
 async def acreate_realtime_client_secret(
     model: Optional[str] = None,
-    session: Optional[Dict[str, Any]] = None,
-    expires_after: Optional[Dict[str, Any]] = None,
+    session: Optional[Union[Dict[str, Any], RealtimeSessionConfig]] = None,
+    expires_after: Optional[Union[Dict[str, Any], RealtimeExpiresAfter]] = None,
     timeout: Optional[float] = None,
     **kwargs,
 ):