ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Preserve legacy httpx compatibility without importing it (#4766)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-08-05 18:09:16 -04:00 • committed by GitHub
commit 875e8e18bd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 257 additions and 249 deletions
Download patch file Download diff file Expand all files Collapse all files

7
tests/server/providers/openapi/test_comprehensive.py
View file

@ -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(

168
tests/server/providers/openapi/test_legacy_client_compat.py
View file

@ -1,20 +1,9 @@
"""Legacy-httpx client compatibility for the OpenAPI integration.
The upgrade guide promises that an existing legacy ``httpx.AsyncClient`` passed
to ``OpenAPIProvider``/``FastMCP.from_openapi`` keeps working via duck-typing.
That requires two things of the OpenAPI request path: requests must be built
through the user's own client (``build_request``), and errors raised by that
client — which are legacy-httpx exceptions, not httpx2 — must still receive the
integration's specific error formatting rather than surfacing as generic
failures.
"""
"""Deprecation bridge for legacy-httpx OpenAPI clients."""
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp import Client, FastMCP, FastMCPDeprecationWarning
from fastmcp.exceptions import ToolError
from fastmcp.server.providers.openapi import OpenAPIProvider
httpx = pytest.importorskip("httpx", reason="legacy httpx not installed")
@ -26,7 +15,6 @@ SPEC = {
"/items": {
"get": {
"operationId": "list_items",
"summary": "List items",
"responses": {
"200": {
"description": "Items",
@ -46,121 +34,77 @@ SPEC = {
}
},
}
},
}
},
}
def _legacy_client(handler) -> "httpx.AsyncClient":
transport = httpx.MockTransport(handler)
return httpx.AsyncClient(transport=transport, base_url="https://api.example.com")
def _server(client) -> FastMCP:
mcp = FastMCP("Legacy Client Server")
mcp.add_provider(OpenAPIProvider(openapi_spec=SPEC, client=client))
return mcp
async def test_tool_call_with_legacy_client_succeeds():
"""A legacy httpx.AsyncClient drives an OpenAPI tool end-to-end."""
async def test_legacy_client_warns_and_remains_usable() -> None:
def handler(request: "httpx.Request") -> "httpx.Response":
assert isinstance(request, httpx.Request)
return httpx.Response(200, json={"items": ["a", "b"]})
async with _legacy_client(handler) as client:
async with Client(_server(client)) as mcp_client:
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(
transport=transport,
base_url="https://api.example.com",
) as client:
with pytest.warns(
FastMCPDeprecationWarning,
match="httpx.AsyncClient.*deprecated",
):
server = FastMCP.from_openapi(SPEC, client=client)
async with Client(server) as mcp_client:
result = await mcp_client.call_tool("list_items", {})
assert result.structured_content == {"items": ["a", "b"]}
assert result.structured_content == {"items": ["a", "b"]}
async def test_tool_http_error_keeps_openapi_formatting_with_legacy_client():
"""A legacy client's HTTP error still gets the integration's message format.
The handler raises legacy ``httpx.HTTPStatusError``; the catch tuples must
recognize it so the error carries the formatted status + body rather than a
generic failure.
"""
async def test_legacy_client_preserves_http_error_details() -> None:
def handler(request: "httpx.Request") -> "httpx.Response":
return httpx.Response(500, json={"detail": "boom"})
return httpx.Response(404, json={"detail": "items not found"})
async with _legacy_client(handler) as client:
async with Client(_server(client)) as mcp_client:
with pytest.raises(ToolError, match="HTTP error 500") as excinfo:
await mcp_client.call_tool("list_items", {})
assert "boom" in str(excinfo.value)
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 def test_tool_request_error_keeps_openapi_formatting_with_legacy_client():
"""A legacy client's transport error maps to the formatted request error."""
def handler(request: "httpx.Request") -> "httpx.Response":
raise httpx.ConnectError("connection refused")
async with _legacy_client(handler) as client:
async with Client(_server(client)) as mcp_client:
with pytest.raises(ToolError, match="Request error"):
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)
async def test_multipart_tool_call_with_legacy_client():
"""Multipart bodies must materialize and send through a legacy client too."""
spec = {
"openapi": "3.0.0",
"info": {"title": "Upload API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com"}],
"paths": {
"/upload": {
"post": {
"operationId": "upload_file",
"summary": "Upload a file",
"requestBody": {
"required": True,
"content": {
"multipart/form-data": {
"schema": {
"type": "object",
"properties": {"file": {"type": "string"}},
}
}
},
},
"responses": {
"200": {
"description": "Uploaded",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {"ok": {"type": "boolean"}},
}
}
},
}
},
}
}
},
}
received: dict[str, object] = {}
@pytest.mark.parametrize(
("error_kind", "message"),
[
("timeout", "HTTP request timed out (ReadTimeout)"),
("connect", "Request error (ConnectError)"),
],
)
async def test_legacy_client_preserves_transport_error_details(
error_kind: str,
message: str,
) -> None:
def handler(request: "httpx.Request") -> "httpx.Response":
received["content_type"] = request.headers.get("content-type", "")
received["body"] = request.read()
return httpx.Response(200, json={"ok": True})
if error_kind == "timeout":
raise httpx.ReadTimeout("transport failed", request=request)
raise httpx.ConnectError("transport failed", request=request)
async with _legacy_client(handler) as client:
mcp = FastMCP("Legacy Multipart Server")
mcp.add_provider(OpenAPIProvider(openapi_spec=spec, client=client))
async with Client(mcp) as mcp_client:
result = await mcp_client.call_tool("upload_file", {"file": "data"})
assert result.structured_content == {"ok": True}
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)
content_type = received["content_type"]
assert isinstance(content_type, str)
assert "multipart/form-data" in content_type
body = received["body"]
assert isinstance(body, bytes)
assert b"data" in body
async with Client(server) as mcp_client:
with pytest.raises(ToolError) as exc_info:
await mcp_client.call_tool("list_items", {})
assert message in str(exc_info.value)

