From 39dee1bdb32da8c43acdcffb48ec04fa41cd1418 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Fri, 19 Sep 2025 13:35:45 -0400
Subject: [PATCH] feat(experimental/openapi): replace $ref in
 additionalProperties; add tests (#1735)

Co-authored-by: Hsiao, Cooper <cooper.hsiao@commscope.com>
Co-authored-by: Cooper <xcooper1030@gmail.com>
---
 .../experimental/utilities/openapi/schemas.py |   6 ++
 .../utilities/openapi/test_schemas.py         | 101 ++++++++++++++++++
 2 files changed, 107 insertions(+)

diff --git a/src/fastmcp/experimental/utilities/openapi/schemas.py b/src/fastmcp/experimental/utilities/openapi/schemas.py
index 234283fcb..54430b2ed 100644
--- a/src/fastmcp/experimental/utilities/openapi/schemas.py
+++ b/src/fastmcp/experimental/utilities/openapi/schemas.py
@@ -79,6 +79,7 @@ def _replace_ref_with_defs(
 
     Examples:
     - {"type": "object", "properties": {"$ref": "#/components/schemas/..."}}
+    - {"type": "object", "additionalProperties": {"$ref": "#/components/schemas/..."}, "properties": {...}}
     - {"$ref": "#/components/schemas/..."}
     - {"items": {"$ref": "#/components/schemas/..."}}
     - {"anyOf": [{"$ref": "#/components/schemas/..."}]}
@@ -117,6 +118,11 @@ def _replace_ref_with_defs(
     for section in ["anyOf", "allOf", "oneOf"]:
         for i, item in enumerate(schema.get(section, [])):
             schema[section][i] = _replace_ref_with_defs(item)
+    if additionalProperties := schema.get("additionalProperties"):
+        if not isinstance(additionalProperties, bool):
+            schema["additionalProperties"] = _replace_ref_with_defs(
+                additionalProperties
+            )
     if info.get("description", description) and not schema.get("description"):
         schema["description"] = description
     return schema
diff --git a/tests/experimental/openapi_parser/utilities/openapi/test_schemas.py b/tests/experimental/openapi_parser/utilities/openapi/test_schemas.py
index ea8c8e2f6..1f4966d6e 100644
--- a/tests/experimental/openapi_parser/utilities/openapi/test_schemas.py
+++ b/tests/experimental/openapi_parser/utilities/openapi/test_schemas.py
@@ -286,6 +286,107 @@ class TestSchemaProcessing:
         ]
         assert array_item_prop["$ref"] == "#/$defs/RefProp"
 
+    def test_replace_ref_with_defs_in_additional_properties(self):
+        """Test replacing $ref deeply in 'additionalProperties'."""
+
+        add_props_schema = {
+            "description": "An invoice with a fixed header and a flexible set of line items.",
+            "type": "object",
+            "properties": {
+                "invoice_number": {
+                    "type": "string",
+                    "description": "The unique identifier for the invoice.",
+                },
+                "customer_name": {
+                    "type": "string",
+                    "description": "The name of the customer.",
+                },
+                "total_amount": {
+                    "type": "number",
+                    "description": "The total amount of the invoice.",
+                },
+            },
+            "required": ["invoice_number", "customer_name", "total_amount"],
+            "additionalProperties": {"$ref": "#/components/schemas/Link"},
+        }
+
+        # Use our recursive replacement approach
+        result = _replace_ref_with_defs(add_props_schema)
+
+        # Check additional properties
+        add_props = result["additionalProperties"]
+        assert add_props["$ref"] == "#/$defs/Link"
+
+    def test_replace_ref_with_defs_with_bool_additional_properties(self):
+        """Test replacing a bool 'additionalProperties'."""
+
+        add_props_schema = {
+            "description": "An invoice with a fixed header and a flexible set of line items.",
+            "type": "object",
+            "properties": {
+                "invoice_number": {
+                    "type": "string",
+                    "description": "The unique identifier for the invoice.",
+                },
+                "customer_name": {
+                    "type": "string",
+                    "description": "The name of the customer.",
+                },
+                "total_amount": {
+                    "type": "number",
+                    "description": "The total amount of the invoice.",
+                },
+            },
+            "required": ["invoice_number", "customer_name", "total_amount"],
+            "additionalProperties": False,
+        }
+
+        # Use our recursive replacement approach
+        result = _replace_ref_with_defs(add_props_schema)
+
+        # Check additional properties
+        add_props = result["additionalProperties"]
+        assert add_props is False
+
+    def test_replace_ref_with_defs_with_inner_schema_additional_properties(self):
+        """Test replacing a inner schema 'additionalProperties'."""
+
+        add_props_schema = {
+            "description": "An invoice with a fixed header and a flexible set of line items.",
+            "type": "object",
+            "properties": {
+                "invoice_number": {
+                    "type": "string",
+                    "description": "The unique identifier for the invoice.",
+                },
+                "customer_name": {
+                    "type": "string",
+                    "description": "The name of the customer.",
+                },
+                "total_amount": {
+                    "type": "number",
+                    "description": "The total amount of the invoice.",
+                },
+            },
+            "required": ["invoice_number", "customer_name", "total_amount"],
+            "additionalProperties": {
+                "type": "integer",
+                "format": "int32",
+                "description": "The total amount of the invoice.",
+            },
+        }
+
+        # Use our recursive replacement approach
+        result = _replace_ref_with_defs(add_props_schema)
+
+        # Check additional properties
+        add_props = result["additionalProperties"]
+        assert add_props == {
+            "type": "integer",
+            "format": "int32",
+            "description": "The total amount of the invoice.",
+        }
+
     def test_parameter_collision_suffixing_logic(self):
         """Test the specific logic for parameter collision suffixing."""
         # Create a route that would definitely cause collisions