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

[codex] Fix OpenAPI resource template requests (#4407)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-07-05 16:11:03 -07:00 • committed by GitHub
commit 691766b5d0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 393 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

66
fastmcp_slim/fastmcp/server/providers/openapi/components.py
View file

@ -269,6 +269,7 @@ class OpenAPIResource(Resource):
description: str,
mime_type: str = "application/json",
tags: set[str] | None = None,
arguments: dict[str, Any] | None = None,
):
super().__init__(
uri=AnyUrl(uri),
@ -280,6 +281,7 @@ class OpenAPIResource(Resource):
self._client = client
self._route = route
self._director = director
self._arguments = dict(arguments or {})
def __repr__(self) -> str:
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
@ -287,40 +289,23 @@ class OpenAPIResource(Resource):
async def read(self) -> ResourceResult:
"""Fetch the resource data by making an HTTP request."""
try:
path = self._route.path
resource_uri = str(self.uri)
# If this is a templated resource, extract path parameters from the URI
if "{" in path and "}" in path:
parts = resource_uri.split("/")
if len(parts) > 1:
path_params = {}
param_matches = re.findall(r"\{([^}]+)\}", path)
if param_matches:
param_matches.sort(reverse=True)
expected_param_count = len(parts) - 1
for i, param_name in enumerate(param_matches):
if i < expected_param_count:
param_value = parts[-1 - i]
path_params[param_name] = param_value
for param_name, param_value in path_params.items():
path = path.replace(f"{{{param_name}}}", str(param_value))
# Build headers with correct precedence
headers: dict[str, str] = {}
if self._client.headers:
headers.update(self._client.headers)
base_url = str(self._client.base_url) or "http://localhost"
directed_request = self._director.build(
self._route, self._arguments, base_url
)
request = self._client.build_request(
method=directed_request.method,
url=directed_request.url.copy_with(query=None),
params=directed_request.url.params,
headers=directed_request.headers,
content=directed_request.content,
extensions=directed_request.extensions,
)
mcp_headers = get_http_headers()
if mcp_headers:
headers.update(mcp_headers)
request.headers.update(mcp_headers)
response = await self._client.request(
method=self._route.method,
url=path,
headers=headers,
)
response = await self._client.send(request)
response.raise_for_status()
content_type = response.headers.get("content-type", "").lower()
@ -368,6 +353,13 @@ class OpenAPIResource(Resource):
raise ValueError(f"Request error ({type(e).__name__}): {e!s}") from e
def _path_argument_name(route: HTTPRoute, parameter_name: str) -> str:
for argument_name, mapping in route.parameter_map.items():
if mapping["location"] == "path" and mapping["openapi_name"] == parameter_name:
return argument_name
return parameter_name
class OpenAPIResourceTemplate(ResourceTemplate):
"""Resource template implementation for OpenAPI endpoints."""
@ -408,6 +400,17 @@ class OpenAPIResourceTemplate(ResourceTemplate):
) -> Resource:
"""Create a resource with the given parameters."""
uri_parts = [f"{key}={value}" for key, value in params.items()]
arguments = {}
for parameter in self._route.parameters:
if parameter.location != "path":
continue
argument_name = _path_argument_name(self._route, parameter.name)
if parameter.name in params:
arguments[argument_name] = params[parameter.name]
continue
normalized_name = parameter.name.replace("-", "_")
if normalized_name in params:
arguments[argument_name] = params[normalized_name]
return OpenAPIResource(
client=self._client,
@ -418,4 +421,5 @@ class OpenAPIResourceTemplate(ResourceTemplate):
description=self.description or f"Resource for {self._route.path}",
mime_type=self.mime_type,
tags=set(self._route.tags or []),
arguments=arguments,
)

9
fastmcp_slim/fastmcp/utilities/openapi/schemas.py
View file

