From ec5b9c2e40806dd076394dc1869f17895e7955ae Mon Sep 17 00:00:00 2001
From: stuxf <70670632+stuxf@users.noreply.github.com>
Date: Mon, 11 May 2026 18:56:25 -0700
Subject: [PATCH] chore(mcp): validate outbound OAuth URLs and fail closed on
 missing client IP (#27575)

Squash-merged by litellm-agent from stuxf's PR.
---
 litellm/llms/custom_httpx/http_handler.py     |  13 +-
 .../mcp_server/auth/user_api_key_auth_mcp.py  |   8 +-
 .../mcp_server/discoverable_endpoints.py      |  63 +++++++-
 .../_experimental/mcp_server/mcp_debug.py     |   4 +
 .../mcp_server/mcp_server_manager.py          | 129 +++++++++++++---
 .../mcp_server/rest_endpoints.py              |  36 ++++-
 .../proxy/_experimental/mcp_server/server.py  |  53 ++++---
 .../mcp_management_endpoints.py               |  31 +++-
 .../mcp/litellm_proxy_mcp_handler.py          |   9 +-
 .../llms/custom_httpx/test_http_handler.py    |  37 +++++
 .../mcp_server/test_discoverable_endpoints.py | 143 +++++++++++++++++-
 .../mcp_server/test_mcp_debug.py              |  65 ++++++++
 .../mcp_server/test_mcp_server.py             |   6 +
 .../mcp_server/test_mcp_server_manager.py     | 143 ++++++++++++++++++
 .../mcp_server/test_rest_endpoints.py         |   8 +-
 .../proxy/auth/test_mcp_ip_filtering.py       |  74 ++++++++-
 .../test_mcp_management_endpoints.py          |  61 ++++++++
 17 files changed, 807 insertions(+), 76 deletions(-)

diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py
index af18c666679..6d716bd9f52 100644
--- a/litellm/llms/custom_httpx/http_handler.py
+++ b/litellm/llms/custom_httpx/http_handler.py
@@ -611,6 +611,7 @@ class AsyncHTTPHandler:
         logging_obj: Optional[LiteLLMLoggingObject] = None,
         files: Optional[RequestFiles] = None,
         content: Any = None,
+        follow_redirects: Optional[bool] = None,
     ):
         start_time = time.time()
         try:
@@ -633,7 +634,10 @@ class AsyncHTTPHandler:
                 files=files,
                 content=request_content,
             )
-            response = await self.client.send(req, stream=stream)
+            send_kwargs: Dict[str, Any] = {"stream": stream}
+            if follow_redirects is not None:
+                send_kwargs["follow_redirects"] = follow_redirects
+            response = await self.client.send(req, **send_kwargs)
             response.raise_for_status()
             return response
         except (httpx.RemoteProtocolError, httpx.ConnectError):
@@ -650,6 +654,7 @@ class AsyncHTTPHandler:
                     params=params,
                     headers=headers,
                     stream=stream,
+                    follow_redirects=follow_redirects,
                 )
             finally:
                 await new_client.aclose()
@@ -853,6 +858,7 @@ class AsyncHTTPHandler:
         headers: Optional[dict] = None,
         stream: bool = False,
         content: Any = None,
+        follow_redirects: Optional[bool] = None,
     ):
         """
         Making POST request for a single connection client.
@@ -865,7 +871,10 @@ class AsyncHTTPHandler:
         req = client.build_request(
             "POST", url, data=request_data, json=json, params=params, headers=headers, content=request_content  # type: ignore
         )
-        response = await client.send(req, stream=stream)
+        send_kwargs: Dict[str, Any] = {"stream": stream}
+        if follow_redirects is not None:
+            send_kwargs["follow_redirects"] = follow_redirects
+        response = await client.send(req, **send_kwargs)
         response.raise_for_status()
         return response
 
diff --git a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
index a05af66118c..8af6ee98a5a 100644
--- a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
+++ b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
@@ -213,6 +213,7 @@ class MCPRequestHandler:
         # Inline imports avoid a circular dependency: mcp_server_manager imports
         # from this module.
         from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
             global_mcp_server_manager,
         )
         from litellm.types.mcp import MCPAuth
@@ -229,7 +230,12 @@ class MCPRequestHandler:
             return False
 
         for name in target_names:
-            server = global_mcp_server_manager.get_mcp_server_by_name(name)
+            # Metadata lookup ("does this server use OAuth2?") used to decide
+            # whether anonymous OAuth2 fallback is allowed for this path.
+            # Not an access check — the IP gate doesn't apply.
+            server = global_mcp_server_manager.get_mcp_server_by_name(
+                name, client_ip=INTERNAL_REQUEST
+            )
             if server is None or server.auth_type != MCPAuth.oauth2:
                 return False
         return True
diff --git a/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py b/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
index 62691641234..59707f240b9 100644
--- a/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
+++ b/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
@@ -1,11 +1,13 @@
 import json
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Literal, Optional
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 from fastapi import APIRouter, Form, HTTPException, Request
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
 
+import litellm
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.url_utils import SSRFError, validate_url
 from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
     httpxSpecialProvider,
@@ -25,6 +27,43 @@ from litellm.proxy.utils import get_server_root_path
 from litellm.types.mcp import MCPAuth
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
+
+def _validate_mcp_oauth_outbound_url(
+    url: str, role: Literal["token", "registration"]
+) -> tuple[str, str]:
+    """Validate an admin-configured OAuth URL before the proxy makes a request to
+    it. The /token, /register and similar endpoints are reachable without a
+    LiteLLM API key (they sit in the middle of an OAuth handshake), so an
+    unauthenticated caller can trigger the proxy to POST to whatever host the
+    admin configured. Resolve and validate the host before sending so an
+    operator who points an MCP server at an internal IdP doesn't unintentionally
+    expose it as an SSRF probe via the proxy."""
+    if not getattr(litellm, "user_url_validation", True):
+        parsed = urlparse(url)
+        host = parsed.hostname or ""
+        host_header = f"{host}:{parsed.port}" if parsed.port else host
+        return url, host_header
+    try:
+        return validate_url(url)
+    except SSRFError as exc:
+        # The /token and /register endpoints are reachable without an API key,
+        # so the error response goes to an unauthenticated caller. The raw
+        # SSRFError message includes the resolved IP — leaking it would tell
+        # the caller exactly which internal address the operator's IdP lives at,
+        # which is the reconnaissance the SSRF guard is meant to deny. Log the
+        # real reason for operators and return a generic message to the caller.
+        verbose_logger.warning(
+            "MCP OAuth %s URL blocked by SSRF validation: %s", role, exc
+        )
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Configured MCP {role} URL is not safe to call: "
+                "the destination resolves to a blocked address."
+            ),
+        )
+
+
 router = APIRouter(
     tags=["mcp"],
 )
@@ -393,10 +432,17 @@ async def exchange_token_with_server(
             token_data["code_verifier"] = code_verifier
 
     async_client = get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
+    target_url, host_header = _validate_mcp_oauth_outbound_url(
+        mcp_server.token_url, role="token"
+    )
+    # Disable redirect-following so a malicious 30x from the validated host
+    # can't bounce the proxy to an internal target — validate_url only
+    # checked the initial URL.
     response = await async_client.post(
-        mcp_server.token_url,
-        headers={"Accept": "application/json"},
+        target_url,
+        headers={"Accept": "application/json", "Host": host_header},
         data=token_data,
+        follow_redirects=False,
     )
 
     response.raise_for_status()
@@ -498,10 +544,17 @@ async def register_client_with_server(
     async_client = get_async_httpx_client(
         llm_provider=httpxSpecialProvider.Oauth2Register
     )
+    target_url, host_header = _validate_mcp_oauth_outbound_url(
+        mcp_server.registration_url, role="registration"
+    )
+    # Disable redirect-following so a malicious 30x from the validated host
+    # can't bounce the proxy to an internal target — validate_url only
+    # checked the initial URL.
     response = await async_client.post(
-        mcp_server.registration_url,
-        headers=headers,
+        target_url,
+        headers={**headers, "Host": host_header},
         json=register_data,
+        follow_redirects=False,
     )
     response.raise_for_status()
 
diff --git a/litellm/proxy/_experimental/mcp_server/mcp_debug.py b/litellm/proxy/_experimental/mcp_server/mcp_debug.py
index 254f208e231..6b9e47383b2 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_debug.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_debug.py
@@ -300,6 +300,10 @@ class MCPDebug:
         server_auth_type: Optional[str] = None
         auth_resolution = "no-auth"
 
+        # Pass client_ip through unchanged. When IP extraction fails the gate
+        # fails closed for internal-only servers, so the debug response stays
+        # empty rather than leaking outbound_url / server_auth_type metadata
+        # for internal-only servers named in the request.
         for server_name in mcp_servers or []:
             server = global_mcp_server_manager.get_mcp_server_by_name(
                 server_name, client_ip=client_ip
diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index 6ad731e7113..8dff40582f9 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -166,6 +166,21 @@ def _deserialize_json_dict(data: Any) -> Optional[Dict[str, str]]:
         return data
 
 
+class _InternalRequest:
+    """Sentinel passed in place of a client IP to indicate that the call site
+    is internal (background reload, registry maintenance, admin debug) and
+    should bypass IP-based access control. External request handlers must pass
+    a real IP string instead — passing ``None`` now fails closed."""
+
+    __slots__ = ()
+
+    def __repr__(self) -> str:  # pragma: no cover - trivial
+        return "INTERNAL_REQUEST"
+
+
+INTERNAL_REQUEST = _InternalRequest()
+
+
 class MCPServerManager:
     _STDIO_ENV_TEMPLATE_PATTERN = re.compile(r"^\$\{(X-[^}]+)\}$")
 
@@ -1045,28 +1060,56 @@ class MCPServerManager:
         return toolset
 
     def filter_server_ids_by_ip(
-        self, server_ids: List[str], client_ip: Optional[str]
+        self,
+        server_ids: List[str],
+        client_ip: Union[str, "_InternalRequest", None],
     ) -> List[str]:
         """
         Filter server IDs by client IP — external callers only see public servers.
 
