Preserve OpenAPI HTTP error details

fastmcp_slim/fastmcp/server/providers/openapi/components.py

@@ -58,6 +58,21 @@ logger = get_logger(__name__)
 _DEFAULT_MIME_TYPE = "application/json"
 
 
+def _raise_for_status(response: httpx2.Response) -> None:
+    """Raise an OpenAPI-formatted error without relying on client exception types."""
+    if 200 <= response.status_code < 300:
+        return
+
+    error_message = f"HTTP error {response.status_code}: {response.reason_phrase}"
+    try:
+        error_data = response.json()
+        error_message += f" - {error_data}"
+    except (json.JSONDecodeError, ValueError):
+        if response.text:
+            error_message += f" - {response.text}"
+    raise ValueError(error_message)
+
+
 def _extract_mime_type_from_route(route: HTTPRoute) -> str:
     """Extract the primary MIME type from an HTTPRoute's response definitions.
 
@@ -202,7 +217,7 @@ class OpenAPITool(Tool):
             )
 
             response = await self._client.send(request)
-            response.raise_for_status()
+            _raise_for_status(response)
 
             # Try to parse as JSON first
             try:
@@ -229,18 +244,6 @@ class OpenAPITool(Tool):
             except json.JSONDecodeError:
                 return ToolResult(content=response.text)
 
-        except httpx2.HTTPStatusError as e:
-            error_message = (
-                f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
-            )
-            try:
-                error_data = e.response.json()
-                error_message += f" - {error_data}"
-            except (json.JSONDecodeError, ValueError):
-                if e.response.text:
-                    error_message += f" - {e.response.text}"
-            raise ValueError(error_message) from e
-
         except httpx2.TimeoutException as e:
             raise ValueError(f"HTTP request timed out ({type(e).__name__})") from e
 
@@ -303,7 +306,7 @@ class OpenAPIResource(Resource):
                 request.headers.update(mcp_headers)
 
             response = await self._client.send(request)
-            response.raise_for_status()
+            _raise_for_status(response)
 
             content_type = response.headers.get("content-type", "").lower()
 
@@ -331,18 +334,6 @@ class OpenAPIResource(Resource):
                     ]
                 )
 
-        except httpx2.HTTPStatusError as e:
-            error_message = (
-                f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
-            )
-            try:
-                error_data = e.response.json()
-                error_message += f" - {error_data}"
-            except (json.JSONDecodeError, ValueError):
-                if e.response.text:
-                    error_message += f" - {e.response.text}"
-            raise ValueError(error_message) from e
-
         except httpx2.TimeoutException as e:
             raise ValueError(f"HTTP request timed out ({type(e).__name__})") from e
 

tests/server/providers/openapi/test_comprehensive.py

@@ -653,13 +653,6 @@ class TestOpenAPIComprehensive:
         mock_response.json.return_value = {"code": 404, "message": "User not found"}
         mock_response.text = json.dumps({"code": 404, "message": "User not found"})
 
-        # Configure raise_for_status to raise HTTPStatusError
-        def raise_for_status():
-            raise httpx2.HTTPStatusError(
-                "404 Not Found", request=Mock(), response=mock_response
-            )
-
-        mock_response.raise_for_status = raise_for_status
         mock_client.send = AsyncMock(return_value=mock_response)
 
         server = create_openapi_server(

tests/server/providers/openapi/test_legacy_client_compat.py

@@ -3,6 +3,7 @@
 import pytest
 
 from fastmcp import Client, FastMCP, FastMCPDeprecationWarning
+from fastmcp.exceptions import ToolError
 
 httpx = pytest.importorskip("httpx", reason="legacy httpx not installed")
 
@@ -57,3 +58,22 @@ async def test_legacy_client_warns_and_remains_usable() -> None:
             result = await mcp_client.call_tool("list_items", {})
 
     assert result.structured_content == {"items": ["a", "b"]}
+
+
+async def test_legacy_client_preserves_http_error_details() -> None:
+    def handler(request: "httpx.Request") -> "httpx.Response":
+        return httpx.Response(404, json={"detail": "items not found"})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport,
+        base_url="https://api.example.com",
+    ) as client:
+        with pytest.warns(FastMCPDeprecationWarning):
+            server = FastMCP.from_openapi(SPEC, client=client)
+
+        async with Client(server) as mcp_client:
+            with pytest.raises(ToolError, match="HTTP error 404") as exc_info:
+                await mcp_client.call_tool("list_items", {})
+
+    assert "items not found" in str(exc_info.value)