fix(core): resolve $ref to $defs defined in nested property schemas

python/packages/autogen-core/src/autogen_core/utils/_json_to_pydantic.py

@@ -297,6 +297,30 @@ def _extract_field_type(self, key: str, value: Dict[str, Any], model_name: str,
     def _json_schema_to_model(
         self, schema: Dict[str, Any], model_name: str, root_schema: Dict[str, Any]
     ) -> Type[BaseModel]:
+        # Process any $defs in this schema that haven't been processed yet.
+        # This handles nested schemas (e.g., a property with its own $defs)
+        # by merging them into the root schema's $defs so that _resolve_ref
+        # and get_ref can find them.
+        if "$defs" in schema and schema is not root_schema:
+            if "$defs" not in root_schema:
+                root_schema["$defs"] = {}
+            for def_name, def_schema in schema["$defs"].items():
+                if def_name not in root_schema["$defs"]:
+                    root_schema["$defs"][def_name] = def_schema
+            # Process object-type definitions into the model cache
+            for def_name in schema["$defs"]:
+                if def_name not in self._model_cache:
+                    def_schema = root_schema["$defs"][def_name]
+                    # Only cache object-type definitions as models;
+                    # enum and primitive types are resolved inline via _resolve_ref.
+                    if def_schema.get("type") == "object" and "properties" in def_schema:
+                        self._model_cache[def_name] = None
+            for def_name in list(self._model_cache):
+                if self._model_cache[def_name] is None and def_name in root_schema.get("$defs", {}):
+                    def_schema = root_schema["$defs"][def_name]
+                    if def_schema.get("type") == "object" and "properties" in def_schema:
+                        self._model_cache[def_name] = self.json_schema_to_pydantic(def_schema, def_name, root_schema)
+
         if "allOf" in schema:
             merged: Dict[str, Any] = {"type": "object", "properties": {}, "required": []}
             for s in schema["allOf"]:
@@ -315,7 +339,18 @@ def _json_schema_to_model(
         for key, value in schema.get("properties", {}).items():
             if "$ref" in value:
                 ref_name = value["$ref"].split("/")[-1]
-                field_type = self.get_ref(ref_name)
+                if ref_name in self._model_cache:
+                    field_type = self.get_ref(ref_name)
+                else:
+                    # Resolve inline for non-model definitions (enums, primitives, etc.)
+                    resolved = self._resolve_ref(value["$ref"], root_schema)
+                    merged_value = {**resolved, **{k: v for k, v in value.items() if k != "$ref"}}
+                    if "enum" in merged_value:
+                        field_type = Literal[tuple(merged_value["enum"])]
+                    elif merged_value.get("type") == "object" and "properties" in merged_value:
+                        field_type = self._json_schema_to_model(merged_value, f"{model_name}_{key}", root_schema)
+                    else:
+                        field_type = self._extract_field_type(key, merged_value, model_name, root_schema)
             elif "anyOf" in value:
                 sub_models = self._resolve_union_types(value["anyOf"])
                 field_type = Union[tuple(sub_models)]

python/packages/autogen-core/tests/test_json_to_pydantic.py

@@ -1042,3 +1042,128 @@ def test_nested_arrays_with_object_schemas() -> None:
     assert alice.name == "Alice"  # type: ignore[attr-defined]
     assert alice.role == "Senior Developer"  # type: ignore[attr-defined]
     assert alice.skills == ["Python", "JavaScript", "Docker"]  # type: ignore[attr-defined]
+
+
+def test_nested_defs_enum_ref() -> None:
+    """Test that $defs inside a nested property are resolved correctly.
+
+    Reproduces the bug from https://github.com/microsoft/autogen/issues/7129
+    where mcp_server_tools fails with ReferenceNotFoundError when a nested
+    property defines its own $defs with enum types referenced via $ref.
+    """
+    schema = {
+        "type": "object",
+        "properties": {
+            "base": {
+                "properties": {
+                    "query": {"type": "string"},
+                    "max_results": {"default": 100, "type": "integer", "minimum": 1, "maximum": 1000},
+                },
+                "required": ["query"],
+                "type": "object",
+            },
+            "windows_params": {
+                "$defs": {
+                    "WindowsSortOption": {
+                        "enum": [1, 2, 3, 4, 5, 6, 7, 8, 11, 12, 13, 14],
+                        "title": "WindowsSortOption",
+                        "type": "integer",
+                    }
+                },
+                "properties": {
+                    "match_path": {"default": False, "type": "boolean"},
+                    "sort_by": {"$ref": "#/$defs/WindowsSortOption", "default": 1},
+                },
+                "title": "WindowsSpecificParams",
+                "type": "object",
+            },
+        },
+        "required": ["base"],
+        "additionalProperties": False,
+    }
+
+    converter = _JSONSchemaToPydantic()
+    Model = converter.json_schema_to_pydantic(schema, "SearchModel")
+
+    assert "base" in Model.model_fields
+    assert "windows_params" in Model.model_fields
+
+    # Test that the model can be instantiated with valid data
+    instance = Model(base={"query": "test"}, windows_params={"sort_by": 3, "match_path": True})
+    assert instance.base.query == "test"  # type: ignore[attr-defined]
+    assert instance.windows_params.sort_by == 3  # type: ignore[attr-defined]
+    assert instance.windows_params.match_path is True  # type: ignore[attr-defined]
+
+    # Test with default values
+    instance2 = Model(base={"query": "test2"})
+    assert instance2.windows_params is None  # type: ignore[attr-defined]
+
+
+def test_nested_defs_object_ref() -> None:
+    """Test that $defs inside a nested property with object-type definitions work."""
+    schema = {
+        "type": "object",
+        "properties": {
+            "config": {
+                "$defs": {
+                    "SubConfig": {
+                        "type": "object",
+                        "properties": {
+                            "enabled": {"type": "boolean", "default": True},
+                            "level": {"type": "integer", "default": 1},
+                        },
+                        "title": "SubConfig",
+                    }
+                },
+                "properties": {
+                    "name": {"type": "string"},
+                    "sub": {"$ref": "#/$defs/SubConfig"},
+                },
+                "required": ["name"],
+                "type": "object",
+            }
+        },
+        "required": ["config"],
+    }
+
+    converter = _JSONSchemaToPydantic()
+    Model = converter.json_schema_to_pydantic(schema, "ConfigModel")
+
+    instance = Model(config={"name": "test", "sub": {"enabled": False, "level": 3}})
+    assert instance.config.name == "test"  # type: ignore[attr-defined]
+    assert instance.config.sub.enabled is False  # type: ignore[attr-defined]
+    assert instance.config.sub.level == 3  # type: ignore[attr-defined]
+
+
+def test_multiple_nested_defs() -> None:
+    """Test multiple properties each having their own $defs."""
+    schema = {
+        "type": "object",
+        "properties": {
+            "param_a": {
+                "$defs": {
+                    "TypeA": {"enum": ["x", "y", "z"], "type": "string"}
+                },
+                "properties": {
+                    "value": {"$ref": "#/$defs/TypeA", "default": "x"}
+                },
+                "type": "object",
+            },
+            "param_b": {
+                "$defs": {
+                    "TypeB": {"enum": [10, 20, 30], "type": "integer"}
+                },
+                "properties": {
+                    "count": {"$ref": "#/$defs/TypeB", "default": 10}
+                },
+                "type": "object",
+            },
+        },
+    }
+
+    converter = _JSONSchemaToPydantic()
+    Model = converter.json_schema_to_pydantic(schema, "MultiDefModel")
+
+    instance = Model(param_a={"value": "y"}, param_b={"count": 20})
+    assert instance.param_a.value == "y"  # type: ignore[attr-defined]
+    assert instance.param_b.count == 20  # type: ignore[attr-defined]