From 748a374f428599c671105902ddf6d7e4023da238 Mon Sep 17 00:00:00 2001 From: William Easton Date: Tue, 14 Apr 2026 09:52:28 -0500 Subject: [PATCH] fix: OpenAPI request director content-type dispatch and cookie params MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🤖 Generated with Claude Code Co-Authored-By: Claude Opus 4.6 (1M context) --- src/fastmcp/utilities/openapi/director.py | 53 +++++- tests/utilities/openapi/test_director.py | 200 +++++++++++++++++++++- 2 files changed, 240 insertions(+), 13 deletions(-) diff --git a/src/fastmcp/utilities/openapi/director.py b/src/fastmcp/utilities/openapi/director.py index 8980e3d6a..fa03aa992 100644 --- a/src/fastmcp/utilities/openapi/director.py +++ b/src/fastmcp/utilities/openapi/director.py @@ -54,8 +54,8 @@ class RequestDirector: ) # Step 1: Un-flatten arguments into path, query, body, etc. using parameter map - path_params, query_params, header_params, body = self._unflatten_arguments( - route, flat_args + path_params, query_params, header_params, cookie_params, body = ( + self._unflatten_arguments(route, flat_args) ) logger.debug( @@ -80,9 +80,40 @@ class RequestDirector: if route.request_body and route.request_body.content_schema: declared_content_type = next(iter(route.request_body.content_schema)) + cookies = cookie_params if cookie_params else None + # Step 6: Handle request body if body is not None: - if isinstance(body, dict | list): + if declared_content_type == "multipart/form-data" and isinstance( + body, dict + ): + # httpx requires files= for multipart/form-data encoding. + # Wrap plain values as (None, value) tuples so httpx treats + # them as form fields rather than file uploads. + files = { + k: v if isinstance(v, tuple) else (None, v) for k, v in body.items() + } + return httpx.Request( + method=method, + url=url, + params=params, + headers=headers, + files=files, + cookies=cookies, + ) + elif ( + declared_content_type == "application/x-www-form-urlencoded" + and isinstance(body, dict) + ): + return httpx.Request( + method=method, + url=url, + params=params, + headers=headers, + data=body, + cookies=cookies, + ) + elif isinstance(body, dict | list): if ( declared_content_type is not None and declared_content_type != "application/json" @@ -108,11 +139,12 @@ class RequestDirector: headers=headers, json=json_body, content=content, + cookies=cookies, ) def _unflatten_arguments( self, route: HTTPRoute, flat_args: dict[str, Any] - ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], Any]: + ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], Any]: """ Maps flat arguments back to their OpenAPI locations using the parameter map. @@ -121,11 +153,12 @@ class RequestDirector: flat_args: Flat arguments from LLM call Returns: - Tuple of (path_params, query_params, header_params, body) + Tuple of (path_params, query_params, header_params, cookie_params, body) """ path_params = {} query_params = {} header_params = {} + cookie_params = {} body_props = {} # Use parameter map to route arguments to correct locations @@ -150,6 +183,8 @@ class RequestDirector: query_params[openapi_name] = value elif location == "header": header_params[openapi_name] = value + elif location == "cookie": + cookie_params[openapi_name] = value elif location == "body": body_props[openapi_name] = value else: @@ -173,13 +208,15 @@ class RequestDirector: # Check if it's a suffixed parameter (e.g., id__path) if "__" in arg_name: base_name, location = arg_name.rsplit("__", 1) - if location in ["path", "query", "header"]: + if location in ["path", "query", "header", "cookie"]: if location == "path": path_params[base_name] = value elif location == "query": query_params[base_name] = value elif location == "header": header_params[base_name] = value + elif location == "cookie": + cookie_params[base_name] = value continue # Check if it's a known parameter @@ -191,6 +228,8 @@ class RequestDirector: query_params[arg_name] = value elif location == "header": header_params[arg_name] = value + elif location == "cookie": + cookie_params[arg_name] = value else: # Assume it's a body property body_props[arg_name] = value @@ -222,7 +261,7 @@ class RequestDirector: else: body = body_props - return path_params, query_params, header_params, body + return path_params, query_params, header_params, cookie_params, body # Delimiter per OpenAPI style when explode=false _STYLE_DELIMITERS: ClassVar[dict[str, str]] = { diff --git a/tests/utilities/openapi/test_director.py b/tests/utilities/openapi/test_director.py index 56008b241..c3e1a4f3b 100644 --- a/tests/utilities/openapi/test_director.py +++ b/tests/utilities/openapi/test_director.py @@ -530,8 +530,8 @@ class TestContentTypeHandling: request = director.build(route, {"name": "test"}, "https://example.com") assert request.headers["content-type"] == "application/json" - def test_non_json_content_type_falls_through(self, director): - """Non-JSON types like multipart/form-data don't get JSON-serialized.""" + def test_non_json_content_type_uses_native_encoding(self, director): + """Non-JSON types like multipart/form-data use httpx's native encoding.""" route = HTTPRoute( path="/upload", method="POST", @@ -551,10 +551,8 @@ class TestContentTypeHandling: ) request = director.build(route, {"file": "data"}, "https://example.com") - # Should fall through to httpx's json= path (not manually serialized - # with a multipart/form-data header), since the content type isn't - # JSON-compatible. - assert request.headers["content-type"] == "application/json" + # multipart/form-data should be sent as multipart, not JSON + assert "multipart/form-data" in request.headers["content-type"] class TestQueryParameterSerialization: @@ -1152,3 +1150,193 @@ class TestPathTraversalPrevention: # Verify traversal didn't escape the users/ prefix assert decoded.startswith("https://api.example.com/api/v1/users/") assert url.startswith("https://api.example.com/api/v1/users/") + + +class TestContentTypeDispatch: + """Test that non-JSON content types are dispatched correctly.""" + + @pytest.fixture + def director(self): + spec = SchemaPath.from_dict( + { + "openapi": "3.0.0", + "info": {"title": "t", "version": "1"}, + "paths": {}, + } + ) + return RequestDirector(spec) + + @pytest.fixture + def multipart_route(self): + return HTTPRoute( + path="/upload", + method="POST", + operation_id="upload_file", + request_body=RequestBodyInfo( + required=True, + content_schema={ + "multipart/form-data": { + "type": "object", + "properties": { + "filename": {"type": "string"}, + "data": {"type": "string"}, + }, + } + }, + ), + parameter_map={ + "filename": {"location": "body", "openapi_name": "filename"}, + "data": {"location": "body", "openapi_name": "data"}, + }, + ) + + @pytest.fixture + def form_urlencoded_route(self): + return HTTPRoute( + path="/login", + method="POST", + operation_id="login", + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/x-www-form-urlencoded": { + "type": "object", + "properties": { + "username": {"type": "string"}, + "password": {"type": "string"}, + }, + } + }, + ), + parameter_map={ + "username": {"location": "body", "openapi_name": "username"}, + "password": {"location": "body", "openapi_name": "password"}, + }, + ) + + @pytest.fixture + def json_route(self): + return HTTPRoute( + path="/items", + method="POST", + operation_id="create_item", + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "type": "object", + "properties": { + "name": {"type": "string"}, + }, + } + }, + ), + parameter_map={ + "name": {"location": "body", "openapi_name": "name"}, + }, + ) + + def test_multipart_form_data_not_sent_as_json(self, director, multipart_route): + """multipart/form-data bodies should use httpx data= not json=.""" + request = director.build( + multipart_route, + {"filename": "test.txt", "data": "hello"}, + ) + content_type = request.headers.get("content-type", "") + assert "multipart/form-data" in content_type + # Should NOT be JSON-encoded + assert "application/json" not in content_type + + def test_form_urlencoded_not_sent_as_json(self, director, form_urlencoded_route): + """application/x-www-form-urlencoded bodies should use httpx data=.""" + request = director.build( + form_urlencoded_route, + {"username": "alice", "password": "secret"}, + ) + content_type = request.headers.get("content-type", "") + assert "application/x-www-form-urlencoded" in content_type + assert "application/json" not in content_type + # Verify the body is form-encoded + body_text = request.content.decode("utf-8") + assert "username=alice" in body_text + assert "password=secret" in body_text + + def test_json_body_still_works(self, director, json_route): + """application/json bodies should still be sent as JSON (regression).""" + request = director.build( + json_route, + {"name": "widget"}, + ) + content_type = request.headers.get("content-type", "") + assert "application/json" in content_type + body = json.loads(request.content.decode("utf-8")) + assert body == {"name": "widget"} + + +class TestCookieParameters: + """Test that cookie parameters are properly routed.""" + + @pytest.fixture + def director(self): + spec = SchemaPath.from_dict( + { + "openapi": "3.0.0", + "info": {"title": "t", "version": "1"}, + "paths": {}, + } + ) + return RequestDirector(spec) + + @pytest.fixture + def cookie_route(self): + return HTTPRoute( + path="/dashboard", + method="GET", + operation_id="get_dashboard", + parameters=[ + ParameterInfo( + name="session_id", + location="cookie", + required=True, + schema={"type": "string"}, + ), + ], + parameter_map={ + "session_id": { + "location": "cookie", + "openapi_name": "session_id", + }, + }, + ) + + def test_cookie_parameter_set_on_request(self, director, cookie_route): + """Cookie parameters should appear as cookies on the request.""" + request = director.build( + cookie_route, + {"session_id": "abc123"}, + ) + assert "session_id" in request.headers.get("cookie", "") + assert "abc123" in request.headers.get("cookie", "") + + def test_cookie_parameter_with_fallback_mapping(self, director): + """Cookie parameters should work in fallback (no parameter_map) mode.""" + route = HTTPRoute( + path="/dashboard", + method="GET", + operation_id="get_dashboard", + parameters=[ + ParameterInfo( + name="session_id", + location="cookie", + required=True, + schema={"type": "string"}, + ), + ], + # No parameter_map — triggers fallback path + ) + request = director.build( + route, + {"session_id": "xyz789"}, + ) + assert "session_id" in request.headers.get("cookie", "") + assert "xyz789" in request.headers.get("cookie", "")