-        Returns server_ids unchanged when client_ip is None (no filtering).
+        See ``filter_server_ids_by_ip_with_info`` for the contract on
+        ``client_ip``: ``None`` fails closed, ``INTERNAL_REQUEST`` bypasses
+        gating, real IPs apply the existing rules.
         """
         filtered, _ = self.filter_server_ids_by_ip_with_info(server_ids, client_ip)
         return filtered
 
+    def _resolve_unknown_client_ip(
+        self, client_ip: Union[str, "_InternalRequest", None]
+    ) -> Union[str, "_InternalRequest", None]:
+        """Promote unknown ``client_ip=None`` to ``INTERNAL_REQUEST`` when the
+        operator has explicitly opted in via
+        ``general_settings.mcp_allow_unknown_client_ip: true``. Lets
+        deployments behind ASGI middleware where ``request.client`` is
+        legitimately ``None`` opt back to the previous fail-open behavior."""
+        if client_ip is None:
+            general_settings = self._get_general_settings()
+            if general_settings.get("mcp_allow_unknown_client_ip", False):
+                return INTERNAL_REQUEST
+        return client_ip
+
     def filter_server_ids_by_ip_with_info(
-        self, server_ids: List[str], client_ip: Optional[str]
+        self,
+        server_ids: List[str],
+        client_ip: Union[str, "_InternalRequest", None],
     ) -> Tuple[List[str], int]:
         """
         Filter server IDs by client IP — external callers only see public servers.
 
-        Returns (filtered_ids, ip_blocked_count) where ip_blocked_count is the number
-        of servers that were blocked because the client IP is not allowed to access them.
-        Returns server_ids unchanged (with 0 blocked) when client_ip is None.
+        Returns (filtered_ids, ip_blocked_count) where ip_blocked_count is the
+        number of servers that were blocked because the client IP is not
+        allowed to access them. ``None`` fails closed: external request
+        handlers must extract a real IP via
+        ``IPAddressUtils.get_mcp_client_ip(request)``. Internal callers
+        (admin debug, registry maintenance) should pass
+        ``INTERNAL_REQUEST`` to bypass IP gating.
         """
-        if client_ip is None:
+        client_ip = self._resolve_unknown_client_ip(client_ip)
+        if client_ip is INTERNAL_REQUEST:
             return server_ids, 0
+        # Don't short-circuit on client_ip is None — public servers should
+        # still be reachable, only internal-only servers fail closed. The
+        # per-server gate enforces both halves.
         allowed = []
         blocked = 0
         for sid in server_ids:
@@ -3066,29 +3109,59 @@ class MCPServerManager:
             return {}
 
     def _is_server_accessible_from_ip(
-        self, server: MCPServer, client_ip: Optional[str]
+        self,
+        server: MCPServer,
+        client_ip: Union[str, _InternalRequest, None],
     ) -> bool:
         """
         Check if a server is accessible from the given client IP.
 
-        - If client_ip is None, no IP filtering is applied (internal callers).
-        - If the server has available_on_public_internet=True, it's always accessible.
-        - Otherwise, only internal/private IPs can access it.
+        - ``INTERNAL_REQUEST`` bypasses IP filtering (internal callers, admin
+          debug, registry maintenance).
+        - If the server has ``available_on_public_internet=True`` (or is in
+          ``litellm.public_mcp_servers``), it's always accessible — including
+          when ``client_ip`` is ``None``. Otherwise a request to a
+          legitimately-public OAuth endpoint behind ASGI middleware that
+          nulls ``request.client`` would 404 with no actionable diagnostic.
+        - For non-public servers, ``None`` fails closed: external request
+          handlers must extract a real IP via
+          ``IPAddressUtils.get_mcp_client_ip(request)``. Earlier behaviour
+          treated ``None`` as "no filter," which let external callers reach
+          internal-only servers when IP extraction silently failed. Operators
+          behind ASGI middleware or load-balancer setups where
+          ``request.client`` is legitimately ``None`` can opt back to the
+          previous fail-open behaviour by setting
+          ``general_settings.mcp_allow_unknown_client_ip: true``.
+        - For non-public servers with a real IP, only internal/private IPs
+          can access it.
         """
-        if client_ip is None:
+        client_ip = self._resolve_unknown_client_ip(client_ip)
+        if client_ip is INTERNAL_REQUEST:
             return True
+        # Public servers are reachable regardless of caller IP — and crucially
+        # even when IP extraction failed (client_ip is None). Otherwise a
+        # request to a legitimately-public OAuth endpoint behind an ASGI
+        # middleware that nulls request.client would 404 with no actionable
+        # diagnostic for operators.
         if server.available_on_public_internet:
             return True
         # Check backwards compat: litellm.public_mcp_servers
         public_ids = set(litellm.public_mcp_servers or [])
         if server.server_id in public_ids:
             return True
-        # Non-public server: only accessible from internal IPs
+        if client_ip is None:
+            # Non-public server with unknown caller IP fails closed. Earlier
+            # behaviour treated None as "no filter," letting external callers
+            # reach internal-only servers when IP extraction silently failed.
+            return False
+        # Non-public server: only accessible from internal IPs. The two
+        # early returns above narrow client_ip to ``str`` for the type
+        # checker; cast keeps mypy happy without a runtime assert.
         general_settings = self._get_general_settings()
         internal_networks = IPAddressUtils.parse_internal_networks(
             general_settings.get("mcp_internal_ip_ranges")
         )
-        return IPAddressUtils.is_internal_ip(client_ip, internal_networks)
+        return IPAddressUtils.is_internal_ip(cast(str, client_ip), internal_networks)
 
     def get_mcp_server_by_id(self, server_id: str) -> Optional[MCPServer]:
         """
@@ -3179,7 +3252,9 @@ class MCPServerManager:
         return result
 
     def get_mcp_server_by_name(
-        self, server_name: str, client_ip: Optional[str] = None
+        self,
+        server_name: str,
+        client_ip: Union[str, _InternalRequest, None] = None,
     ) -> Optional[MCPServer]:
         """
         Get the MCP Server from the server name.
@@ -3191,8 +3266,15 @@ class MCPServerManager:
 
         Args:
             server_name: The server name to look up.
-            client_ip: Optional client IP for access control. When provided,
-                       non-public servers are hidden from external IPs.
+            client_ip: External request IP for IP-based access control, or
+                       ``INTERNAL_REQUEST`` for internal callers (admin debug,
+                       registry maintenance) that intentionally bypass IP
+                       gating. ``None`` fails closed: external request handlers
+                       must extract a real IP via
+                       ``IPAddressUtils.get_mcp_client_ip(request)``. Earlier
+                       behaviour silently bypassed gating on ``None``, which
+                       let request handlers that forgot to pass an IP reach
+                       internal-only servers.
         """
         registry = self.get_registry()
         # Pass 1: Match by alias (highest priority)
@@ -3216,18 +3298,23 @@ class MCPServerManager:
         return None
 
     def get_filtered_registry(
-        self, client_ip: Optional[str] = None
+        self, client_ip: Union[str, "_InternalRequest", None] = None
     ) -> Dict[str, MCPServer]:
         """
         Get registry filtered by client IP access control.
 
         Args:
-            client_ip: Optional client IP. When provided, non-public servers
-                       are hidden from external IPs. When None, returns all servers.
+            client_ip: Real client IP (filter applied), ``INTERNAL_REQUEST``
+                       (return full registry, internal callers only), or
+                       ``None`` (fails closed: returns empty registry).
+                       External request handlers must pass a real IP.
         """
         registry = self.get_registry()
-        if client_ip is None:
+        client_ip = self._resolve_unknown_client_ip(client_ip)
+        if client_ip is INTERNAL_REQUEST:
             return registry
+        # Don't short-circuit on client_ip is None — public servers should
+        # still be reachable, only internal-only servers fail closed.
         return {
             k: v
             for k, v in registry.items()
diff --git a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
index 829863d2dbb..3515227799e 100644
--- a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
+++ b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
@@ -380,7 +380,9 @@ if MCP_AVAILABLE:
         # Resolve a server name to its UUID if needed
         _name_resolved = None
         if server_id not in allowed_server_ids:
-            _name_resolved = global_mcp_server_manager.get_mcp_server_by_name(server_id)
+            _name_resolved = global_mcp_server_manager.get_mcp_server_by_name(
+                server_id, client_ip=rest_client_ip
+            )
             if _name_resolved is not None and _name_resolved.server_id in set(
                 allowed_server_ids
             ):
@@ -393,11 +395,23 @@ if MCP_AVAILABLE:
             )
             if (
                 _server is not None
-                and rest_client_ip is not None
                 and not global_mcp_server_manager._is_server_accessible_from_ip(
                     _server, rest_client_ip
                 )
             ):
