diff --git a/docs/integrations/claude-desktop.mdx b/docs/integrations/claude-desktop.mdx index a69bebb0b..1e46e47a1 100644 --- a/docs/integrations/claude-desktop.mdx +++ b/docs/integrations/claude-desktop.mdx @@ -48,6 +48,79 @@ if __name__ == "__main__": mcp.run() ``` +## Claude Desktop Compatibility + + +Claude Desktop fails to properly handle JSON schemas that contain `$ref` references, sending null values instead of valid parameter values. This commonly occurs with complex FastMCP servers that use nested Pydantic models, enums, or advanced type annotations, causing parameter validation errors. + + +### Schema Compatibility Issues + +When FastMCP generates JSON schemas for complex Python types, it creates `$defs` sections with `$ref` references for reusable components: + +```json Problematic Schema (causes null parameters) +{ + "$defs": { + "TestColor": {"enum": ["red", "green", "blue"], title: "TestColor", "type": "string"} + }, + "properties": { + "color": {"$ref": "#/$defs/TestColor"} + } +} +``` + +**Result**: Claude Desktop sends null values instead of valid enum options: +```json +{ + "color": null // ❌ Should be "red", "green", or "blue" +} +``` + +### Solution: Schema Dereferencing + +FastMCP provides automatic schema dereferencing that resolves `$ref` references in properties while preserving the original `$defs` section: + +```json Compatible Schema (works with Claude Desktop) +{ + "$defs": { + "TestColor": {"enum": ["red", "green", "blue"], title: "TestColor", "type": "string"} + }, + "properties": { + "color": {"enum": ["red", "green", "blue"], title: "TestColor", "type": "string"} + } +} +``` + +**Result**: Claude Desktop now sends proper parameter values: +```json +{ + "color": "red" // ✅ Valid enum value +} +``` + +### Enabling Dereferencing + + + +Enable dereferencing in your server code: + +```python server.py {3,4} +from fastmcp import FastMCP +import fastmcp + +fastmcp.settings.dereference_json_schemas = True + +mcp = FastMCP(name="Compatible Test") + +# Your tools with complex types will now be Claude Desktop compatible +@mcp.tool +def test_enum(color: TestColor) -> dict: + """Process a complex task - schema will be automatically dereferenced.""" + return {"status": "success", "color": color.value} +``` + + + ## Install the Server ### FastMCP CLI diff --git a/src/fastmcp/prompts/prompt.py b/src/fastmcp/prompts/prompt.py index 1477fab53..13e7b8531 100644 --- a/src/fastmcp/prompts/prompt.py +++ b/src/fastmcp/prompts/prompt.py @@ -14,6 +14,7 @@ from mcp.types import PromptArgument as MCPPromptArgument from pydantic import Field, TypeAdapter +import fastmcp from fastmcp.exceptions import PromptError from fastmcp.server.dependencies import get_context from fastmcp.utilities.components import FastMCPComponent @@ -195,7 +196,11 @@ def from_function( else: prune_params = None - parameters = compress_schema(parameters, prune_params=prune_params) + parameters = compress_schema( + parameters, + prune_params=prune_params, + dereference_refs=fastmcp.settings.dereference_json_schemas, + ) # Convert parameters to PromptArguments arguments: list[PromptArgument] = [] diff --git a/src/fastmcp/resources/template.py b/src/fastmcp/resources/template.py index 1b23081d2..b725093be 100644 --- a/src/fastmcp/resources/template.py +++ b/src/fastmcp/resources/template.py @@ -15,6 +15,7 @@ validate_call, ) +import fastmcp from fastmcp.resources.resource import Resource from fastmcp.server.dependencies import get_context from fastmcp.utilities.components import FastMCPComponent @@ -274,7 +275,11 @@ def from_function( # compress the schema prune_params = [context_kwarg] if context_kwarg else None - parameters = compress_schema(parameters, prune_params=prune_params) + parameters = compress_schema( + parameters, + prune_params=prune_params, + dereference_refs=fastmcp.settings.dereference_json_schemas, + ) # ensure the arguments are properly cast fn = validate_call(fn) diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py index bcdf93c6d..ddad511e6 100644 --- a/src/fastmcp/settings.py +++ b/src/fastmcp/settings.py @@ -212,6 +212,26 @@ def normalize_log_level(cls, v): ), ] = "path" + dereference_json_schemas: Annotated[ + bool, + Field( + default=False, + description=inspect.cleandoc( + """ + If True, all JSON schemas generated for tools, prompts, and resources will have + their $refs resolved in properties while preserving the original $defs section. + This fixes compatibility issues with clients like Claude Desktop that fail to properly + handle $ref references, sending null values instead of valid enum parameters. + + When enabled, schemas with references like {"$ref": "#/$defs/EnumName"} in properties + will be expanded to include the full definition inline, while keeping the $defs section + intact for reference. This prevents parameter validation errors and ensures clients + can properly generate forms and send valid parameter values. + """ + ), + ), + ] = False + client_init_timeout: Annotated[ float | None, Field( diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py index 0cb31523f..a1255b6ae 100644 --- a/src/fastmcp/tools/tool.py +++ b/src/fastmcp/tools/tool.py @@ -19,6 +19,7 @@ from mcp.types import Tool as MCPTool from pydantic import Field, PydanticSchemaGenerationError +import fastmcp from fastmcp.server.dependencies import get_context from fastmcp.utilities.components import FastMCPComponent from fastmcp.utilities.json_schema import compress_schema @@ -376,7 +377,11 @@ def from_function( input_type_adapter = get_cached_typeadapter(fn) input_schema = input_type_adapter.json_schema() - input_schema = compress_schema(input_schema, prune_params=prune_params) + input_schema = compress_schema( + input_schema, + prune_params=prune_params, + dereference_refs=fastmcp.settings.dereference_json_schemas, + ) output_schema = None # Get the return annotation from the signature @@ -436,7 +441,10 @@ def from_function( else: output_schema = base_schema - output_schema = compress_schema(output_schema) + output_schema = compress_schema( + output_schema, + dereference_refs=fastmcp.settings.dereference_json_schemas, + ) except PydanticSchemaGenerationError as e: if "_UnserializableType" not in str(e): diff --git a/src/fastmcp/utilities/json_schema.py b/src/fastmcp/utilities/json_schema.py index d0d9e37cd..ded24c999 100644 --- a/src/fastmcp/utilities/json_schema.py +++ b/src/fastmcp/utilities/json_schema.py @@ -1,6 +1,176 @@ from __future__ import annotations from collections import defaultdict +from copy import deepcopy +from typing import Any + + +def _detect_self_reference(schema: dict) -> bool: + """ + Detect if the schema contains self-referencing definitions. + + Args: + schema: The JSON schema to check + + Returns: + True if self-referencing is detected + """ + defs = schema.get("$defs", {}) + + def find_refs_in_value(value: Any, parent_def: str) -> bool: + """Check if a value contains a reference to its parent definition.""" + if isinstance(value, dict): + if "$ref" in value: + ref_path = value["$ref"] + # Check if this references the parent definition + if ref_path == f"#/$defs/{parent_def}": + return True + # Check all values in the dict + for v in value.values(): + if find_refs_in_value(v, parent_def): + return True + elif isinstance(value, list): + # Check all items in the list + for item in value: + if find_refs_in_value(item, parent_def): + return True + return False + + # Check each definition for self-reference + for def_name, def_content in defs.items(): + if find_refs_in_value(def_content, def_name): + # Self-reference detected, return original schema + return True + + return False + + +def dereference_json_schema(schema: dict, max_depth: int = 50) -> dict: + """ + Dereference a JSON schema by resolving $ref references while preserving $defs only when corner cases occur. + + This function flattens schema properties by: + 1. Check for self-reference - if found, return original schema with $defs + 2. When encountering $refs in properties, resolve them on-demand + 3. Track visited definitions globally to prevent circular expansion + 4. Only preserve original $defs if corner cases are encountered: + - Self-reference detected + - Circular references between definitions + - Reference depth exceeds max_depth + - Reference not found in $defs + + Args: + schema: The JSON schema to flatten + max_depth: Maximum depth for resolving references (default: 5) + + Returns: + Schema with references resolved in properties, keeping $defs only when corner cases occur + """ + # Step 1: Check for self-reference + if _detect_self_reference(schema): + # Self-referencing detected, return original schema with $defs + return schema + + # Make a deep copy to work with + result = deepcopy(schema) + + # Keep original $defs for potential corner cases + defs = deepcopy(schema.get("$defs", {})) + + # Track corner cases that require preserving $defs + corner_cases_detected = { + "circular_ref": False, + "max_depth_reached": False, + "ref_not_found": False, + } + + # Step 2: Define resolution function that tracks visits globally and corner cases + def resolve_refs_in_value(value: Any, depth: int, visiting: set[str]) -> Any: + """ + Recursively resolve $refs in a value. + + Args: + value: The value to process + depth: Current depth in resolution + visiting: Set of definitions currently being resolved (for cycle detection) + + Returns: + Value with $refs resolved (or kept if corner cases occur) + """ + if depth >= max_depth: + corner_cases_detected["max_depth_reached"] = True + return value + + if isinstance(value, dict): + if "$ref" in value: + ref_path = value["$ref"] + + # Only handle internal references to $defs + if ref_path.startswith("#/$defs/"): + def_name = ref_path.split("/")[-1] + + # Check for circular reference + if def_name in visiting: + # Circular reference detected, keep the $ref + corner_cases_detected["circular_ref"] = True + return value + + if def_name in defs: + # Add to visiting set + visiting.add(def_name) + + # Get the definition and resolve any refs within it + resolved = resolve_refs_in_value( + deepcopy(defs[def_name]), depth + 1, visiting + ) + + # Remove from visiting set + visiting.remove(def_name) + + # Merge resolved definition with additional properties + # Additional properties from the original object take precedence + for key, val in value.items(): + if key != "$ref": + resolved[key] = val + + return resolved + else: + # Definition not found, keep the $ref + corner_cases_detected["ref_not_found"] = True + return value + else: + # External ref or other type - keep as is + return value + else: + # Regular dict - process all values + return { + key: resolve_refs_in_value(val, depth, visiting) + for key, val in value.items() + } + elif isinstance(value, list): + # Process each item in the list + return [resolve_refs_in_value(item, depth, visiting) for item in value] + else: + # Primitive value - return as is + return value + + # Step 3: Process main schema properties with shared visiting set + for key, value in result.items(): + if key != "$defs": + # Each top-level property gets its own visiting set + # This allows the same definition to be used in different contexts + result[key] = resolve_refs_in_value(value, 0, set()) + + # Step 4: Conditionally preserve $defs based on corner cases + if any(corner_cases_detected.values()): + # Corner case detected, preserve original $defs + if "$defs" in schema: # Only add if original schema had $defs + result["$defs"] = defs + else: + # No corner cases, remove $defs if it exists + result.pop("$defs", None) + + return result def _prune_param(schema: dict, param: str) -> dict: @@ -186,6 +356,7 @@ def compress_schema( prune_defs: bool = True, prune_additional_properties: bool = True, prune_titles: bool = False, + dereference_refs: bool = False, ) -> dict: """ Remove the given parameters from the schema. @@ -196,6 +367,7 @@ def compress_schema( prune_defs: Whether to remove unused definitions prune_additional_properties: Whether to remove additionalProperties: false prune_titles: Whether to remove title fields from the schema + dereference_refs: Whether to completely flatten by inlining all $refs (fixes Claude Desktop crashes). """ # Remove specific parameters if requested for param in prune_params or []: @@ -210,4 +382,8 @@ def compress_schema( prune_defs=prune_defs, ) + # Dereference all $refs if requested + if dereference_refs: + schema = dereference_json_schema(schema) + return schema diff --git a/tests/utilities/test_json_schema.py b/tests/utilities/test_json_schema.py index 4d5d8a1e9..9bf3d20b0 100644 --- a/tests/utilities/test_json_schema.py +++ b/tests/utilities/test_json_schema.py @@ -1,6 +1,8 @@ from fastmcp.utilities.json_schema import ( + _detect_self_reference, _prune_param, compress_schema, + dereference_json_schema, ) # Wrapper for backward compatibility with tests @@ -432,3 +434,868 @@ def test_prune_nested_additional_properties(self): "additionalProperties" not in result["properties"]["foo"]["properties"]["nested"] ) + + +class TestDetectSelfReference: + """Tests for the _detect_self_reference function.""" + + def test_no_self_reference(self): + """Test schema with normal references (no self-references).""" + schema = { + "$defs": { + "Color": {"enum": ["red", "green", "blue"], "type": "string"}, + "Size": {"enum": ["small", "medium", "large"], "type": "string"}, + }, + "properties": { + "color": {"$ref": "#/$defs/Color"}, + "size": {"$ref": "#/$defs/Size"}, + }, + } + assert not _detect_self_reference(schema) + + def test_cross_references_not_self_references(self): + """Test that cross-references (A->B->A) are not detected as self-references.""" + schema = { + "$defs": { + "A": {"type": "object", "properties": {"b": {"$ref": "#/$defs/B"}}}, + "B": {"type": "object", "properties": {"a": {"$ref": "#/$defs/A"}}}, + }, + "properties": {"root": {"$ref": "#/$defs/A"}}, + } + assert not _detect_self_reference(schema) + + def test_direct_self_reference(self): + """Test detection of direct self-reference (Node -> Node).""" + schema = { + "$defs": { + "Node": { + "type": "object", + "properties": {"child": {"$ref": "#/$defs/Node"}}, + } + }, + "properties": {"root": {"$ref": "#/$defs/Node"}}, + } + assert _detect_self_reference(schema) + + +class TestDereferenceJsonSchema: + """Tests for the dereference_json_schema function.""" + + # ===== Basic Behavior ===== + + def test_empty_schema(self): + """Test dereferencing an empty schema.""" + schema = {} + result = dereference_json_schema(schema) + assert result == {} + + def test_schema_without_defs(self): + """Test dereferencing a schema without $defs.""" + schema = { + "type": "object", + "properties": {"name": {"type": "string"}, "age": {"type": "integer"}}, + } + result = dereference_json_schema(schema) + assert result == schema + + def test_basic_reference_resolution(self): + """Test basic reference resolution removes $defs for simple schemas and resolves properties.""" + schema = { + "$defs": {"Color": {"enum": ["red", "green", "blue"], "type": "string"}}, + "properties": {"color": {"$ref": "#/$defs/Color"}}, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "color": {"enum": ["red", "green", "blue"], "type": "string"} + }, + } + assert result == expected + assert "$defs" not in result # No corner cases, so $defs should be removed + + # ===== Schema Structure Variations ===== + + def test_deep_nesting_references(self): + """Test dereferencing with multiple levels of nested references.""" + schema = { + "$defs": { + "Level3": {"type": "string", "enum": ["a", "b", "c"]}, + "Level2": { + "type": "object", + "properties": {"level3": {"$ref": "#/$defs/Level3"}}, + }, + "Level1": { + "type": "object", + "properties": {"level2": {"$ref": "#/$defs/Level2"}}, + }, + }, + "properties": {"root": {"$ref": "#/$defs/Level1"}}, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "root": { + "type": "object", + "properties": { + "level2": { + "type": "object", + "properties": { + "level3": {"type": "string", "enum": ["a", "b", "c"]} + }, + } + }, + } + }, + } + assert result == expected + assert "$defs" not in result + + def test_mixed_refs_and_non_refs(self): + """Test schema with mix of references and direct definitions.""" + schema = { + "$defs": {"Status": {"enum": ["active", "inactive"], "type": "string"}}, + "properties": { + "id": {"type": "integer"}, + "name": {"type": "string"}, + "status": {"$ref": "#/$defs/Status"}, + "metadata": { + "type": "object", + "properties": { + "created": {"type": "string", "format": "date-time"} + }, + }, + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "id": {"type": "integer"}, + "name": {"type": "string"}, + "status": {"enum": ["active", "inactive"], "type": "string"}, + "metadata": { + "type": "object", + "properties": { + "created": {"type": "string", "format": "date-time"} + }, + }, + }, + } + assert result == expected + assert "$defs" not in result + + def test_reference_in_oneof(self): + """Test dereferencing references within oneOf constructs.""" + schema = { + "$defs": { + "Dog": {"type": "object", "properties": {"breed": {"type": "string"}}}, + "Cat": {"type": "object", "properties": {"color": {"type": "string"}}}, + }, + "properties": { + "pet": {"oneOf": [{"$ref": "#/$defs/Dog"}, {"$ref": "#/$defs/Cat"}]} + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "pet": { + "oneOf": [ + {"type": "object", "properties": {"breed": {"type": "string"}}}, + {"type": "object", "properties": {"color": {"type": "string"}}}, + ] + } + }, + } + assert result == expected + + def test_reference_in_allof(self): + """Test dereferencing references within allOf constructs.""" + schema = { + "$defs": { + "Base": {"type": "object", "properties": {"id": {"type": "integer"}}}, + "Extended": { + "type": "object", + "properties": {"name": {"type": "string"}}, + }, + }, + "properties": { + "item": { + "allOf": [{"$ref": "#/$defs/Base"}, {"$ref": "#/$defs/Extended"}] + } + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "item": { + "allOf": [ + {"type": "object", "properties": {"id": {"type": "integer"}}}, + {"type": "object", "properties": {"name": {"type": "string"}}}, + ] + } + }, + } + assert result == expected + + def test_conditional_schemas_with_refs(self): + """Test dereferencing with if/then/else containing references.""" + schema = { + "$defs": { + "StringType": {"type": "string"}, + "NumberType": {"type": "number"}, + "NamePattern": {"pattern": "^[A-Za-z]+$"}, + }, + "if": {"$ref": "#/$defs/StringType"}, + "then": {"$ref": "#/$defs/NamePattern"}, + "else": {"$ref": "#/$defs/NumberType"}, + } + result = dereference_json_schema(schema) + + expected = { + "if": {"type": "string"}, + "then": {"pattern": "^[A-Za-z]+$"}, + "else": {"type": "number"}, + } + assert result == expected + assert "$defs" not in result + + def test_pattern_properties_with_refs(self): + """Test dereferencing references within patternProperties.""" + schema = { + "$defs": { + "EmailType": {"type": "string", "format": "email"}, + "PhoneType": {"type": "string", "pattern": "^\\+?[1-9]\\d{1,14}$"}, + }, + "type": "object", + "patternProperties": { + "^email": {"$ref": "#/$defs/EmailType"}, + "^phone": {"$ref": "#/$defs/PhoneType"}, + }, + } + result = dereference_json_schema(schema) + + expected = { + "type": "object", + "patternProperties": { + "^email": {"type": "string", "format": "email"}, + "^phone": {"type": "string", "pattern": "^\\+?[1-9]\\d{1,14}$"}, + }, + } + assert result == expected + + def test_additional_properties_with_refs(self): + """Test dereferencing references in additionalProperties.""" + schema = { + "$defs": { + "FlexibleValue": { + "oneOf": [ + {"type": "string"}, + {"type": "number"}, + {"type": "boolean"}, + ] + } + }, + "type": "object", + "properties": {"name": {"type": "string"}}, + "additionalProperties": {"$ref": "#/$defs/FlexibleValue"}, + } + result = dereference_json_schema(schema) + + expected = { + "type": "object", + "properties": {"name": {"type": "string"}}, + "additionalProperties": { + "oneOf": [{"type": "string"}, {"type": "number"}, {"type": "boolean"}] + }, + } + assert result == expected + + def test_refs_in_complex_compositions(self): + """Test references in complex schema compositions.""" + schema = { + "$defs": { + "Animal": { + "type": "object", + "properties": {"name": {"type": "string"}}, + }, + "Dog": { + "type": "object", + "properties": { + "breed": {"type": "string"}, + "isGoodBoy": {"type": "boolean", "default": True}, + }, + }, + "Cat": { + "type": "object", + "properties": { + "livesLeft": {"type": "integer", "minimum": 1, "maximum": 9} + }, + }, + }, + "properties": { + "pet": { + "allOf": [ + {"$ref": "#/$defs/Animal"}, + {"anyOf": [{"$ref": "#/$defs/Dog"}, {"$ref": "#/$defs/Cat"}]}, + ] + } + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "pet": { + "allOf": [ + {"type": "object", "properties": {"name": {"type": "string"}}}, + { + "anyOf": [ + { + "type": "object", + "properties": { + "breed": {"type": "string"}, + "isGoodBoy": { + "type": "boolean", + "default": True, + }, + }, + }, + { + "type": "object", + "properties": { + "livesLeft": { + "type": "integer", + "minimum": 1, + "maximum": 9, + } + }, + }, + ] + }, + ] + } + }, + } + assert result == expected + + # ===== Property Merging and Precedence ===== + + def test_property_merging_during_resolution(self): + """Test that additional properties are merged when resolving references.""" + schema = { + "$defs": {"BaseString": {"type": "string", "minLength": 1}}, + "properties": { + "name": { + "$ref": "#/$defs/BaseString", + "title": "Full Name", + "default": "Anonymous", + "maxLength": 100, + } + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "name": { + "type": "string", + "minLength": 1, + "title": "Full Name", + "default": "Anonymous", + "maxLength": 100, + } + }, + } + assert result == expected + + def test_property_override_precedence(self): + """Test that properties in referring object take precedence over referenced ones.""" + schema = { + "$defs": { + "BaseType": { + "type": "string", + "title": "Base Title", + "description": "Base description", + } + }, + "properties": { + "field": { + "$ref": "#/$defs/BaseType", + "title": "Override Title", # This should override the base title + "default": "default_value", # This should be added + } + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "field": { + "title": "Override Title", # Should keep the override + "default": "default_value", # Should keep the additional property + "type": "string", # Should get from base + "description": "Base description", # Should get from base + } + }, + } + assert result == expected + + def test_refs_with_additional_keywords_complex(self): + """Test references combined with many additional keywords.""" + schema = { + "$defs": {"BaseString": {"type": "string", "minLength": 1}}, + "properties": { + "complex_field": { + "$ref": "#/$defs/BaseString", + "title": "Complex Field", + "description": "A field with many constraints", + "default": "default_value", + "examples": ["example1", "example2"], + "maxLength": 50, + "pattern": "^[a-zA-Z0-9]+$", + "format": "alphanumeric", + } + }, + } + result = dereference_json_schema(schema) + + expected = { + "properties": { + "complex_field": { + "title": "Complex Field", + "description": "A field with many constraints", + "default": "default_value", + "examples": ["example1", "example2"], + "maxLength": 50, + "pattern": "^[a-zA-Z0-9]+$", + "format": "alphanumeric", + "type": "string", # From the referenced schema + "minLength": 1, # From the referenced schema + } + }, + } + assert result == expected + + # ===== Corner Cases (Preserving $defs) ===== + + def test_self_reference_preserves_defs(self): + """Test that self-referencing schemas preserve $defs (corner case).""" + schema = { + "$defs": { + "Node": { + "type": "object", + "properties": {"child": {"$ref": "#/$defs/Node"}}, + } + }, + "properties": {"root": {"$ref": "#/$defs/Node"}}, + } + result = dereference_json_schema(schema) + assert result == schema # Should be unchanged due to self-reference + assert "$defs" in result + + def test_circular_reference_preserves_defs(self): + """Test that circular references preserve $defs (corner case).""" + schema = { + "$defs": { + "NodeA": { + "type": "object", + "properties": { + "value": {"type": "string"}, + "nodeB": {"$ref": "#/$defs/NodeB"}, + }, + }, + "NodeB": { + "type": "object", + "properties": { + "value": {"type": "integer"}, + "nodeA": {"$ref": "#/$defs/NodeA"}, # Circular reference + }, + }, + }, + "properties": {"root": {"$ref": "#/$defs/NodeA"}}, + } + result = dereference_json_schema(schema) + assert "$defs" in result # Corner case detected, $defs preserved + + # The function should do partial resolution but preserve circular refs + # Original $defs should be unchanged + assert result["$defs"] == schema["$defs"] + + # Root should be partially resolved but contain circular ref + assert result["properties"]["root"]["type"] == "object" + assert result["properties"]["root"]["properties"]["value"]["type"] == "string" + assert "$ref" in str( + result["properties"]["root"]["properties"]["nodeB"]["properties"]["nodeA"] + ) + + def test_missing_reference_preserves_defs(self): + """Test that missing references preserve $defs (corner case).""" + schema = { + "$defs": {"Color": {"enum": ["red", "green", "blue"], "type": "string"}}, + "properties": { + "color": {"$ref": "#/$defs/NonExistent"} + }, # Missing reference + } + result = dereference_json_schema(schema) + assert "$defs" in result # Corner case detected, $defs preserved + assert result["properties"]["color"] == { + "$ref": "#/$defs/NonExistent" + } # Ref preserved + + # ===== Depth Limit Handling and Performance ===== + + def test_max_depth_within_limit_removes_defs(self): + """Test that staying within max_depth limit removes $defs (no corner case).""" + # Create a 5-level chain with max_depth=6 (within limit) + schema = { + "$defs": { + "Level1": {"$ref": "#/$defs/Level2"}, + "Level2": {"$ref": "#/$defs/Level3"}, + "Level3": {"$ref": "#/$defs/Level4"}, + "Level4": {"$ref": "#/$defs/Level5"}, + "Level5": {"type": "string", "maxLength": 100}, + }, + "properties": {"depth_field": {"$ref": "#/$defs/Level1"}}, + } + result = dereference_json_schema(schema, max_depth=6) + + # Should resolve fully and remove $defs (within depth limit, no corner case) + expected = { + "properties": {"depth_field": {"type": "string", "maxLength": 100}}, + } + assert result == expected + assert "$defs" not in result # Within limit, should resolve fully + + def test_max_depth_exceeded_preserves_defs(self): + """Test that exceeding max_depth preserves $defs (corner case).""" + # Create a chain longer than max_depth (6 levels vs max_depth=5) + schema = { + "$defs": { + "Level1": {"$ref": "#/$defs/Level2"}, + "Level2": {"$ref": "#/$defs/Level3"}, + "Level3": {"$ref": "#/$defs/Level4"}, + "Level4": {"$ref": "#/$defs/Level5"}, + "Level5": {"$ref": "#/$defs/Level6"}, + "Level6": {"type": "string", "maxLength": 100}, + }, + "properties": {"deep_field": {"$ref": "#/$defs/Level1"}}, + } + result = dereference_json_schema(schema, max_depth=5) + + # Should preserve $defs when max_depth is exceeded (corner case detected) + assert "$defs" in result + assert "deep_field" in result["properties"] + + result_small_depth = dereference_json_schema(schema, max_depth=2) + assert "$defs" in result_small_depth # Corner case: max_depth exceeded + + @staticmethod + def _generate_linear_nested_schema(depth: int) -> dict: + """ + Generate a JSON schema with linear nesting of specified depth. + Helper method for testing deep schemas. + """ + if depth <= 0: + return {"type": "string", "maxLength": 100} + + schema = { + "$defs": {}, + "properties": {"nested_field": {"$ref": "#/$defs/Level1"}}, + } + + # Create the chain of references + for i in range(1, depth + 1): + if i == depth: + # Last level - actual type definition + schema["$defs"][f"Level{i}"] = { + "type": "string", + "maxLength": 100, + "description": f"Final level {i}", + } + else: + # Intermediate level - reference to next level + schema["$defs"][f"Level{i}"] = { + "$ref": f"#/$defs/Level{i + 1}", + "description": f"Level {i} pointing to Level {i + 1}", + } + + return schema + + def test_generated_depth_50_within_limit(self): + """Test Python-generated 50-level deep schema with sufficient max_depth.""" + schema = self._generate_linear_nested_schema(50) + + # Test with max_depth=60 (higher than schema depth) to ensure full resolution + result = dereference_json_schema(schema, max_depth=60) + + # Should resolve fully and remove $defs (no corner case) + # Note: The description comes from Level 1 due to property precedence rules + expected = { + "properties": { + "nested_field": { + "type": "string", + "maxLength": 100, + "description": "Level 1 pointing to Level 2", + } + }, + } + assert result == expected + assert "$defs" not in result # Full resolution, no corner case + + def test_generated_depth_50_reaches_default_limit(self): + """Test Python-generated 50-level deep schema reaching new default max_depth=50.""" + schema = self._generate_linear_nested_schema(50) + + # Test with default max_depth=50 (equal to schema depth) + result = dereference_json_schema(schema) # Uses default max_depth=50 + + # Since depth==max_depth, this hits the limit and preserves $defs (corner case) + assert ( + "$defs" in result + ) # Corner case: depth reaches max_depth limit, $defs preserved + assert len(result["$defs"]) == 50 # All 50 levels should be preserved + + def test_generated_depth_55_exceeds_default_limit(self): + """Test Python-generated 55-level deep schema exceeding new default max_depth=50.""" + schema = self._generate_linear_nested_schema(55) + + # Test with default max_depth=50 (less than schema depth) + result = dereference_json_schema(schema) # Uses default max_depth=50 + + # Should preserve $defs due to max_depth exceeded (corner case) + assert "$defs" in result # Corner case: max_depth exceeded, $defs preserved + assert len(result["$defs"]) == 55 # All 55 levels should be preserved + + # Should still do partial resolution for accessible levels + assert "nested_field" in result["properties"] + + def test_generated_depth_100_full_resolution(self): + """Test Python-generated 100-level deep schema with full resolution.""" + schema = self._generate_linear_nested_schema(100) + + # Test with max_depth=120 to ensure full resolution + result = dereference_json_schema(schema, max_depth=120) + + # Should resolve fully despite the depth + # Note: The description comes from Level 1 due to property precedence rules + expected = { + "properties": { + "nested_field": { + "type": "string", + "maxLength": 100, + "description": "Level 1 pointing to Level 2", + } + }, + } + assert result == expected + assert "$defs" not in result # Full resolution, no corner case + + def test_generated_depth_200_performance_boundary(self): + """Test Python-generated 200-level deep schema to verify performance boundaries.""" + schema = self._generate_linear_nested_schema(200) + + # Test that it can handle 200 levels when max_depth allows it + result = dereference_json_schema(schema, max_depth=250) + + # Should resolve fully + # Note: The description comes from Level 1 due to property precedence rules + expected = { + "properties": { + "nested_field": { + "type": "string", + "maxLength": 100, + "description": "Level 1 pointing to Level 2", + } + }, + } + assert result == expected + assert "$defs" not in result # Full resolution + + # Test that it preserves structure when max_depth is limited + result_limited = dereference_json_schema(schema, max_depth=10) + assert "$defs" in result_limited # Corner case: max_depth exceeded + assert len(result_limited["$defs"]) == 200 # All levels preserved + + def test_generated_depth_comparison_with_manual_schema(self): + """Compare Python-generated deep schema behavior with manually created schema.""" + # Generate a 10-level schema programmatically + generated_schema = self._generate_linear_nested_schema(10) + + # Create equivalent manual schema + manual_schema = { + "$defs": { + "Level1": { + "$ref": "#/$defs/Level2", + "description": "Level 1 pointing to Level 2", + }, + "Level2": { + "$ref": "#/$defs/Level3", + "description": "Level 2 pointing to Level 3", + }, + "Level3": { + "$ref": "#/$defs/Level4", + "description": "Level 3 pointing to Level 4", + }, + "Level4": { + "$ref": "#/$defs/Level5", + "description": "Level 4 pointing to Level 5", + }, + "Level5": { + "$ref": "#/$defs/Level6", + "description": "Level 5 pointing to Level 6", + }, + "Level6": { + "$ref": "#/$defs/Level7", + "description": "Level 6 pointing to Level 7", + }, + "Level7": { + "$ref": "#/$defs/Level8", + "description": "Level 7 pointing to Level 8", + }, + "Level8": { + "$ref": "#/$defs/Level9", + "description": "Level 8 pointing to Level 9", + }, + "Level9": { + "$ref": "#/$defs/Level10", + "description": "Level 9 pointing to Level 10", + }, + "Level10": { + "type": "string", + "maxLength": 100, + "description": "Final level 10", + }, + }, + "properties": {"nested_field": {"$ref": "#/$defs/Level1"}}, + } + + # Both should behave identically when dereferenced + generated_result = dereference_json_schema(generated_schema, max_depth=15) + manual_result = dereference_json_schema(manual_schema, max_depth=15) + + assert generated_result == manual_result + assert "$defs" not in generated_result # Both should fully resolve + assert "$defs" not in manual_result + + +class TestCompressSchemaWithDereference: + """Tests for schema flattening via the dereference_refs parameter in compress_schema.""" + + def test_integration_with_compress_schema(self): + """Test that dereference_refs parameter works with compress_schema.""" + schema = { + "$defs": { + "Priority": {"enum": ["low", "medium", "high"], "type": "string"} + }, + "properties": { + "title": {"type": "string"}, + "priority": {"$ref": "#/$defs/Priority"}, + "context_param": {"type": "string"}, # This will be pruned + }, + "required": ["title", "priority", "context_param"], + } + + # Test with flattening enabled + result = compress_schema( + schema, prune_params=["context_param"], dereference_refs=True + ) + + expected = { + "properties": { + "title": {"type": "string"}, + "priority": {"enum": ["low", "medium", "high"], "type": "string"}, + }, + "required": ["title", "priority"], + } + assert result == expected + assert "context_param" not in result["properties"] + + def test_compress_schema_with_self_references(self): + """Test compress_schema behavior with self-referencing schemas.""" + schema = { + "$defs": { + "Node": { + "type": "object", + "properties": { + "value": {"type": "string"}, + "child": {"$ref": "#/$defs/Node"}, + }, + } + }, + "properties": { + "tree": {"$ref": "#/$defs/Node"}, + "unused_param": {"type": "string"}, + }, + "required": ["tree", "unused_param"], + } + + result = compress_schema( + schema, prune_params=["unused_param"], dereference_refs=True + ) + + # Self-referencing schema should be returned with only parameter pruning applied + expected = { + "$defs": { + "Node": { + "type": "object", + "properties": { + "value": {"type": "string"}, + "child": {"$ref": "#/$defs/Node"}, + }, + } + }, + "properties": { + "tree": {"$ref": "#/$defs/Node"} # Should remain as reference + }, + "required": ["tree"], + } + assert result == expected + assert "$defs" in result # Corner case preserves defs + + def test_compress_schema_multiple_operations(self): + """Test compress_schema with multiple operations including flattening.""" + schema = { + "type": "object", + "title": "TestSchema", + "additionalProperties": False, + "$defs": { + "Status": {"enum": ["active", "inactive"], "type": "string"}, + "UnusedType": {"type": "number"}, # Will be pruned + }, + "properties": { + "name": {"type": "string", "title": "Name Field"}, + "status": {"$ref": "#/$defs/Status", "title": "Status Field"}, + "remove_me": {"type": "string"}, # Will be pruned + }, + "required": ["name", "status", "remove_me"], + } + + result = compress_schema( + schema, + prune_params=["remove_me"], + prune_defs=True, + prune_additional_properties=True, + prune_titles=True, + dereference_refs=True, + ) + + expected = { + "type": "object", + "properties": { + "name": {"type": "string"}, # Title removed + "status": { + "enum": ["active", "inactive"], + "type": "string", + }, # Flattened and title removed + }, + "required": ["name", "status"], + } + assert result == expected + assert "title" not in result + assert "additionalProperties" not in result + assert "$defs" not in result + assert "remove_me" not in result["properties"]