bugfix

Author: shira-ayal

fastapi_mcp/http_tools.py

@@ -16,7 +16,8 @@
 from .openapi_utils import (
     clean_schema_for_display,
     generate_example_from_schema,
-    parse_param_schema_for_python_type_and_default,
+    get_python_type_and_default,
+    get_single_param_type_from_schema,
     resolve_schema_references,
     PYTHON_TYPE_IMPORTS,
 )
@@ -101,18 +102,22 @@ def create_mcp_tools_from_openapi(
 
 def _create_http_tool_function(function_template: Callable, properties: Dict[str, Any], additional_variables: Dict[str, Any]) -> Callable:
     # Build parameter string with type hints
-    param_list = []
-    param_list_with_defaults = []
-    for name, schema in properties.items():
-        type_hint, has_default_value = parse_param_schema_for_python_type_and_default(schema)
+    parsed_parameters = {}
+    parsed_parameters_with_defaults = {}
+    for param_name, parsed_param_schema in properties.items():
+        type_hint, has_default_value = get_python_type_and_default(parsed_param_schema)
         if has_default_value:
-            param_list_with_defaults.append(f"{name}: {type_hint}")
+            parsed_parameters_with_defaults[param_name] = f"{param_name}: {type_hint}"
         else:
-            param_list.append(f"{name}: {type_hint}")
-    parameters_str = ", ".join(param_list + param_list_with_defaults)
+            parsed_parameters[param_name] = f"{param_name}: {type_hint}"
+
+    parsed_parameters_keys = list(parsed_parameters.keys()) + list(parsed_parameters_with_defaults.keys())
+    parsed_parameters_values = list(parsed_parameters.values()) + list(parsed_parameters_with_defaults.values())
+    parameters_str = ", ".join(parsed_parameters_values)
+    kwargs_str = ', '.join([f"'{k}': {k}" for k in parsed_parameters_keys])
 
     dynamic_function_body = f"""async def dynamic_http_tool_function({parameters_str}):
-        kwargs = {{{', '.join([f"'{k}': {k}" for k in properties.keys()])}}}
+        kwargs = {{{kwargs_str}}}
         return await http_tool_function_template(**kwargs)
     """
 
@@ -302,7 +307,6 @@ def create_http_tool(
     query_params = []
     header_params = []
     body_params = []
-
     for param in parameters:
         param_name = param.get("name")
         param_in = param.get("in")
@@ -360,10 +364,12 @@ def create_http_tool(
         param_required = param.get("required", False)
 
         properties[param_name] = {
-            "type": param_schema.get("type", "string"),
+            "type": get_single_param_type_from_schema(param_schema),
             "title": param_name,
             "description": param_desc,
         }
+        if "default" in param_schema:
+            properties[param_name]["default"] = param_schema["default"]
 
         if param_required:
             required_props.append(param_name)
@@ -373,8 +379,13 @@ def create_http_tool(
         param_schema = param.get("schema", {})
         param_required = param.get("required", False)
 
-        properties[param_name] = param_schema
-        properties[param_name]["title"] = param_name
+        # properties[param_name] = param_schema
+        properties[param_name] = {
+            "type": get_single_param_type_from_schema(param_schema),
+            "title": param_name,
+        }
+        if "default" in param_schema:
+            properties[param_name]["default"] = param_schema["default"]
 
         if param_required:
             required_props.append(param_name)

fastapi_mcp/openapi_utils.py

@@ -107,36 +107,32 @@
     "unknown": "Any"
 }
 
-
-def parse_param_schema_for_python_type_and_default(param_schema: Dict[str, Any]) -> tuple[str, bool]:
+def get_single_param_type_from_schema(param_schema: Dict[str, Any]) -> str:
     """
-    Parse OpenAPI parameters into a python type and default value string.
-    
-    Args:
-        parameters: List of OpenAPI parameter objects
-        
-    Returns:
-        A tuple containing:
-            - A string representing the Python type annotation (e.g. "str", "Optional[int]", etc.)
-            - A boolean indicating whether a default value is present
+    Get the type of a parameter from the schema. 
+    If the schema is a union type, return the first type.
     """
-    # Handle anyOf case
     if "anyOf" in param_schema:
-        # Get a set of possible types for this parameter
         types = {schema.get("type") for schema in param_schema["anyOf"] if schema.get("type")}
-        # If null is one of the types, make None the default value
         if "null" in types:
             types.remove("null")
-            if types:
-                return f"Optional[{OPENAPI_PYTHON_TYPES_MAP.get(next(iter(types)), 'Any')}] = None", True
-            return "Optional[str] = None", True
-        return f"Union[{', '.join([OPENAPI_PYTHON_TYPES_MAP.get(t, 'Any') for t in types])}]", False
-    
+        if types:
+            return next(iter(types))
+        return "string"
+    return param_schema.get("type", "string")
+
+def get_python_type_and_default(parsed_param_schema: Dict[str, Any]) -> tuple[str, bool]:
+    """
+    Parse parameters into a python type and default value string.
+    Returns:
+        A tuple containing:
+            - A string representing the Python type annotation (e.g. "str", "int = 0", etc.)
+            - A boolean indicating whether a default value is present
+    """
     # Handle direct type specification
-    python_type = OPENAPI_PYTHON_TYPES_MAP.get(param_schema.get("type", ""), 'Any')
-    default_value = param_schema.get("default")
-    if default_value is not None:   
-        return f"{python_type} = {default_value}", True
+    python_type = OPENAPI_PYTHON_TYPES_MAP.get(parsed_param_schema.get("type", ""), "Any")
+    if "default" in parsed_param_schema:   
+        return f"{python_type} = {parsed_param_schema.get("default")}", True
     return python_type, False
 
 def resolve_schema_references(schema_part: Dict[str, Any], reference_schema: Dict[str, Any]) -> Dict[str, Any]: