Narrow proxy metadata validation fallback

🤖 Generated with OpenAI Codex

fastmcp_slim/fastmcp/server/providers/proxy.py

@@ -1194,7 +1194,11 @@ class ProxyMetadataMiddleware(Middleware):
                 and result_type not in mcp_types.CORE_RESULT_TYPES
             ):
                 return None
-            result = mcp_types.DiscoverResult.model_validate(raw)
+            try:
+                result = mcp_types.DiscoverResult.model_validate(raw)
+            except ValidationError as error:
+                logger.debug("Could not read upstream server metadata: %r", error)
+                return None
             return _UpstreamServerMetadata.from_discover(result)
         return _UpstreamServerMetadata.from_client(client)
 
@@ -1209,7 +1213,7 @@ class ProxyMetadataMiddleware(Middleware):
                 return await self._read_connected(client)
             async with client:
                 return await self._read_connected(client)
-        except (MCPError, ValidationError, *_PROXY_TRANSPORT_ERRORS) as error:
+        except (MCPError, *_PROXY_TRANSPORT_ERRORS) as error:
             if isinstance(error, RuntimeError) and not _has_transport_cause(error):
                 raise
             logger.debug("Could not read upstream server metadata: %r", error)

tests/server/providers/proxy/test_server_metadata.py

@@ -374,6 +374,24 @@ async def test_invalid_backend_client_negotiation_is_not_ignored():
             pass
 
 
+async def test_unrelated_client_validation_error_is_not_ignored():
+    class InvalidClient(ProxyClient):
+        async def __aenter__(self) -> ProxyClient:
+            mcp_types.Implementation.model_validate({})
+            return self
+
+    provider = ProxyProvider(lambda: InvalidClient(make_upstream()))
+    gateway = FastMCP(
+        "gateway",
+        providers=[provider],
+        middleware=[ProxyMetadataMiddleware(provider)],
+    )
+
+    with pytest.raises(MCPError):
+        async with Client(gateway, mode="auto"):
+            pass
+
+
 @pytest.mark.parametrize("mode", ["legacy", "auto"])
 async def test_forwarded_metadata_does_not_alias_connected_backend(mode: str):
     backend_info = mcp_types.Implementation(name="shared-backend", version="1.0")