fix: preserve array items property in union schemas for MCP compatibility

examples/01_basic_usage_example.py

@@ -1,4 +1,4 @@
-from examples.shared.apps.items import app # The FastAPI app
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
@@ -15,4 +15,4 @@
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000)

examples/02_full_schema_description_example.py

@@ -1,8 +1,8 @@
-
 """
 This example shows how to describe the full response schema instead of just a response example.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
@@ -22,5 +22,5 @@
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/03_custom_exposed_endpoints_example.py

@@ -6,7 +6,8 @@
 - You can combine operation filtering with tag filtering (e.g., use `include_operations` with `include_tags`)
 - When combining filters, a greedy approach will be taken. Endpoints matching either criteria will be included
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
@@ -24,7 +25,7 @@
 
 # Filter by excluding specific operation IDs
 exclude_operations_mcp = FastApiMCP(
-    app,    
+    app,
     name="Item API MCP - Excluded Operations",
     exclude_operations=["create_item", "update_item", "delete_item"],
 )

examples/04_separate_server_example.py

@@ -2,6 +2,7 @@
 This example shows how to run the MCP server and the FastAPI app separately.
 You can create an MCP server from one FastAPI app, and mount it to a different app.
 """
+
 from fastapi import FastAPI
 
 from examples.shared.apps.items import app
@@ -30,4 +31,4 @@
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(mcp_app, host="0.0.0.0", port=8000)
+    uvicorn.run(mcp_app, host="0.0.0.0", port=8000)

examples/05_reregister_tools_example.py

@@ -1,15 +1,16 @@
 """
 This example shows how to re-register tools if you add endpoints after the MCP server was created.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
 
 setup_logging()
 
-mcp = FastApiMCP(app) # Add MCP server to the FastAPI app
-mcp.mount() # MCP server
+mcp = FastApiMCP(app)  # Add MCP server to the FastAPI app
+mcp.mount()  # MCP server
 
 
 # This endpoint will not be registered as a tool, since it was added after the MCP instance was created
@@ -24,5 +25,5 @@ async def new_endpoint():
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/06_custom_mcp_router_example.py

@@ -1,15 +1,16 @@
 """
 This example shows how to mount the MCP server to a specific APIRouter, giving a custom mount path.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi import APIRouter
 from fastapi_mcp import FastApiMCP
 
 setup_logging()
 
-other_router = APIRouter(prefix="/other/route")    
+other_router = APIRouter(prefix="/other/route")
 app.include_router(other_router)
 
 mcp = FastApiMCP(app)
@@ -21,5 +22,5 @@
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/07_configure_http_timeout_example.py

@@ -2,7 +2,8 @@
 This example shows how to configure the HTTP client timeout for the MCP server.
 In case you have API endpoints that take longer than 5 seconds to respond, you can increase the timeout.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 import httpx
@@ -12,14 +13,11 @@
 setup_logging()
 
 
-mcp = FastApiMCP(
-    app,
-    http_client=httpx.AsyncClient(timeout=20)
-)
+mcp = FastApiMCP(app, http_client=httpx.AsyncClient(timeout=20))
 mcp.mount()
 
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/08_auth_example_token_passthrough.py

@@ -21,7 +21,8 @@
 }
 ```
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi import Depends
@@ -34,11 +35,13 @@
 # Scheme for the Authorization header
 token_auth_scheme = HTTPBearer()
 
+
 # Create a private endpoint
 @app.get("/private")
-async def private(token = Depends(token_auth_scheme)):
+async def private(token=Depends(token_auth_scheme)):
     return token.credentials
 
+
 # Create the MCP server with the token auth scheme
 mcp = FastApiMCP(
     app,
@@ -54,5 +57,5 @@ async def private(token = Depends(token_auth_scheme)):
 
 if __name__ == "__main__":
     import uvicorn
-    
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)

fastapi_mcp/openapi/convert.py

@@ -9,6 +9,7 @@
     generate_example_from_schema,
     resolve_schema_references,
     get_single_param_type_from_schema,
+    simplify_union_schema,
 )
 
 logger = logging.getLogger(__name__)
@@ -219,15 +220,18 @@ def convert_openapi_to_mcp_tools(
                 param_desc = param.get("description", "")
                 param_required = param.get("required", False)
 
-                properties[param_name] = param_schema.copy()
+                # Simplify union schemas to single type for MCP compatibility
+                simplified_schema = simplify_union_schema(param_schema)
+
+                properties[param_name] = simplified_schema.copy()
                 properties[param_name]["title"] = param_name
                 if param_desc:
                     properties[param_name]["description"] = param_desc
 
                 if "type" not in properties[param_name]:
                     properties[param_name]["type"] = get_single_param_type_from_schema(param_schema)
 
-                if "default" in param_schema:
+                if "default" in param_schema and "default" not in properties[param_name]:
                     properties[param_name]["default"] = param_schema["default"]
 
                 if param_required:
@@ -239,15 +243,18 @@ def convert_openapi_to_mcp_tools(
                 param_desc = param.get("description", "")
                 param_required = param.get("required", False)
 
-                properties[param_name] = param_schema.copy()
+                # Simplify union schemas to single type for MCP compatibility
+                simplified_schema = simplify_union_schema(param_schema)
+
+                properties[param_name] = simplified_schema.copy()
                 properties[param_name]["title"] = param_name
                 if param_desc:
                     properties[param_name]["description"] = param_desc
 
                 if "type" not in properties[param_name]:
                     properties[param_name]["type"] = get_single_param_type_from_schema(param_schema)
 
-                if "default" in param_schema:
+                if "default" in param_schema and "default" not in properties[param_name]:
                     properties[param_name]["default"] = param_schema["default"]
 
                 if param_required:

fastapi_mcp/openapi/utils.py

@@ -16,6 +16,42 @@ def get_single_param_type_from_schema(param_schema: Dict[str, Any]) -> str:
     return param_schema.get("type", "string")
 
 
+def simplify_union_schema(param_schema: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Simplify a union schema to a single schema for MCP compatibility.
+    Preserves important properties like 'items' for arrays.
+
+    Args:
+        param_schema: The parameter schema that may contain anyOf/oneOf
+
+    Returns:
+        A simplified schema with a single type
+    """
+    # Return as-is if not a union
+    if "anyOf" not in param_schema and "oneOf" not in param_schema:
+        return param_schema
+
+    # Get union schemas
+    union_schemas = param_schema.get("anyOf", param_schema.get("oneOf", []))
+
+    # Filter out null types
+    non_null_schemas = [s for s in union_schemas if s.get("type") != "null"]
+
+    if not non_null_schemas:
+        # All schemas were null, return string as default
+        return {"type": "string"}
+
+    # Use the first non-null schema, preserving all its properties
+    result = non_null_schemas[0].copy()
+
+    # Preserve important properties from parent schema
+    for key in ["title", "description", "default"]:
+        if key in param_schema and key not in result:
+            result[key] = param_schema[key]
+
+    return result
+
+
 def resolve_schema_references(schema_part: Dict[str, Any], reference_schema: Dict[str, Any]) -> Dict[str, Any]:
     """
     Resolve schema references in OpenAPI schemas.