diff --git a/src/fastmcp/experimental/utilities/openapi/schemas.py b/src/fastmcp/experimental/utilities/openapi/schemas.py index 54430b2ed..906d06a53 100644 --- a/src/fastmcp/experimental/utilities/openapi/schemas.py +++ b/src/fastmcp/experimental/utilities/openapi/schemas.py @@ -350,7 +350,7 @@ def _combine_schemas_and_map_params( if route.request_body.required: required.append("body") parameter_map["body"] = {"location": "body", "openapi_name": "body"} - else: + elif body_props: # Normal case: body has properties for prop_name, prop_schema in body_props.items(): properties[prop_name] = prop_schema @@ -363,6 +363,22 @@ def _combine_schemas_and_map_params( if route.request_body.required: required.extend(body_schema.get("required", [])) + else: + # Handle direct array/primitive schemas (like list[str] parameters from FastAPI) + # Use the schema title as parameter name, fall back to generic name + param_name = body_schema.get("title", "body").lower() + + # Clean the parameter name to be valid + import re + + param_name = re.sub(r"[^a-zA-Z0-9_]", "_", param_name) + if not param_name or param_name[0].isdigit(): + param_name = "body_data" + + properties[param_name] = body_schema + if route.request_body.required: + required.append(param_name) + parameter_map[param_name] = {"location": "body", "openapi_name": param_name} result = { "type": "object", diff --git a/tests/experimental/openapi_parser/utilities/openapi/test_direct_array_schemas.py b/tests/experimental/openapi_parser/utilities/openapi/test_direct_array_schemas.py new file mode 100644 index 000000000..02d641e08 --- /dev/null +++ b/tests/experimental/openapi_parser/utilities/openapi/test_direct_array_schemas.py @@ -0,0 +1,323 @@ +"""Test handling of direct array schemas in request bodies (FastAPI list parameters).""" + +from fastmcp.experimental.utilities.openapi.models import ( + HTTPRoute, + ParameterInfo, + RequestBodyInfo, +) +from fastmcp.experimental.utilities.openapi.schemas import ( + _combine_schemas_and_map_params, +) + + +class TestDirectArraySchemas: + """Test handling of direct array schemas like those generated by FastAPI for list[str] parameters.""" + + def test_simple_direct_array_schema(self): + """Test route with direct array request body schema.""" + route = HTTPRoute( + path="/simple_list", + method="POST", + operation_id="simple_list_tool", + summary="Simple List Tool", + description="A simple tool that takes a list of strings.", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "items": {"type": "string"}, + "type": "array", + "title": "Values", + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should create a single parameter from the title + assert combined_schema["type"] == "object" + assert "values" in combined_schema["properties"] + assert "values" in combined_schema["required"] + + # Check the array schema is preserved + values_schema = combined_schema["properties"]["values"] + assert values_schema["type"] == "array" + assert values_schema["items"]["type"] == "string" + assert values_schema["title"] == "Values" + + # Check parameter mapping + assert "values" in param_map + assert param_map["values"]["location"] == "body" + assert param_map["values"]["openapi_name"] == "values" + + def test_int_array_schema(self): + """Test route with integer array request body schema.""" + route = HTTPRoute( + path="/int_list", + method="POST", + operation_id="int_list_tool", + summary="Integer List Tool", + description="A simple tool that takes a list of integers.", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "items": {"type": "integer"}, + "type": "array", + "title": "Numbers", + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should create a single parameter from the title + assert combined_schema["type"] == "object" + assert "numbers" in combined_schema["properties"] + assert "numbers" in combined_schema["required"] + + # Check the array schema is preserved + numbers_schema = combined_schema["properties"]["numbers"] + assert numbers_schema["type"] == "array" + assert numbers_schema["items"]["type"] == "integer" + assert numbers_schema["title"] == "Numbers" + + def test_mixed_params_with_direct_array(self): + """Test route with both URL parameters and direct array request body.""" + route = HTTPRoute( + path="/mixed_params", + method="POST", + operation_id="mixed_params_tool", + summary="Mixed Params Tool", + description="A tool with both list and simple parameters.", + parameters=[ + ParameterInfo( + name="prefix", + location="query", + required=True, + schema={"type": "string", "title": "Prefix"}, + ) + ], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "type": "array", + "items": {"type": "string"}, + "title": "Values", + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should have both parameters + assert combined_schema["type"] == "object" + assert "prefix" in combined_schema["properties"] # From query param + assert "values" in combined_schema["properties"] # From body array + assert "prefix" in combined_schema["required"] + assert "values" in combined_schema["required"] + + # Check query parameter + prefix_schema = combined_schema["properties"]["prefix"] + assert prefix_schema["type"] == "string" + + # Check body array parameter + values_schema = combined_schema["properties"]["values"] + assert values_schema["type"] == "array" + assert values_schema["items"]["type"] == "string" + + # Check parameter mappings + assert param_map["prefix"]["location"] == "query" + assert param_map["values"]["location"] == "body" + + def test_direct_array_no_title(self): + """Test direct array schema without title.""" + route = HTTPRoute( + path="/no_title", + method="POST", + operation_id="no_title_tool", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "type": "array", + "items": {"type": "string"}, + # No title + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should fall back to "body" (default title falls back) + assert combined_schema["type"] == "object" + assert "body" in combined_schema["properties"] + assert "body" in combined_schema["required"] + + # Check parameter mapping + assert param_map["body"]["location"] == "body" + assert param_map["body"]["openapi_name"] == "body" + + def test_direct_primitive_schema(self): + """Test direct primitive (non-array) request body schema.""" + route = HTTPRoute( + path="/primitive", + method="POST", + operation_id="primitive_tool", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": {"type": "string", "title": "Message"} + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should create a single parameter from the title + assert combined_schema["type"] == "object" + assert "message" in combined_schema["properties"] + assert "message" in combined_schema["required"] + + # Check the primitive schema is preserved + message_schema = combined_schema["properties"]["message"] + assert message_schema["type"] == "string" + assert message_schema["title"] == "Message" + + def test_direct_array_optional(self): + """Test direct array schema that's not required.""" + route = HTTPRoute( + path="/optional_list", + method="POST", + operation_id="optional_list_tool", + parameters=[], + request_body=RequestBodyInfo( + required=False, # Not required + content_schema={ + "application/json": { + "type": "array", + "items": {"type": "string"}, + "title": "OptionalValues", + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should create parameter but not mark as required + assert combined_schema["type"] == "object" + assert "optionalvalues" in combined_schema["properties"] + assert "optionalvalues" not in combined_schema["required"] + + def test_direct_array_title_sanitization(self): + """Test that titles with special characters are sanitized.""" + route = HTTPRoute( + path="/special_chars", + method="POST", + operation_id="special_chars_tool", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "type": "array", + "items": {"type": "string"}, + "title": "Special-Chars & Spaces!", + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should sanitize the title to a valid parameter name (& becomes multiple underscores) + assert combined_schema["type"] == "object" + assert "special_chars___spaces_" in combined_schema["properties"] + + def test_direct_array_title_starting_with_number(self): + """Test that titles starting with numbers are handled.""" + route = HTTPRoute( + path="/number_start", + method="POST", + operation_id="number_start_tool", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "type": "array", + "items": {"type": "string"}, + "title": "123Values", + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should fall back to "body_data" since "123values" starts with a number + assert combined_schema["type"] == "object" + assert "body_data" in combined_schema["properties"] + + def test_preserve_existing_behavior_object_body(self): + """Test that existing object-based request bodies still work.""" + route = HTTPRoute( + path="/object_body", + method="POST", + operation_id="object_body_tool", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "age": {"type": "integer"}, + }, + "required": ["name"], + } + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should handle object bodies as before + assert combined_schema["type"] == "object" + assert "name" in combined_schema["properties"] + assert "age" in combined_schema["properties"] + assert "name" in combined_schema["required"] + assert "age" not in combined_schema["required"] + + def test_preserve_existing_behavior_ref_body(self): + """Test that $ref-based request bodies still work.""" + route = HTTPRoute( + path="/ref_body", + method="POST", + operation_id="ref_body_tool", + parameters=[], + request_body=RequestBodyInfo( + required=True, + content_schema={ + "application/json": {"$ref": "#/components/schemas/UserCreate"} + }, + ), + ) + + combined_schema, param_map = _combine_schemas_and_map_params(route) + + # Should handle $ref bodies as before (refs get converted to $defs) + assert combined_schema["type"] == "object" + assert "body" in combined_schema["properties"] + assert combined_schema["properties"]["body"]["$ref"] == "#/$defs/UserCreate"