+                if rest_client_ip is None:
+                    raise HTTPException(
+                        status_code=403,
+                        detail={
+                            "error": "client_ip_unknown",
+                            "message": (
+                                "Cannot determine client IP for IP-based access "
+                                "control. If the proxy is behind a load balancer "
+                                "or reverse proxy, configure use_x_forwarded_for "
+                                "and mcp_trusted_proxy_ranges in general_settings."
+                            ),
+                        },
+                    )
                 raise HTTPException(
                     status_code=403,
                     detail={
@@ -470,7 +484,9 @@ if MCP_AVAILABLE:
         # Resolve a server name to its UUID if needed
         _name_resolved = None
         if server_id not in allowed_server_ids:
-            _name_resolved = global_mcp_server_manager.get_mcp_server_by_name(server_id)
+            _name_resolved = global_mcp_server_manager.get_mcp_server_by_name(
+                server_id, client_ip=rest_client_ip
+            )
             if _name_resolved is not None and _name_resolved.server_id in set(
                 allowed_server_ids
             ):
@@ -483,11 +499,23 @@ if MCP_AVAILABLE:
             )
             if (
                 _server is not None
-                and rest_client_ip is not None
                 and not global_mcp_server_manager._is_server_accessible_from_ip(
                     _server, rest_client_ip
                 )
             ):
+                if rest_client_ip is None:
+                    raise HTTPException(
+                        status_code=403,
+                        detail={
+                            "error": "client_ip_unknown",
+                            "message": (
+                                "Cannot determine client IP for IP-based access "
+                                "control. If the proxy is behind a load balancer "
+                                "or reverse proxy, configure use_x_forwarded_for "
+                                "and mcp_trusted_proxy_ranges in general_settings."
+                            ),
+                        },
+                    )
                 raise HTTPException(
                     status_code=403,
                     detail={
diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py
index 276a6e8a3bb..009b503976d 100644
--- a/litellm/proxy/_experimental/mcp_server/server.py
+++ b/litellm/proxy/_experimental/mcp_server/server.py
@@ -154,6 +154,7 @@ if MCP_AVAILABLE:
     )
     from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
         MCPServerManager,
+        _InternalRequest,
         global_mcp_server_manager,
     )
     from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import (
@@ -850,27 +851,24 @@ if MCP_AVAILABLE:
     async def _get_allowed_mcp_servers(
         user_api_key_auth: Optional[UserAPIKeyAuth],
         mcp_servers: Optional[List[str]],
-        client_ip: Optional[str] = None,
+        client_ip: Union[str, _InternalRequest, None] = None,
     ) -> List[MCPServer]:
         """Return allowed MCP servers for a request after applying filters.
 
         Args:
             user_api_key_auth: The authenticated user's API key info.
             mcp_servers: Optional list of server names to filter to.
-            client_ip: Client IP for IP-based access control. If None, falls back to
-                      auth context. Pass explicitly from request handlers for safety.
-        Note: If client_ip is None and auth context is not set, IP filtering is skipped.
-              This is intentional for internal callers but may indicate a bug if called
-              from a request handler without proper context setup.
+            client_ip: Client IP for IP-based access control. Pass an explicit
+                      string from external request handlers, the
+                      ``INTERNAL_REQUEST`` sentinel from internal callers
+                      (admin debug, registry maintenance, background work),
+                      or ``None`` to fall back to the auth-context IP.
+                      ``None`` after the auth-context fallback fails closed
+                      — an external request that can't be attributed to an IP
+                      will not reach internal-only servers.
         """
-        # Use explicit client_ip if provided, otherwise try auth context
         if client_ip is None:
             client_ip = _get_client_ip_from_context()
-            if client_ip is None:
-                verbose_logger.debug(
-                    "MCP _get_allowed_mcp_servers called without client_ip and no auth context. "
-                    "IP filtering will be skipped. This is expected for internal calls."
-                )
 
         allowed_mcp_server_ids = (
             await global_mcp_server_manager.get_allowed_mcp_servers(user_api_key_auth)
@@ -887,15 +885,28 @@ if MCP_AVAILABLE:
             allowed_mcp_server_ids,
         )
         if _ip_blocked > 0:
-            verbose_logger.debug(
-                "MCP IP filtering: %d server(s) are not accessible from client IP %s "
-                "because they are restricted to internal networks. "
-                "No tools from those servers will be returned. "
-                "To expose a server externally, set 'available_on_public_internet: true' "
-                "in its configuration.",
-                _ip_blocked,
-                client_ip,
-            )
+            if client_ip is None:
+                # IP extraction failed (no X-Forwarded-* header, missing
+                # trusted-proxy config, etc.). Fail-closed at the gate
+                # silently dropped the servers — tell the operator to fix
+                # IP forwarding, NOT to expose the server publicly.
+                verbose_logger.debug(
+                    "MCP IP filtering: %d server(s) hidden because client IP "
+                    "could not be determined for this request. Fix request-IP "
+                    "extraction (X-Forwarded-For + trusted_proxies) so the "
+                    "gate can evaluate access.",
+                    _ip_blocked,
+                )
+            else:
+                verbose_logger.debug(
+                    "MCP IP filtering: %d server(s) are not accessible from "
+                    "client IP %s because they are restricted to internal "
+                    "networks. No tools from those servers will be returned. "
+                    "To expose a server externally, set "
+                    "'available_on_public_internet: true' in its configuration.",
+                    _ip_blocked,
+                    client_ip,
+                )
         allowed_mcp_servers: List[MCPServer] = []
         for allowed_mcp_server_id in allowed_mcp_server_ids:
             mcp_server = global_mcp_server_manager.get_mcp_server_by_id(
diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py
index 7bda0f87ccd..b4d2427707d 100644
--- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py
+++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py
@@ -1557,14 +1557,35 @@ if MCP_AVAILABLE:
         if server is None:
             # Fall back to real DB/config server (e.g. for the user-side OAuth flow
             # which calls these endpoints with a real server_id, not a temp session id).
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                INTERNAL_REQUEST,
+            )
             from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 
-            client_ip = IPAddressUtils.get_mcp_client_ip(request) if request else None
-            server = global_mcp_server_manager.get_mcp_server_by_id(
-                server_id
-            ) or global_mcp_server_manager.get_mcp_server_by_name(
-                server_id, client_ip=client_ip
+            # Programmatic callers (no Request) bypass the IP gate explicitly.
+            # Real HTTP callers go through the normal extraction path.
+            client_ip = (
+                IPAddressUtils.get_mcp_client_ip(request)
+                if request
+                else INTERNAL_REQUEST
             )
+            # get_mcp_server_by_id alone does not apply IP gating, so an
+            # external caller hitting /server/oauth/{server_id}/{authorize,
+            # token,register} with the UUID of an internal-only server would
+            # bypass the IP restriction. Apply the gate to the id-lookup
+            # result before falling back to the name lookup (which gates).
+            server = global_mcp_server_manager.get_mcp_server_by_id(server_id)
+            if (
+                server is not None
+                and not global_mcp_server_manager._is_server_accessible_from_ip(
+                    server, client_ip
+                )
+            ):
+                server = None
+            if server is None:
+                server = global_mcp_server_manager.get_mcp_server_by_name(
+                    server_id, client_ip=client_ip
+                )
         if server is None:
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
diff --git a/litellm/responses/mcp/litellm_proxy_mcp_handler.py b/litellm/responses/mcp/litellm_proxy_mcp_handler.py
index 94cff6922b5..2e6a0668517 100644
--- a/litellm/responses/mcp/litellm_proxy_mcp_handler.py
+++ b/litellm/responses/mcp/litellm_proxy_mcp_handler.py
@@ -191,6 +191,7 @@ class LiteLLM_Proxy_MCP_Handler:
             List names of allowed MCP servers
         """
         from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
             global_mcp_server_manager,
         )
         from litellm.proxy._experimental.mcp_server.server import (
@@ -216,7 +217,13 @@ class LiteLLM_Proxy_MCP_Handler:
         resolved_mcp_servers: List[str] = []
         resolved_toolset_ids: List[str] = []
         for name in mcp_servers:
-            if not global_mcp_server_manager.get_mcp_server_by_name(name):
+            # Server-side disambiguation between MCP server names and toolset
+            # names. Access control is enforced downstream by
+            # _get_allowed_mcp_servers_from_mcp_server_names; this lookup is
+            # purely "is `name` a known server?", so bypass the IP gate.
+            if not global_mcp_server_manager.get_mcp_server_by_name(
+                name, client_ip=INTERNAL_REQUEST
+            ):
                 try:
                     from litellm.proxy.proxy_server import prisma_client
 
diff --git a/tests/test_litellm/llms/custom_httpx/test_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_http_handler.py
index 26f50e8e492..47170163392 100644
--- a/tests/test_litellm/llms/custom_httpx/test_http_handler.py
+++ b/tests/test_litellm/llms/custom_httpx/test_http_handler.py
@@ -798,3 +798,40 @@ def test_get_httpx_client_applies_httpx_timeout_object_without_mocking_handler()
         assert handler.client.timeout == t
     finally:
         handler.close()
+
+
+@pytest.mark.asyncio
+async def test_post_retry_propagates_follow_redirects():
+    """
+    AsyncHTTPHandler.post() retries connection errors via
+    single_connection_post_request(). When a caller passes
+    follow_redirects=False (e.g. the MCP OAuth SSRF guard), the retry
+    must honor it too — otherwise a transient connection error followed
+    by a 30x on retry can bypass the redirect block and hit an internal
+    address.
+    """
+    from unittest.mock import AsyncMock
+
+    handler = AsyncHTTPHandler()
+    try:
+        with (
+            patch.object(
+                handler.client,
+                "send",
+                side_effect=httpx.RemoteProtocolError("forced retry"),
+            ),
+            patch.object(
+                handler,
+                "single_connection_post_request",
+                new=AsyncMock(return_value=MagicMock(status_code=200)),
+            ) as mock_retry,
+        ):
+            await handler.post("https://example.com/token", follow_redirects=False)
+        assert mock_retry.await_count == 1
+        kwargs = mock_retry.await_args.kwargs
+        assert kwargs.get("follow_redirects") is False, (
+            "follow_redirects must reach the retry path so the SSRF "
+            "redirect block holds across reconnect"
+        )
+    finally:
+        await handler.close()
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py
index 581324d47d2..6e4c2305e1c 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py
@@ -6,19 +6,19 @@ import pytest
 from fastapi import HTTPException
 
 
-# Fixture to mock IP address check for all MCP tests
-# This prevents tests from failing due to IP-based access control
+# Fixture to bypass MCP IP-based access control for all OAuth flow tests.
+# Mock requests don't carry a real client IP context; the bypass uses the
+# explicit INTERNAL_REQUEST sentinel because passing None now fails closed
+# in the gate function (see MCPServerManager._is_server_accessible_from_ip).
 @pytest.fixture(autouse=True)
 def mock_mcp_client_ip():
-    """Mock IPAddressUtils.get_mcp_client_ip to return None for all tests.
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        INTERNAL_REQUEST,
+    )
 
-    This bypasses IP-based access control in tests, since the MCP server's
-    available_on_public_internet defaults to False and mock requests don't
-    have proper client IP context.
-    """
     with patch(
         "litellm.proxy._experimental.mcp_server.discoverable_endpoints.IPAddressUtils.get_mcp_client_ip",
-        return_value=None,
+        return_value=INTERNAL_REQUEST,
     ):
         yield
 
@@ -91,6 +91,7 @@ async def test_authorize_endpoint_includes_response_type():
 
     # Mock request
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -156,6 +157,7 @@ async def test_authorize_endpoint_preserves_existing_query_params():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -223,6 +225,7 @@ async def test_authorize_endpoint_forwards_pkce_parameters():
 
     # Mock request
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm-proxy.example.com/"
     mock_request.headers = {}
 
@@ -294,6 +297,7 @@ async def test_token_endpoint_forwards_code_verifier():
 
     # Mock request
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm-proxy.example.com/"
     mock_request.headers = {}
 
@@ -371,6 +375,7 @@ async def test_register_client_without_mcp_server_name_returns_dummy():
     global_mcp_server_manager.registry.clear()
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://proxy.litellm.example/"
     mock_request.headers = {}
     with patch(
@@ -419,6 +424,7 @@ async def test_register_client_returns_existing_server_credentials():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://proxy.litellm.example/"
     mock_request.headers = {}
 
@@ -474,6 +480,7 @@ async def test_register_client_remote_registration_success():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://proxy.litellm.example/"
     mock_request.headers = {}
 
@@ -503,6 +510,13 @@ async def test_register_client_remote_registration_success():
                 "litellm.proxy._experimental.mcp_server.discoverable_endpoints.get_async_httpx_client",
                 return_value=mock_async_client,
             ),
+            patch(
+                "litellm.proxy._experimental.mcp_server.discoverable_endpoints.validate_url",
+                return_value=(
+                    "https://provider.example/oauth/register",
+                    "provider.example",
+                ),
+            ),
         ):
             response = await register_client(
                 request=mock_request, mcp_server_name=oauth2_server.server_name
@@ -522,6 +536,7 @@ async def test_register_client_remote_registration_success():
     assert call_args.kwargs["headers"] == {
         "Content-Type": "application/json",
         "Accept": "application/json",
+        "Host": "provider.example",
     }
     assert call_args.kwargs["json"]["redirect_uris"] == [
         "https://proxy.litellm.example/callback"
@@ -573,6 +588,7 @@ async def test_authorize_endpoint_respects_x_forwarded_proto():
 
     # Mock request with http base_url but X-Forwarded-Proto: https
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://litellm.example.com/"  # HTTP
     mock_request.headers = {"X-Forwarded-Proto": "https"}  # Behind HTTPS proxy
 
@@ -641,6 +657,7 @@ async def test_token_endpoint_respects_x_forwarded_proto():
 
     # Mock request with http base_url but X-Forwarded-Proto: https
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://litellm-proxy.example.com/"  # HTTP
     mock_request.headers = {"X-Forwarded-Proto": "https"}  # Behind HTTPS proxy
 
@@ -719,6 +736,7 @@ async def test_oauth_protected_resource_respects_x_forwarded_proto():
 
     # Mock request with http base_url but X-Forwarded-Proto: https
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://litellm.example.com/"  # HTTP
     mock_request.headers = {"X-Forwarded-Proto": "https"}  # Behind HTTPS proxy
 
@@ -774,6 +792,7 @@ async def test_oauth_authorization_server_respects_x_forwarded_proto():
 
     # Mock request with http base_url but X-Forwarded-Proto: https
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://litellm.example.com/"  # HTTP
     mock_request.headers = {"X-Forwarded-Proto": "https"}  # Behind HTTPS proxy
 
@@ -812,6 +831,7 @@ async def test_register_client_respects_x_forwarded_proto():
 
     # Mock request with http base_url but X-Forwarded-Proto: https
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://proxy.litellm.example/"  # HTTP
     mock_request.headers = {"X-Forwarded-Proto": "https"}  # Behind HTTPS proxy
 
@@ -871,6 +891,7 @@ async def test_authorize_endpoint_respects_x_forwarded_host():
     # Internal: http://localhost:8888/github/mcp
     # External: https://proxy.example.com/github/mcp
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://localhost:8888/github/mcp"
     mock_request.headers = {
         "X-Forwarded-Proto": "https",
@@ -943,6 +964,7 @@ async def test_token_endpoint_respects_x_forwarded_host():
 
     # Mock request simulating nginx proxy without port in host
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://localhost:8888/github/mcp"
     mock_request.headers = {
         "X-Forwarded-Proto": "https",
@@ -1123,6 +1145,7 @@ def test_get_request_base_url_comprehensive(
         pytest.skip("MCP discoverable endpoints not available")
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = base_url
 
     headers = {}
@@ -1209,6 +1232,7 @@ def test_get_request_base_url_xff_trust_gate(
         pytest.skip("MCP discoverable endpoints not available")
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://localhost:4000/"
     mock_request.client = MagicMock()
     mock_request.client.host = direct_ip
@@ -1253,6 +1277,7 @@ def test_xff_misconfig_warning_emitted_once(caplog):
     ip_address_utils._warned_xff_without_trusted_ranges = False
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "http://localhost:4000/"
     mock_request.client = MagicMock()
     mock_request.client.host = "203.0.113.5"
@@ -1323,6 +1348,7 @@ async def test_oauth_protected_resource_returns_empty_scopes_when_none():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -1377,6 +1403,7 @@ async def test_oauth_authorization_server_returns_empty_scopes_when_none():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -1445,6 +1472,7 @@ async def test_authorize_root_resolves_single_oauth2_server():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1498,6 +1526,7 @@ async def test_authorize_root_fails_with_multiple_oauth2_servers():
     global_mcp_server_manager.registry[server2.server_id] = server2
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1536,6 +1565,7 @@ async def test_authorize_root_does_not_resolve_private_server_for_external_clien
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1577,6 +1607,7 @@ async def test_token_root_resolves_single_oauth2_server():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1642,6 +1673,7 @@ async def test_token_root_does_not_resolve_private_server_for_external_client():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1686,6 +1718,7 @@ async def test_register_root_resolves_single_oauth2_server():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1723,6 +1756,7 @@ async def test_register_root_does_not_resolve_private_server_for_external_client
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1765,6 +1799,7 @@ async def test_discovery_root_includes_server_name_prefix():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -1805,6 +1840,7 @@ async def test_discovery_root_does_not_expose_private_server_for_external_client
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
@@ -2006,6 +2042,7 @@ async def test_oauth_authorize_includes_scopes_from_server_config():
     )
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -2064,6 +2101,7 @@ async def test_oauth_authorize_prefers_request_scope_over_server_config():
     )
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -2133,6 +2171,7 @@ async def test_token_endpoint_refresh_token_grant():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://proxy.litellm.example/"
     mock_request.headers = {}
 
@@ -2267,6 +2306,7 @@ async def test_authorize_endpoint_rejects_non_loopback_redirect_uri():
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -2314,6 +2354,7 @@ async def test_authorize_endpoint_accepts_ipv4_loopback_range_and_ipv6_full_form
     global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
 
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -2443,6 +2484,7 @@ async def test_token_endpoint_sets_no_store_cache_control():
         token_url="https://provider.com/oauth/token",
     )
     mock_request = MagicMock(spec=Request)
+    mock_request.client = MagicMock(host="127.0.0.1")
     mock_request.base_url = "https://litellm.example.com/"
     mock_request.headers = {}
 
@@ -2473,3 +2515,88 @@ async def test_token_endpoint_sets_no_store_cache_control():
 
     assert response.headers["cache-control"] == "no-store"
     assert response.headers["pragma"] == "no-cache"
+
+
+class TestOutboundOAuthURLValidation:
+    """
+    /token and /register can be hit without a LiteLLM API key (the caller is
+    mid-OAuth-handshake). They forward the request to whatever URL the admin
+    configured for the MCP server. Lock the new contract: the outbound URL is
+    SSRF-validated via litellm_core_utils.url_utils.validate_url before any
+    proxy POST, so an admin-configured internal IdP can't be probed by an
+    unauthenticated caller through the proxy.
+    """
+
+    def test_validator_rejects_internal_url(self):
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_mcp_oauth_outbound_url(
+                "http://127.0.0.1:8080/token", role="token"
+            )
+        assert exc_info.value.status_code == 400
+        detail = str(exc_info.value.detail)
+        assert "token" in detail
+        # The /token endpoint is unauthenticated — leaking the resolved IP
+        # would hand reconnaissance to the caller. Detail must stay generic.
+        assert "127.0.0.1" not in detail
+        assert "blocked address" in detail
+
+    def test_validator_rejects_rfc1918_url(self):
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_mcp_oauth_outbound_url(
+                "http://192.168.1.10/oauth/register", role="registration"
+            )
+        detail = str(exc_info.value.detail)
+        assert "192.168.1.10" not in detail
+        assert "registration" in detail
+
+    def test_validator_passes_when_validation_disabled(self):
+        # Operators who explicitly opt out via litellm.user_url_validation = False
+        # still get the URL passed through unchanged.
+        import litellm
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        original = getattr(litellm, "user_url_validation", True)
+        try:
+            litellm.user_url_validation = False
+            url, host = _validate_mcp_oauth_outbound_url(
+                "http://10.0.0.1:5000/token", role="token"
+            )
+            assert url == "http://10.0.0.1:5000/token"
+            # Host header includes the explicit port from the URL.
+            assert host == "10.0.0.1:5000"
+
+            # When the URL omits the port, the Host header omits it too.
+            url2, host2 = _validate_mcp_oauth_outbound_url(
+                "http://10.0.0.1/token", role="token"
+            )
+            assert host2 == "10.0.0.1"
+        finally:
+            litellm.user_url_validation = original
+
+    def test_validator_accepts_public_url(self):
+        # The public-URL acceptance path does live DNS resolution, which is
+        # flaky in CI. Patch validate_url to confirm the helper passes the
+        # validated tuple through.
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        with patch(
+            "litellm.proxy._experimental.mcp_server.discoverable_endpoints.validate_url",
+            return_value=("https://provider.example/token", "provider.example"),
+        ):
+            url, host = _validate_mcp_oauth_outbound_url(
+                "https://provider.example/token", role="token"
+            )
+        assert url == "https://provider.example/token"
+        assert host == "provider.example"
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_debug.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_debug.py
index 468bd946ae9..bf7232e4d2b 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_debug.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_debug.py
@@ -265,3 +265,68 @@ class TestWrapSendWithDebugHeaders:
         asyncio.run(wrapped(body_msg))
 
         assert captured[0] == body_msg
+
+
+class TestMaybeBuildDebugHeadersIPGate:
+    """
+    The maybe_build_debug_headers entry point used to promote a missing
+    client_ip to INTERNAL_REQUEST so debug metadata wasn't silently dropped.
+    That bypass let an external caller (in deployments where IP extraction
+    fails) read x-mcp-debug-outbound-url / x-mcp-debug-server-auth-type for
+    internal-only servers named in the request — the IP gate would
+    otherwise hide them. Lock the fixed contract: pass client_ip straight
+    through so the gate fails closed for internal-only servers when IP
+    extraction fails.
+    """
+
+    def _debug_headers_for(self, client_ip, *, server_is_internal_only=True):
+        from unittest.mock import patch
+
+        server = MagicMock()
+        server.url = "https://internal.example/mcp"
+        server.auth_type = "oauth2"
+        server.alias = "alias"
+        server.server_name = "server"
+        server.has_client_credentials = False
+        server.authentication_token = None
+
+        manager = MagicMock()
+
+        # The fix forwards client_ip to get_mcp_server_by_name; we mirror the
+        # gate's contract: None on an internal-only server returns None,
+        # while a real internal IP returns the server.
+        def _by_name(name, client_ip):
+            if server_is_internal_only and client_ip is None:
+                return None
+            return server
+
+        manager.get_mcp_server_by_name.side_effect = _by_name
+
+        with patch(
+            "litellm.proxy._experimental.mcp_server.mcp_server_manager."
+            "global_mcp_server_manager",
+            manager,
+        ):
+            return MCPDebug.maybe_build_debug_headers(
+                raw_headers={MCP_DEBUG_REQUEST_HEADER: "true"},
+                scope={"type": "http", "headers": []},
+                mcp_servers=["internal-only-server"],
+                mcp_auth_header=None,
+                mcp_server_auth_headers=None,
+                oauth2_headers=None,
+                client_ip=client_ip,
+            )
+
+    def test_unknown_ip_does_not_leak_internal_server_metadata(self):
+        headers = self._debug_headers_for(client_ip=None, server_is_internal_only=True)
+        # Outbound URL must NOT echo the internal-only server's URL.
+        assert headers.get("x-mcp-debug-outbound-url") == "(unknown)"
+        assert headers.get("x-mcp-debug-server-auth-type") == "(none)"
+        assert headers.get("x-mcp-debug-auth-resolution") == "no-auth"
+
+    def test_internal_ip_still_resolves_metadata(self):
+        headers = self._debug_headers_for(
+            client_ip="10.0.0.1", server_is_internal_only=True
+        )
+        assert headers.get("x-mcp-debug-outbound-url") == "https://internal.example/mcp"
+        assert headers.get("x-mcp-debug-server-auth-type") == "oauth2"
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
index a7649502bde..cd759dd73ad 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
@@ -1154,6 +1154,12 @@ async def test_mcp_routing_with_conflicting_alias_and_group_name():
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager._get_tools_from_server",
             mock_get_tools_spy,
         ),
+        # Unit test has no HTTP request context — give the IP gate a real
+        # internal IP so it doesn't fail closed and filter every server out.
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_client_ip_from_context",
+            return_value="127.0.0.1",
+        ),
     ):
         mcp_servers_from_path = _get_mcp_servers_in_path(test_path)
 
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index 11e9dbbdd57..08afb975d29 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -3311,5 +3311,148 @@ class TestOAuthDiscoverySSRFGuard:
         mock_client.get.assert_not_called()
 
 
+class TestIPGatingFailClosed:
+    """
+    The IP-gating helper used to fail open when ``client_ip`` was ``None``,
+    which let request handlers that couldn't determine a client IP reach
+    internal-only MCP servers. Lock the new contract: ``None`` fails closed,
+    ``INTERNAL_REQUEST`` bypasses gating, real IPs apply the existing rules.
+    """
+
+    def _internal_server(self):
+        from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+        return MCPServer(
+            server_id="internal-1",
+            name="internal",
+            server_name="internal",
+            transport=MCPTransport.http,
+            url="https://internal.local/mcp",
+            available_on_public_internet=False,
+        )
+
+    def _public_server(self):
+        from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+        return MCPServer(
+            server_id="public-1",
+            name="public",
+            server_name="public",
+            transport=MCPTransport.http,
+            url="https://public.example/mcp",
+            available_on_public_internet=True,
+        )
+
+    @pytest.mark.parametrize(
+        "server_kind,client_ip_kind,expected",
+        [
+            # None fails closed for non-public servers — missing IP signals
+            # an external request that couldn't be attributed and must not
+            # reach internal-only servers.
+            ("internal", "none", False),
+            # None still lets public servers through — otherwise a request
+            # to a public OAuth endpoint behind ASGI middleware that nulls
+            # request.client would 404 with no actionable diagnostic.
+            ("public", "none", True),
+            # INTERNAL_REQUEST bypasses gating for both visibilities.
+            ("internal", "internal_request", True),
+            ("public", "internal_request", True),
+            # Real external IP applies the existing visibility rules.
+            ("internal", "external", False),
+            ("public", "external", True),
+        ],
+    )
+    def test_gate_contract(self, server_kind, client_ip_kind, expected):
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        server = (
+            self._internal_server()
+            if server_kind == "internal"
+            else self._public_server()
+        )
+        client_ip = {
+            "none": None,
+            "internal_request": INTERNAL_REQUEST,
+            "external": "8.8.8.8",
+        }[client_ip_kind]
+        assert manager._is_server_accessible_from_ip(server, client_ip) is expected
+
+    def test_get_filtered_registry_fails_closed_on_none(self):
+        # The wrapper used to return the full registry on None client_ip,
+        # which let unauth callers (e.g. _resolve_oauth2_server_for_root_endpoints
+        # when get_mcp_client_ip returned None) auto-select internal-only
+        # servers. Now fails closed; INTERNAL_REQUEST is the explicit bypass.
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        server = self._internal_server()
+        manager.registry[server.server_id] = server
+
+        assert manager.get_filtered_registry() == {}
+        assert manager.get_filtered_registry(client_ip=None) == {}
+        assert manager.get_filtered_registry(client_ip="8.8.8.8") == {}
+        assert manager.get_filtered_registry(client_ip=INTERNAL_REQUEST) == {
+            server.server_id: server
+        }
+
+    def test_filter_server_ids_by_ip_fails_closed_on_none(self):
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        server = self._internal_server()
+        manager.registry[server.server_id] = server
+        sids = [server.server_id]
+
+        # None fails closed: zero allowed, all reported as blocked.
+        allowed, blocked = manager.filter_server_ids_by_ip_with_info(sids, None)
+        assert (allowed, blocked) == ([], 1)
+
+        # External IP applies filter — internal server blocked.
+        allowed, blocked = manager.filter_server_ids_by_ip_with_info(sids, "8.8.8.8")
+        assert (allowed, blocked) == ([], 1)
+
+        # INTERNAL_REQUEST bypasses gating.
+        allowed, blocked = manager.filter_server_ids_by_ip_with_info(
+            sids, INTERNAL_REQUEST
+        )
+        assert (allowed, blocked) == (sids, 0)
+
+    def test_get_mcp_server_by_name_fails_closed_on_none(self):
+        # External request handlers must extract a real client IP. Passing
+        # None silently bypassed gating before; now it fails closed.
+        # Internal callers (admin debug, registry maintenance) must use
+        # INTERNAL_REQUEST explicitly.
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        server = self._internal_server()
+        manager.registry[server.server_id] = server
+
+        # Default client_ip is None — fails closed for internal-only servers.
+        result = manager.get_mcp_server_by_name("internal")
+        assert result is None
+
+        # External IP can't reach a non-public server.
+        result = manager.get_mcp_server_by_name("internal", client_ip="8.8.8.8")
+        assert result is None
+
+        # INTERNAL_REQUEST sentinel bypasses gating for internal callers.
+        result = manager.get_mcp_server_by_name("internal", client_ip=INTERNAL_REQUEST)
+        assert result is server
+
+
 if __name__ == "__main__":
     pytest.main([__file__])
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_rest_endpoints.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_rest_endpoints.py
index f4feac68fcc..3f171390497 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_rest_endpoints.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_rest_endpoints.py
@@ -595,7 +595,9 @@ class TestListToolsRestAPI:
         monkeypatch.setattr(
             rest_endpoints.global_mcp_server_manager,
             "get_mcp_server_by_name",
-            lambda name: stub_server if name == "my-server" else None,
+            lambda name, client_ip=None, **kwargs: (
+                stub_server if name == "my-server" else None
+            ),
             raising=False,
         )
         monkeypatch.setattr(
@@ -658,7 +660,9 @@ class TestListToolsRestAPI:
         monkeypatch.setattr(
             rest_endpoints.global_mcp_server_manager,
             "get_mcp_server_by_name",
-            lambda name: stub_server if name == "restricted-server" else None,
+            lambda name, client_ip=None, **kwargs: (
+                stub_server if name == "restricted-server" else None
+            ),
             raising=False,
         )
         monkeypatch.setattr(
diff --git a/tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py b/tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py
index 3b13ef3641f..078e68ecc8d 100644
--- a/tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py
+++ b/tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py
@@ -85,12 +85,49 @@ class TestMCPServerIPFiltering:
 
     @patch("litellm.public_mcp_servers", [])
     @patch("litellm.proxy.proxy_server.general_settings", {})
-    def test_no_ip_means_no_filtering(self):
+    def test_no_ip_fails_closed(self):
+        # Missing client_ip on an external request fails closed; internal
+        # callers must pass INTERNAL_REQUEST explicitly to bypass IP gating.
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+        )
+
         priv = _make_server("priv", available_on_public_internet=False)
         manager = _make_manager([priv])
 
-        result = manager.filter_server_ids_by_ip(["priv"], client_ip=None)
-        assert result == ["priv"]
+        assert manager.filter_server_ids_by_ip(["priv"], client_ip=None) == []
+        assert manager.filter_server_ids_by_ip(
+            ["priv"], client_ip=INTERNAL_REQUEST
+        ) == ["priv"]
+
+    @patch("litellm.public_mcp_servers", [])
+    @patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"mcp_allow_unknown_client_ip": True},
+    )
+    def test_unknown_client_ip_opt_out_restores_fail_open(self):
+        # Operators behind ASGI middleware where request.client is legitimately
+        # None can opt back to the pre-fix fail-open behavior with explicit
+        # consent via general_settings.mcp_allow_unknown_client_ip: true.
+        priv = _make_server("priv", available_on_public_internet=False)
+        manager = _make_manager([priv])
+
+        assert manager.filter_server_ids_by_ip(["priv"], client_ip=None) == ["priv"]
+
+    @patch("litellm.public_mcp_servers", [])
+    @patch("litellm.proxy.proxy_server.general_settings", {})
+    def test_public_server_reachable_when_ip_unknown(self):
+        # A legitimately-public server must stay reachable even when IP
+        # extraction failed (client_ip=None). Otherwise a request to a
+        # public OAuth endpoint behind ASGI middleware that nulls
+        # request.client would 404 with no actionable diagnostic.
+        pub = _make_server("pub", available_on_public_internet=True)
+        priv = _make_server("priv", available_on_public_internet=False)
+        manager = _make_manager([pub, priv])
+
+        assert manager.filter_server_ids_by_ip(["pub", "priv"], client_ip=None) == [
+            "pub"
+        ]
 
 
 class TestFilterServerIdsByIpWithInfo:
@@ -124,15 +161,40 @@ class TestFilterServerIdsByIpWithInfo:
 
     @patch("litellm.public_mcp_servers", [])
     @patch("litellm.proxy.proxy_server.general_settings", {})
-    def test_no_ip_returns_all_with_zero_blocked(self):
+    def test_no_ip_fails_closed_reports_all_blocked(self):
+        # Missing client_ip on an external request fails closed: all servers
+        # are reported blocked. Internal callers pass INTERNAL_REQUEST.
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+        )
+
         priv = _make_server("priv", available_on_public_internet=False)
         manager = _make_manager([priv])
 
         allowed, blocked = manager.filter_server_ids_by_ip_with_info(
             ["priv"], client_ip=None
         )
-        assert allowed == ["priv"]
-        assert blocked == 0
+        assert (allowed, blocked) == ([], 1)
+
+        allowed, blocked = manager.filter_server_ids_by_ip_with_info(
+            ["priv"], client_ip=INTERNAL_REQUEST
+        )
+        assert (allowed, blocked) == (["priv"], 0)
+
+    @patch("litellm.public_mcp_servers", [])
+    @patch("litellm.proxy.proxy_server.general_settings", {})
+    def test_no_ip_lets_public_through_blocks_private(self):
+        # When IP extraction fails (client_ip=None), public servers stay
+        # reachable; only non-public servers are blocked.
+        pub = _make_server("pub", available_on_public_internet=True)
+        priv = _make_server("priv", available_on_public_internet=False)
+        manager = _make_manager([pub, priv])
+
+        allowed, blocked = manager.filter_server_ids_by_ip_with_info(
+            ["pub", "priv"], client_ip=None
+        )
+        assert allowed == ["pub"]
+        assert blocked == 1
 
     @patch("litellm.public_mcp_servers", [])
     @patch("litellm.proxy.proxy_server.general_settings", {})
diff --git a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py
index f0909afcbf6..aa28d4212f5 100644
--- a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py
+++ b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py
@@ -1365,6 +1365,67 @@ class TestTemporaryMCPSessionEndpoints:
 
         assert exc_info.value.status_code == 404
 
+    @pytest.mark.asyncio
+    async def test_get_cached_temporary_mcp_server_id_lookup_applies_ip_gate(self):
+        """
+        /server/oauth/{server_id}/{authorize,token,register} resolves the
+        server via _get_cached_temporary_mcp_server_or_404, which used to do
+        get_mcp_server_by_id(server_id) OR get_mcp_server_by_name(...). The
+        id-lookup branch did not apply IP gating, letting an external caller
+        hit an internal-only server by UUID. The fix gates the id-lookup
+        result before falling back to name lookup.
+        """
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            _InternalRequest,
+        )
+        from litellm.proxy.management_endpoints.mcp_management_endpoints import (
+            _get_cached_temporary_mcp_server_or_404,
+        )
+
+        internal_only_server = generate_mock_mcp_server_config_record(
+            server_id="internal-only", name="Internal Only"
+        )
+        internal_only_server.available_on_public_internet = False
+        admin_auth = generate_mock_user_api_key_auth(
+            user_role=LitellmUserRoles.PROXY_ADMIN,
+        )
+
+        mock_manager = MagicMock()
+        mock_manager.get_mcp_server_by_id.return_value = internal_only_server
+        mock_manager.get_mcp_server_by_name.return_value = None
+
+        # External IP: gate denies, both id and name lookups should miss → 404.
+        def _gate(server, client_ip):
+            if isinstance(client_ip, _InternalRequest):
+                return True
+            return False  # external IP, server is internal-only
+
+        mock_manager._is_server_accessible_from_ip.side_effect = _gate
+        external_request = _make_mock_request(ip="8.8.8.8")
+
+        with (
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.get_cached_temporary_mcp_server",
+                return_value=None,
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager",
+                mock_manager,
+            ),
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                await _get_cached_temporary_mcp_server_or_404(
+                    "internal-only", admin_auth, request=external_request
+                )
+
+        assert (
+            exc_info.value.status_code == 404
+        ), "External caller must NOT be able to reach internal-only server by UUID"
+        # Gate must have been consulted on the id-lookup result.
+        mock_manager._is_server_accessible_from_ip.assert_any_call(
+            internal_only_server, "8.8.8.8"
+        )
+
     @pytest.mark.asyncio
     async def test_get_cached_temporary_mcp_server_non_admin_denied(self):
         """Non-admin without access to the server gets 403, not the server."""