33
tests/server/test_legacy_httpx_errors.py Normal file
View file

@ -0,0 +1,33 @@
"""Compatibility tests for legacy-httpx exceptions raised by user code."""
import pytest
from fastmcp import FastMCP
from fastmcp.exceptions import ResourceError, ToolError
httpx = pytest.importorskip("httpx", reason="legacy httpx not installed")
async def test_legacy_httpx_rate_limit_remains_actionable() -> None:
server = FastMCP("Legacy httpx errors", mask_error_details=True)
@server.tool
def rate_limited() -> None:
request = httpx.Request("GET", "https://example.com")
response = httpx.Response(429, request=request)
raise httpx.HTTPStatusError("rate limited", request=request, response=response)
with pytest.raises(ToolError, match="Rate limited by upstream API"):
await server.call_tool("rate_limited", {})
async def test_legacy_httpx_resource_timeout_remains_actionable() -> None:
server = FastMCP("Legacy httpx errors", mask_error_details=True)
@server.resource("resource://timed-out")
def timed_out() -> str:
request = httpx.Request("GET", "https://example.com")
raise httpx.ReadTimeout("timed out", request=request)
with pytest.raises(ResourceError, match="Upstream request timed out"):
await server.read_resource("resource://timed-out")

41
tests/test_no_legacy_httpx.py
View file

@ -6,11 +6,9 @@ masks clean-install regressions: an accidental ``import httpx`` (directly or
via a third-party integration such as authlib's httpx client) passes CI but
breaks any install without those extras.
This test simulates the clean install by running a subprocess that blocks
legacy httpx imports at the meta-path level, then imports the modules that
have historically regressed. The defensive user-compat shim in
``fastmcp.server.server`` catches ImportError by design and must keep working
when httpx is absent.
These tests simulate a clean install by blocking legacy httpx imports at the
meta-path level and verify that ordinary server startup leaves both legacy
packages unloaded.
"""
import subprocess
@ -45,6 +43,28 @@ _BLOCKER_SCRIPT = textwrap.dedent(
"""
)
_STARTUP_SCRIPT = textwrap.dedent(
"""
import sys
from fastmcp import FastMCP
server = FastMCP("Legacy httpx import guard")
app = server.http_app(transport="http", stateless_http=True)
assert app is not None
loaded = [
name
for name in sys.modules
if name == "httpx"
or name.startswith("httpx.")
or name == "httpcore"
or name.startswith("httpcore.")
]
assert not loaded, loaded
"""
)
@pytest.mark.subprocess_heavy
def test_fastmcp_imports_without_legacy_httpx():
@ -58,3 +78,14 @@ def test_fastmcp_imports_without_legacy_httpx():
f"Import failed with legacy httpx blocked:\n{result.stderr}"
)
assert "OK" in result.stdout
@pytest.mark.subprocess_heavy
def test_default_http_app_does_not_load_legacy_httpx():
result = subprocess.run(
[sys.executable, "-c", _STARTUP_SCRIPT],
capture_output=True,
text=True,
timeout=30,
)
assert result.returncode == 0, result.stderr