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

fix: OpenAPI request director content-type dispatch and cookie params

Browse source 
🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
William Easton 2026-04-14 09:52:28 -05:00
commit 748a374f42
No known key found for this signature in database
2 changed files with 240 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

53
src/fastmcp/utilities/openapi/director.py
View file

@ -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]] = {

200
tests/utilities/openapi/test_director.py
View file

@ -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", "")