diff --git a/src/fastmcp/utilities/openapi/director.py b/src/fastmcp/utilities/openapi/director.py index 537fdffdc..6b2d6eb95 100644 --- a/src/fastmcp/utilities/openapi/director.py +++ b/src/fastmcp/utilities/openapi/director.py @@ -1,6 +1,6 @@ """Request director using openapi-core for stateless HTTP request building.""" -from typing import Any +from typing import Any, ClassVar from urllib.parse import quote, urljoin import httpx @@ -8,11 +8,22 @@ from jsonschema_path import SchemaPath from fastmcp.utilities.logging import get_logger -from .models import HTTPRoute +from .models import HTTPRoute, ParameterInfo logger = get_logger(__name__) +def _query_scalar_to_str(value: Any) -> str: + """Convert a scalar to its query-string representation. + + Booleans are lowercased to match JSON/OpenAPI conventions (true/false) + rather than Python's str(True) → "True". + """ + if isinstance(value, bool): + return "true" if value else "false" + return str(value) + + class RequestDirector: """Builds httpx.Request objects from HTTPRoute and arguments using openapi-core.""" @@ -50,24 +61,27 @@ class RequestDirector: f"Unflattened - path: {path_params}, query: {query_params}, headers: {header_params}, body: {body}" ) - # Step 2: Build base URL with path parameters + # Step 2: Serialize query parameters according to OpenAPI style/explode + query_params = self._serialize_query_params(route, query_params) + + # Step 3: Build base URL with path parameters url = self._build_url(route.path, path_params, base_url) - # Step 3: Prepare request data + # Step 4: Prepare request data method: str = route.method.upper() params = query_params if query_params else None headers = header_params if header_params else None json_body: dict[str, Any] | list[Any] | None = None content: str | bytes | None = None - # Step 4: Handle request body + # Step 5: Handle request body if body is not None: if isinstance(body, dict | list): json_body = body else: content = body - # Step 5: Create httpx.Request + # Step 6: Create httpx.Request return httpx.Request( method=method, url=url, @@ -191,6 +205,65 @@ class RequestDirector: return path_params, query_params, header_params, body + # Delimiter per OpenAPI style when explode=false + _STYLE_DELIMITERS: ClassVar[dict[str, str]] = { + "form": ",", + "spaceDelimited": " ", + "pipeDelimited": "|", + } + + def _serialize_query_params( + self, + route: HTTPRoute, + query_params: dict[str, Any], + ) -> dict[str, Any]: + """ + Serialize query parameter values according to their OpenAPI style/explode settings. + + By default (style=form, explode=true), list values are passed through as-is + so httpx repeats the key (e.g. values=a&values=b). When explode=false, + list values are joined with the style-appropriate delimiter: + - form (default): comma (values=a,b) + - pipeDelimited: pipe (values=a|b) + - spaceDelimited: space (values=a%20b) + """ + if not query_params: + return query_params + + # Build a lookup from openapi_name -> ParameterInfo for query params + param_lookup: dict[str, ParameterInfo] = { + p.name: p for p in route.parameters if p.location == "query" + } + + serialized: dict[str, Any] = {} + for key, value in query_params.items(): + param_info = param_lookup.get(key) + if param_info is not None: + explode = param_info.explode if param_info.explode is not None else True + if not explode: + style = param_info.style or "form" + delimiter = self._STYLE_DELIMITERS.get(style, ",") + if isinstance(value, list): + if not value: + continue + serialized[key] = delimiter.join( + _query_scalar_to_str(v) for v in value + ) + continue + elif isinstance(value, dict): + if not value: + continue + # form,explode=false on objects: key,value pairs + # e.g. {"R": 100, "G": 200} → "R,100,G,200" + parts: list[str] = [] + for k, v in value.items(): + parts.append(_query_scalar_to_str(k)) + parts.append(_query_scalar_to_str(v)) + serialized[key] = delimiter.join(parts) + continue + serialized[key] = value + return serialized + def _build_url( self, path_template: str, path_params: dict[str, Any], base_url: str ) -> str: diff --git a/tests/utilities/openapi/test_director.py b/tests/utilities/openapi/test_director.py index edad3eab2..4f7a8f514 100644 --- a/tests/utilities/openapi/test_director.py +++ b/tests/utilities/openapi/test_director.py @@ -385,6 +385,305 @@ class TestRequestDirector: assert body_data == {"prop1": "value1", "prop2": "value2"} +class TestQueryParameterSerialization: + """Test that query parameters respect OpenAPI explode/style settings.""" + + @pytest.fixture + def director(self, basic_openapi_30_spec): + spec = SchemaPath.from_dict(basic_openapi_30_spec) + return RequestDirector(spec) + + def test_explode_true_repeats_keys(self, director): + """Default behavior: explode=true sends values=a&values=b.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="values", + location="query", + required=True, + schema={"type": "array", "items": {"type": "string"}}, + explode=True, + ) + ], + parameter_map={ + "values": {"location": "query", "openapi_name": "values"}, + }, + ) + + request = director.build( + route, {"values": ["hello", "world"]}, "https://example.com" + ) + url = str(request.url) + assert "values=hello" in url + assert "values=world" in url + + def test_explode_false_comma_joins(self, director): + """explode=false sends values=hello,world.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="values", + location="query", + required=True, + schema={"type": "array", "items": {"type": "string"}}, + explode=False, + ) + ], + parameter_map={ + "values": {"location": "query", "openapi_name": "values"}, + }, + ) + + request = director.build( + route, {"values": ["hello", "world"]}, "https://example.com" + ) + url = str(request.url) + assert "values=hello%2Cworld" in url or "values=hello,world" in url + # Must NOT have repeated keys + assert url.count("values=") == 1 + + def test_explode_none_defaults_to_true(self, director): + """When explode is unset, OpenAPI default for form style is explode=true.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="tags", + location="query", + required=True, + schema={"type": "array", "items": {"type": "string"}}, + explode=None, + ) + ], + parameter_map={ + "tags": {"location": "query", "openapi_name": "tags"}, + }, + ) + + request = director.build(route, {"tags": ["a", "b"]}, "https://example.com") + url = str(request.url) + assert "tags=a" in url + assert "tags=b" in url + + def test_explode_false_with_integers(self, director): + """explode=false works with non-string values.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="ids", + location="query", + required=True, + schema={"type": "array", "items": {"type": "integer"}}, + explode=False, + ) + ], + parameter_map={ + "ids": {"location": "query", "openapi_name": "ids"}, + }, + ) + + request = director.build(route, {"ids": [1, 2, 3]}, "https://example.com") + url = str(request.url) + assert "ids=1%2C2%2C3" in url or "ids=1,2,3" in url + assert url.count("ids=") == 1 + + def test_scalar_query_param_unaffected_by_explode(self, director): + """Non-list values pass through regardless of explode setting.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="get_item", + parameters=[ + ParameterInfo( + name="name", + location="query", + required=True, + schema={"type": "string"}, + explode=False, + ) + ], + parameter_map={ + "name": {"location": "query", "openapi_name": "name"}, + }, + ) + + request = director.build(route, {"name": "foo"}, "https://example.com") + assert "name=foo" in str(request.url) + + def test_pipe_delimited_explode_false(self, director): + """style=pipeDelimited, explode=false sends ids=1|2|3.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="ids", + location="query", + required=True, + schema={"type": "array", "items": {"type": "string"}}, + explode=False, + style="pipeDelimited", + ) + ], + parameter_map={ + "ids": {"location": "query", "openapi_name": "ids"}, + }, + ) + + request = director.build(route, {"ids": ["1", "2", "3"]}, "https://example.com") + url = str(request.url) + assert "ids=1%7C2%7C3" in url or "ids=1|2|3" in url + assert url.count("ids=") == 1 + + def test_space_delimited_explode_false(self, director): + """style=spaceDelimited, explode=false sends ids=1%202%203.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="ids", + location="query", + required=True, + schema={"type": "array", "items": {"type": "string"}}, + explode=False, + style="spaceDelimited", + ) + ], + parameter_map={ + "ids": {"location": "query", "openapi_name": "ids"}, + }, + ) + + request = director.build(route, {"ids": ["1", "2", "3"]}, "https://example.com") + url = str(request.url) + assert "ids=1+2+3" in url or "ids=1%202%203" in url + assert url.count("ids=") == 1 + + def test_explode_false_booleans_lowercased(self, director): + """Booleans serialize as true/false, not True/False.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="flags", + location="query", + required=True, + schema={"type": "array", "items": {"type": "boolean"}}, + explode=False, + ) + ], + parameter_map={ + "flags": {"location": "query", "openapi_name": "flags"}, + }, + ) + + request = director.build(route, {"flags": [True, False]}, "https://example.com") + url = str(request.url) + assert "true" in url and "false" in url + assert "True" not in url and "False" not in url + + def test_explode_false_empty_list_omitted(self, director): + """Empty list with explode=false omits the parameter entirely.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="ids", + location="query", + required=False, + schema={"type": "array", "items": {"type": "string"}}, + explode=False, + ) + ], + parameter_map={ + "ids": {"location": "query", "openapi_name": "ids"}, + }, + ) + + request = director.build(route, {"ids": []}, "https://example.com") + assert "ids" not in str(request.url) + + def test_explode_false_dict_value(self, director): + """style=form, explode=false on objects serializes as key,value pairs.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="color", + location="query", + required=True, + schema={ + "type": "object", + "properties": { + "R": {"type": "integer"}, + "G": {"type": "integer"}, + "B": {"type": "integer"}, + }, + }, + explode=False, + style="form", + ) + ], + parameter_map={ + "color": {"location": "query", "openapi_name": "color"}, + }, + ) + + request = director.build( + route, + {"color": {"R": 100, "G": 200, "B": 150}}, + "https://example.com", + ) + url = str(request.url) + assert "color=R" in url + assert url.count("color=") == 1 + # Should contain alternating key,value pairs + assert "100" in url and "200" in url and "150" in url + + def test_explode_false_empty_dict_omitted(self, director): + """Empty dict with explode=false omits the parameter.""" + route = HTTPRoute( + path="/items", + method="GET", + operation_id="list_items", + parameters=[ + ParameterInfo( + name="filter", + location="query", + required=False, + schema={"type": "object"}, + explode=False, + ) + ], + parameter_map={ + "filter": {"location": "query", "openapi_name": "filter"}, + }, + ) + + request = director.build(route, {"filter": {}}, "https://example.com") + assert "filter" not in str(request.url) + + class TestRequestDirectorIntegration: """Test RequestDirector with real parsed routes."""