@ -1,5 +1,6 @@
"""Schema manipulation utilities for OpenAPI operations."""
from collections import Counter
from typing import Any
from fastmcp.utilities.logging import get_logger
@ -294,13 +295,17 @@ def _combine_schemas_and_map_params(
body_props = body_schema.get("properties", {})
# Detect collisions: parameters that exist in both body and path/query/header
# Detect collisions: parameters that exist in multiple non-body locations
# or between body and path/query/header/cookie.
all_non_body_params = set()
for location_params in param_names_by_location.values():
all_non_body_params.update(location_params)
body_param_names = set(body_props.keys())
colliding_params = all_non_body_params & body_param_names
non_body_param_counts = Counter(param.name for param in route.parameters)
colliding_params = (all_non_body_params & body_param_names) | {
name for name, count in non_body_param_counts.items() if count > 1
}
# Add parameters with suffixes for collisions
for param in route.parameters:

310
tests/server/providers/openapi/test_openapi_features.py
View file

@ -1,5 +1,6 @@
"""Tests for OpenAPI feature support in OpenAPIProvider."""
from typing import Any
from unittest.mock import AsyncMock, Mock
import httpx
@ -705,6 +706,315 @@ class TestResourceTemplateMimeType:
assert templates[0].mimeType == "application/json"
class TestResourceTemplateRequestBuilding:
@pytest.fixture
def path_param_spec(self) -> dict[str, Any]:
return {
"openapi": "3.0.0",
"info": {"title": "User API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/api/v1"}],
"paths": {
"/users/{id}": {
"get": {
"operationId": "get_user",
"summary": "Get user",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {
"200": {
"description": "User data",
"content": {
"application/json": {"schema": {"type": "object"}}
},
}
},
}
}
},
}
async def test_resource_template_encodes_matched_path_params(
self, path_param_spec: dict[str, Any]
):
seen_urls: list[httpx.URL] = []
async def handler(request: httpx.Request) -> httpx.Response:
seen_urls.append(request.url)
return httpx.Response(200, json={"ok": True})
route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)]
async with httpx.AsyncClient(
base_url="https://api.example.com/api/v1",
transport=httpx.MockTransport(handler),
) as client:
provider = OpenAPIProvider(
openapi_spec=path_param_spec,
client=client,
route_maps=route_maps,
)
mcp = FastMCP("Test")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
await mcp_client.read_resource(
"resource://get_user/..%2F..%2Fadmin%2Fsecret"
)
assert seen_urls == [
httpx.URL(
"https://api.example.com/api/v1/users/%2E%2E%2F%2E%2E%2Fadmin%2Fsecret"
)
]
async def test_resource_template_ignores_unmatched_query_string(
self, path_param_spec: dict[str, Any]
):
seen_urls: list[httpx.URL] = []
async def handler(request: httpx.Request) -> httpx.Response:
seen_urls.append(request.url)
return httpx.Response(200, json={"ok": True})
route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)]
async with httpx.AsyncClient(
base_url="https://api.example.com/api/v1",
transport=httpx.MockTransport(handler),
) as client:
provider = OpenAPIProvider(
openapi_spec=path_param_spec,
client=client,
route_maps=route_maps,
)
mcp = FastMCP("Test")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
await mcp_client.read_resource("resource://get_user/alice?admin=true")
assert seen_urls == [httpx.URL("https://api.example.com/api/v1/users/alice")]
async def test_resource_template_preserves_hyphenated_path_params(self):
seen_urls: list[httpx.URL] = []
async def handler(request: httpx.Request) -> httpx.Response:
seen_urls.append(request.url)
return httpx.Response(200, json={"ok": True})
spec = {
"openapi": "3.0.0",
"info": {"title": "User API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/api/v1"}],
"paths": {
"/users/{user-id}": {
"get": {
"operationId": "get_user",
"parameters": [
{
"name": "user-id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {
"200": {
"description": "User data",
"content": {
"application/json": {"schema": {"type": "object"}}
},
}
},
}
}
},
}
route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)]
async with httpx.AsyncClient(
base_url="https://api.example.com/api/v1",
transport=httpx.MockTransport(handler),
) as client:
provider = OpenAPIProvider(
openapi_spec=spec,
client=client,
route_maps=route_maps,
)
mcp = FastMCP("Test")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
await mcp_client.read_resource("resource://get_user/abc")
assert seen_urls == [httpx.URL("https://api.example.com/api/v1/users/abc")]
async def test_resource_template_preserves_client_defaults(
self, path_param_spec: dict[str, Any]
):
seen_requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
seen_requests.append(request)
return httpx.Response(200, json={"ok": True})
route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)]
async with httpx.AsyncClient(
base_url="https://api.example.com/api/v1",
params={"api-version": "2026-06-29"},
cookies={"session": "abc123"},
transport=httpx.MockTransport(handler),
) as client:
provider = OpenAPIProvider(
openapi_spec=path_param_spec,
client=client,
route_maps=route_maps,
)
mcp = FastMCP("Test")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
await mcp_client.read_resource("resource://get_user/alice")
assert seen_requests[0].url == httpx.URL(
"https://api.example.com/api/v1/users/alice?api-version=2026-06-29"
)
assert seen_requests[0].headers["cookie"] == "session=abc123"
async def test_resource_template_uses_mapped_path_argument_names(self):
seen_urls: list[httpx.URL] = []
async def handler(request: httpx.Request) -> httpx.Response:
seen_urls.append(request.url)
return httpx.Response(200, json={"ok": True})
spec = {
"openapi": "3.0.0",
"info": {"title": "User API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/api/v1"}],
"paths": {
"/users/{id}": {
"get": {
"operationId": "get_user",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"id": {"type": "string"},
"name": {"type": "string"},
},
}
}
}
},
"responses": {
"200": {
"description": "User data",
"content": {
"application/json": {"schema": {"type": "object"}}
},
}
},
}
}
},
}
route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)]
async with httpx.AsyncClient(
base_url="https://api.example.com/api/v1",
transport=httpx.MockTransport(handler),
) as client:
provider = OpenAPIProvider(
openapi_spec=spec,
client=client,
route_maps=route_maps,
)
mcp = FastMCP("Test")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
await mcp_client.read_resource("resource://get_user/abc")
assert seen_urls == [httpx.URL("https://api.example.com/api/v1/users/abc")]
async def test_resource_template_uses_path_arg_when_query_param_has_same_name(
self,
):
seen_urls: list[httpx.URL] = []
async def handler(request: httpx.Request) -> httpx.Response:
seen_urls.append(request.url)
return httpx.Response(200, json={"ok": True})
spec = {
"openapi": "3.0.0",
"info": {"title": "User API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/api/v1"}],
"paths": {
"/users/{id}": {
"get": {
"operationId": "get_user",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
},
{
"name": "id",
"in": "query",
"required": False,
"schema": {"type": "string"},
},
],
"responses": {
"200": {
"description": "User data",
"content": {
"application/json": {"schema": {"type": "object"}}
},
}
},
}
}
},
}
route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)]
async with httpx.AsyncClient(
base_url="https://api.example.com/api/v1",
transport=httpx.MockTransport(handler),
) as client:
provider = OpenAPIProvider(
openapi_spec=spec,
client=client,
route_maps=route_maps,
)
mcp = FastMCP("Test")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
await mcp_client.read_resource("resource://get_user/abc")
assert seen_urls == [httpx.URL("https://api.example.com/api/v1/users/abc")]
class TestResourceMimeType:
"""Test that OpenAPIResource uses inferred MIME types."""

41
tests/utilities/openapi/test_legacy_compatibility.py
View file

@ -102,6 +102,47 @@ class TestSchemaGeneration:
assert param_map["id"]["location"] == "body"
assert param_map["name"]["location"] == "body"
def test_non_body_parameter_collision_handling(self):
"""Test that same-named parameters in different locations use suffixes."""
route = HTTPRoute(
method="GET",
path="/users/{id}",
operation_id="get_user",
parameters=[
ParameterInfo(
name="id",
location="path",
required=True,
schema={"type": "string"},
),
ParameterInfo(
name="id",
location="query",
required=False,
schema={"type": "string"},
),
ParameterInfo(
name="id",
location="header",
required=False,
schema={"type": "string"},
),
],
)
schema, param_map = _combine_schemas_and_map_params(route)
properties = schema["properties"]
assert "id" not in properties
assert "id__path" in properties
assert "id__query" in properties
assert "id__header" in properties
assert schema["required"] == ["id__path"]
assert param_map["id__path"] == {"location": "path", "openapi_name": "id"}
assert param_map["id__query"] == {"location": "query", "openapi_name": "id"}
assert param_map["id__header"] == {"location": "header", "openapi_name": "id"}
@pytest.mark.parametrize(
"param_type",
[