mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-23 14:04:18 +02:00
Dereference $ref in tool schemas for MCP client compatibility (#2808)
This commit is contained in:
parent
a117316ba1
commit
627c6cdad4
9 changed files with 192 additions and 313 deletions
|
|
@ -20,6 +20,7 @@ dependencies = [
|
|||
"uvicorn>=0.35",
|
||||
"websockets>=15.0.1",
|
||||
"jsonschema-path>=0.3.4",
|
||||
"jsonref>=1.1.0",
|
||||
]
|
||||
|
||||
requires-python = ">=3.10"
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ import fastmcp
|
|||
from fastmcp.server.dependencies import without_injected_parameters
|
||||
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
|
||||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.json_schema import compress_schema, resolve_root_ref
|
||||
from fastmcp.utilities.json_schema import compress_schema
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.types import (
|
||||
Audio,
|
||||
|
|
@ -686,10 +686,6 @@ class ParsedFunction:
|
|||
|
||||
output_schema = compress_schema(output_schema, prune_titles=True)
|
||||
|
||||
# Resolve root-level $ref to meet MCP spec requirement for type: object
|
||||
# Self-referential Pydantic models generate schemas with $ref at root
|
||||
output_schema = resolve_root_ref(output_schema)
|
||||
|
||||
except PydanticSchemaGenerationError as e:
|
||||
if "_UnserializableType" not in str(e):
|
||||
logger.debug(f"Unable to generate schema for type {output_type!r}")
|
||||
|
|
|
|||
|
|
@ -680,7 +680,7 @@ class TransformedTool(Tool):
|
|||
|
||||
if parent_defs:
|
||||
schema["$defs"] = parent_defs
|
||||
schema = compress_schema(schema, prune_defs=True)
|
||||
schema = compress_schema(schema)
|
||||
|
||||
# Create forwarding function that closes over everything it needs
|
||||
async def _forward(**kwargs: Any):
|
||||
|
|
@ -863,7 +863,7 @@ class TransformedTool(Tool):
|
|||
|
||||
if merged_defs:
|
||||
result["$defs"] = merged_defs
|
||||
result = compress_schema(result, prune_defs=True)
|
||||
result = compress_schema(result)
|
||||
|
||||
return result
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,52 @@ from __future__ import annotations
|
|||
from collections import defaultdict
|
||||
from typing import Any
|
||||
|
||||
from jsonref import JsonRefError, replace_refs
|
||||
|
||||
|
||||
def dereference_refs(schema: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Resolve all $ref references in a JSON schema by inlining definitions.
|
||||
|
||||
This function resolves $ref references that point to $defs, replacing them
|
||||
with the actual definition content. This is necessary because some MCP clients
|
||||
(e.g., VS Code Copilot) don't properly handle $ref in tool input schemas.
|
||||
|
||||
For self-referencing/circular schemas where full dereferencing is not possible,
|
||||
this function falls back to resolving only the root-level $ref while preserving
|
||||
$defs for nested references.
|
||||
|
||||
Args:
|
||||
schema: JSON schema dict that may contain $ref references
|
||||
|
||||
Returns:
|
||||
A new schema dict with $ref resolved where possible and $defs removed
|
||||
when no longer needed
|
||||
|
||||
Example:
|
||||
>>> schema = {
|
||||
... "$defs": {"Category": {"enum": ["a", "b"], "type": "string"}},
|
||||
... "properties": {"cat": {"$ref": "#/$defs/Category"}}
|
||||
... }
|
||||
>>> resolved = dereference_refs(schema)
|
||||
>>> # Result: {"properties": {"cat": {"enum": ["a", "b"], "type": "string"}}}
|
||||
"""
|
||||
try:
|
||||
# Use jsonref to resolve all $ref references
|
||||
# proxies=False returns plain dicts (not proxy objects)
|
||||
# lazy_load=False resolves immediately
|
||||
dereferenced = replace_refs(schema, proxies=False, lazy_load=False)
|
||||
|
||||
# Remove $defs since all references have been resolved
|
||||
if isinstance(dereferenced, dict) and "$defs" in dereferenced:
|
||||
dereferenced = {k: v for k, v in dereferenced.items() if k != "$defs"}
|
||||
|
||||
return dereferenced
|
||||
|
||||
except JsonRefError:
|
||||
# Self-referencing/circular schemas can't be fully dereferenced
|
||||
# Fall back to resolving only root-level $ref (for MCP spec compliance)
|
||||
return resolve_root_ref(schema)
|
||||
|
||||
|
||||
def resolve_root_ref(schema: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Resolve $ref at root level to meet MCP spec requirements.
|
||||
|
|
@ -240,31 +286,38 @@ def _single_pass_optimize(
|
|||
def compress_schema(
|
||||
schema: dict,
|
||||
prune_params: list[str] | None = None,
|
||||
prune_defs: bool = True,
|
||||
prune_additional_properties: bool = True,
|
||||
prune_titles: bool = False,
|
||||
) -> dict:
|
||||
"""
|
||||
Remove the given parameters from the schema.
|
||||
Compress and optimize a JSON schema for MCP compatibility.
|
||||
|
||||
This function dereferences all $ref entries (inlining definitions) to ensure
|
||||
compatibility with MCP clients that don't properly handle $ref in schemas
|
||||
(e.g., VS Code Copilot). It also applies various optimizations to reduce
|
||||
schema size.
|
||||
|
||||
Args:
|
||||
schema: The schema to compress
|
||||
prune_params: List of parameter names to remove from properties
|
||||
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 $ref - this inlines all definitions and removes $defs
|
||||
# Required for MCP client compatibility
|
||||
schema = dereference_refs(schema)
|
||||
|
||||
# Remove specific parameters if requested
|
||||
for param in prune_params or []:
|
||||
schema = _prune_param(schema, param=param)
|
||||
|
||||
# Apply combined optimizations in a single tree traversal
|
||||
if prune_titles or prune_additional_properties or prune_defs:
|
||||
if prune_titles or prune_additional_properties:
|
||||
schema = _single_pass_optimize(
|
||||
schema,
|
||||
prune_titles=prune_titles,
|
||||
prune_additional_properties=prune_additional_properties,
|
||||
prune_defs=prune_defs,
|
||||
prune_defs=False,
|
||||
)
|
||||
|
||||
return schema
|
||||
|
|
|
|||
|
|
@ -201,18 +201,15 @@ class TestToolFromFunction:
|
|||
"description": "Create a new user.",
|
||||
"tags": set(),
|
||||
"parameters": {
|
||||
"$defs": {
|
||||
"UserInput": {
|
||||
"properties": {
|
||||
"user": {
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"age": {"type": "integer"},
|
||||
},
|
||||
"required": ["name", "age"],
|
||||
"type": "object",
|
||||
}
|
||||
},
|
||||
"properties": {
|
||||
"user": {"$ref": "#/$defs/UserInput"},
|
||||
},
|
||||
"flag": {"type": "boolean"},
|
||||
},
|
||||
"required": ["user", "flag"],
|
||||
|
|
|
|||
|
|
@ -218,10 +218,14 @@ async def test_hidden_param_prunes_defs():
|
|||
schema = new_tool.parameters
|
||||
# Only 'a' should be visible
|
||||
assert list(schema["properties"].keys()) == ["a"]
|
||||
# $defs should only contain VisibleType, not HiddenType
|
||||
defs = schema.get("$defs", {})
|
||||
assert "VisibleType" in defs
|
||||
assert "HiddenType" not in defs
|
||||
# Schema should be fully dereferenced (no $defs)
|
||||
assert "$defs" not in schema
|
||||
# VisibleType should be inlined in the property
|
||||
assert schema["properties"]["a"] == {
|
||||
"properties": {"x": {"type": "integer"}},
|
||||
"required": ["x"],
|
||||
"type": "object",
|
||||
}
|
||||
|
||||
|
||||
async def test_forward_with_argument_mapping(add_tool):
|
||||
|
|
@ -438,7 +442,10 @@ def test_transform_args_with_parent_defaults():
|
|||
|
||||
new_tool = Tool.from_tool(tool)
|
||||
|
||||
assert new_tool.parameters["$defs"] == tool.parameters["$defs"]
|
||||
# Both tools should have the same dereferenced schema
|
||||
assert new_tool.parameters == tool.parameters
|
||||
# Schema should be fully dereferenced (no $defs)
|
||||
assert "$defs" not in new_tool.parameters
|
||||
|
||||
|
||||
def test_transform_args_validation_unknown_arg(add_tool):
|
||||
|
|
@ -1544,7 +1551,11 @@ class TestInputSchema:
|
|||
assert "examples" not in prop3
|
||||
|
||||
def test_merge_schema_with_defs_precedence(self):
|
||||
"""Test _merge_schema_with_precedence merges $defs correctly."""
|
||||
"""Test _merge_schema_with_precedence merges $defs correctly.
|
||||
|
||||
Note: This tests the raw merge behavior before dereferencing.
|
||||
The final schema output will be dereferenced by compress_schema.
|
||||
"""
|
||||
base_schema = {
|
||||
"type": "object",
|
||||
"properties": {"field1": {"$ref": "#/$defs/BaseType"}},
|
||||
|
|
@ -1567,26 +1578,27 @@ class TestInputSchema:
|
|||
base_schema, override_schema
|
||||
)
|
||||
|
||||
# SharedType should no longer be present on the schema
|
||||
assert "SharedType" not in transformed_tool_schema["$defs"]
|
||||
# SharedType should no longer be present on the schema (unused)
|
||||
assert "SharedType" not in transformed_tool_schema.get("$defs", {})
|
||||
|
||||
# Schema is dereferenced so no $defs in final output
|
||||
assert transformed_tool_schema == snapshot(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"field1": {"$ref": "#/$defs/BaseType"},
|
||||
"field2": {"$ref": "#/$defs/OverrideType"},
|
||||
"field1": {"type": "string", "description": "base"},
|
||||
"field2": {"type": "boolean"},
|
||||
},
|
||||
"required": [],
|
||||
"$defs": {
|
||||
"BaseType": {"type": "string", "description": "base"},
|
||||
"OverrideType": {"type": "boolean"},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
def test_transform_tool_with_complex_defs_pruning(self):
|
||||
"""Test that tool transformation properly prunes unused $defs."""
|
||||
"""Test that tool transformation properly handles hidden params.
|
||||
|
||||
With schema dereferencing, unused types are automatically removed
|
||||
since $defs is eliminated entirely.
|
||||
"""
|
||||
|
||||
class UsedType(BaseModel):
|
||||
value: str
|
||||
|
|
@ -1605,25 +1617,25 @@ class TestInputSchema:
|
|||
complex_tool, transform_args={"unused_param": ArgTransform(hide=True)}
|
||||
)
|
||||
|
||||
assert "UnusedType" not in transformed_tool.parameters["$defs"]
|
||||
# Schema is dereferenced - no $defs
|
||||
assert "$defs" not in transformed_tool.parameters
|
||||
|
||||
assert transformed_tool.parameters == snapshot(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"used_param": {"$ref": "#/$defs/UsedType"}},
|
||||
"required": ["used_param"],
|
||||
"$defs": {
|
||||
"UsedType": {
|
||||
"properties": {
|
||||
"used_param": {
|
||||
"properties": {"value": {"type": "string"}},
|
||||
"required": ["value"],
|
||||
"type": "object",
|
||||
}
|
||||
},
|
||||
"required": ["used_param"],
|
||||
}
|
||||
)
|
||||
|
||||
def test_transform_with_custom_function_preserves_needed_defs(self):
|
||||
"""Test that custom transform functions preserve necessary $defs."""
|
||||
def test_transform_with_custom_function_preserves_needed_types(self):
|
||||
"""Test that custom transform functions preserve necessary types inline."""
|
||||
|
||||
class InputType(BaseModel):
|
||||
data: str
|
||||
|
|
@ -1645,23 +1657,25 @@ class TestInputSchema:
|
|||
transform_args={"input_data": ArgTransform(name="renamed_input")},
|
||||
)
|
||||
|
||||
# Schema is dereferenced - types are inlined
|
||||
assert "$defs" not in transformed.parameters
|
||||
|
||||
assert transformed.parameters == snapshot(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"renamed_input": {"$ref": "#/$defs/InputType"}},
|
||||
"required": ["renamed_input"],
|
||||
"$defs": {
|
||||
"InputType": {
|
||||
"properties": {
|
||||
"renamed_input": {
|
||||
"properties": {"data": {"type": "string"}},
|
||||
"required": ["data"],
|
||||
"type": "object",
|
||||
}
|
||||
},
|
||||
"required": ["renamed_input"],
|
||||
}
|
||||
)
|
||||
|
||||
def test_chained_transforms_preserve_correct_defs(self):
|
||||
"""Test that chained transformations preserve correct $defs."""
|
||||
def test_chained_transforms_inline_types(self):
|
||||
"""Test that chained transformations produce correct inlined schemas."""
|
||||
|
||||
class TypeA(BaseModel):
|
||||
a: str
|
||||
|
|
@ -1682,50 +1696,46 @@ class TestInputSchema:
|
|||
transform_args={"param_c": ArgTransform(hide=True, default=TypeC(c=True))},
|
||||
)
|
||||
|
||||
# Schema is dereferenced - types are inlined
|
||||
assert "$defs" not in transform1.parameters
|
||||
|
||||
assert transform1.parameters == snapshot(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"param_a": {"$ref": "#/$defs/TypeA"},
|
||||
"param_b": {"$ref": "#/$defs/TypeB"},
|
||||
},
|
||||
"required": IsList("param_b", "param_a", check_order=False),
|
||||
"$defs": {
|
||||
"TypeA": {
|
||||
"param_a": {
|
||||
"properties": {"a": {"type": "string"}},
|
||||
"required": ["a"],
|
||||
"type": "object",
|
||||
},
|
||||
"TypeB": {
|
||||
"param_b": {
|
||||
"properties": {"b": {"type": "integer"}},
|
||||
"required": ["b"],
|
||||
"type": "object",
|
||||
},
|
||||
},
|
||||
"required": IsList("param_b", "param_a", check_order=False),
|
||||
}
|
||||
)
|
||||
|
||||
assert "TypeA" in transform1.parameters["$defs"]
|
||||
|
||||
# Second transform: hide param_b
|
||||
transform2 = Tool.from_tool(
|
||||
transform1,
|
||||
transform_args={"param_b": ArgTransform(hide=True, default=TypeB(b=42))},
|
||||
)
|
||||
|
||||
assert "TypeB" not in transform2.parameters["$defs"]
|
||||
assert "$defs" not in transform2.parameters
|
||||
|
||||
assert transform2.parameters == snapshot(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"param_a": {"$ref": "#/$defs/TypeA"}},
|
||||
"required": ["param_a"],
|
||||
"$defs": {
|
||||
"TypeA": {
|
||||
"properties": {
|
||||
"param_a": {
|
||||
"properties": {"a": {"type": "string"}},
|
||||
"required": ["a"],
|
||||
"type": "object",
|
||||
}
|
||||
},
|
||||
"required": ["param_a"],
|
||||
}
|
||||
)
|
||||
|
|
|
|||
|
|
@ -580,8 +580,8 @@ class TestEdgeCases:
|
|||
len(properties) > 0
|
||||
) # Should have some properties from one of the content types
|
||||
|
||||
def test_oneof_reference_preserved(self):
|
||||
"""Test that schemas referenced in oneOf are preserved."""
|
||||
def test_oneof_reference_dereferenced(self):
|
||||
"""Test that schemas referenced in oneOf are dereferenced."""
|
||||
|
||||
schema = {
|
||||
"type": "object",
|
||||
|
|
@ -594,14 +594,14 @@ class TestEdgeCases:
|
|||
|
||||
result = compress_schema(schema)
|
||||
|
||||
# TestSchema should be preserved (referenced in oneOf)
|
||||
assert "TestSchema" in result["$defs"]
|
||||
# $defs should be removed (all refs dereferenced)
|
||||
assert "$defs" not in result
|
||||
|
||||
# UnusedSchema should be removed
|
||||
assert "UnusedSchema" not in result["$defs"]
|
||||
# TestSchema should be inlined in oneOf
|
||||
assert result["properties"]["data"]["oneOf"] == [{"type": "string"}]
|
||||
|
||||
def test_anyof_reference_preserved(self):
|
||||
"""Test that schemas referenced in anyOf are preserved."""
|
||||
def test_anyof_reference_dereferenced(self):
|
||||
"""Test that schemas referenced in anyOf are dereferenced."""
|
||||
|
||||
schema = {
|
||||
"type": "object",
|
||||
|
|
@ -614,11 +614,14 @@ class TestEdgeCases:
|
|||
|
||||
result = compress_schema(schema)
|
||||
|
||||
assert "TestSchema" in result["$defs"]
|
||||
assert "UnusedSchema" not in result["$defs"]
|
||||
# $defs should be removed (all refs dereferenced)
|
||||
assert "$defs" not in result
|
||||
|
||||
def test_allof_reference_preserved(self):
|
||||
"""Test that schemas referenced in allOf are preserved."""
|
||||
# TestSchema should be inlined in anyOf
|
||||
assert result["properties"]["data"]["anyOf"] == [{"type": "string"}]
|
||||
|
||||
def test_allof_reference_dereferenced(self):
|
||||
"""Test that schemas referenced in allOf are dereferenced."""
|
||||
|
||||
schema = {
|
||||
"type": "object",
|
||||
|
|
@ -631,5 +634,8 @@ class TestEdgeCases:
|
|||
|
||||
result = compress_schema(schema)
|
||||
|
||||
assert "TestSchema" in result["$defs"]
|
||||
assert "UnusedSchema" not in result["$defs"]
|
||||
# $defs should be removed (all refs dereferenced)
|
||||
assert "$defs" not in result
|
||||
|
||||
# TestSchema should be inlined in allOf
|
||||
assert result["properties"]["data"]["allOf"] == [{"type": "string"}]
|
||||
|
|
|
|||
|
|
@ -1,18 +1,10 @@
|
|||
from fastmcp.utilities.json_schema import (
|
||||
_prune_param,
|
||||
compress_schema,
|
||||
dereference_refs,
|
||||
resolve_root_ref,
|
||||
)
|
||||
|
||||
# Wrapper for backward compatibility with tests
|
||||
|
||||
|
||||
def _prune_additional_properties(schema):
|
||||
"""Wrapper for compress_schema that only prunes additionalProperties: false."""
|
||||
return compress_schema(
|
||||
schema, prune_defs=False, prune_additional_properties=True, prune_titles=False
|
||||
)
|
||||
|
||||
|
||||
class TestPruneParam:
|
||||
"""Tests for the _prune_param function."""
|
||||
|
|
@ -55,32 +47,28 @@ class TestPruneParam:
|
|||
assert "required" not in result
|
||||
|
||||
|
||||
class TestPruneUnusedDefs:
|
||||
"""Tests for unused definition pruning (via compress_schema)."""
|
||||
class TestDereferenceRefs:
|
||||
"""Tests for the dereference_refs function."""
|
||||
|
||||
def test_removes_unreferenced_defs(self):
|
||||
"""Test that unreferenced definitions are removed."""
|
||||
def test_dereferences_simple_ref(self):
|
||||
"""Test that simple $ref is dereferenced."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
},
|
||||
"$defs": {
|
||||
"foo_def": {"type": "string"},
|
||||
"unused_def": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
result = dereference_refs(schema)
|
||||
|
||||
assert "foo_def" in result["$defs"]
|
||||
assert "unused_def" not in result["$defs"]
|
||||
# $ref should be inlined
|
||||
assert result["properties"]["foo"] == {"type": "string"}
|
||||
# $defs should be removed
|
||||
assert "$defs" not in result
|
||||
|
||||
def test_nested_references_kept(self):
|
||||
"""Test that definitions referenced via nesting are kept."""
|
||||
def test_dereferences_nested_refs(self):
|
||||
"""Test that nested $refs are dereferenced."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
|
|
@ -91,210 +79,59 @@ class TestPruneUnusedDefs:
|
|||
"properties": {"nested": {"$ref": "#/$defs/nested_def"}},
|
||||
},
|
||||
"nested_def": {"type": "string"},
|
||||
"unused_def": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "foo_def" in result["$defs"]
|
||||
assert "nested_def" in result["$defs"]
|
||||
assert "unused_def" not in result["$defs"]
|
||||
result = dereference_refs(schema)
|
||||
|
||||
def test_nested_references_removed(self):
|
||||
"""Test that definitions referenced via nesting in unused defs are removed."""
|
||||
schema = {
|
||||
"properties": {},
|
||||
"$defs": {
|
||||
"foo_def": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/nested_def"}},
|
||||
},
|
||||
"nested_def": {"type": "string"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
# All refs should be inlined
|
||||
assert result["properties"]["foo"]["properties"]["nested"] == {"type": "string"}
|
||||
# $defs should be removed
|
||||
assert "$defs" not in result
|
||||
|
||||
def test_nested_references_with_recursion_kept(self):
|
||||
"""Test that definitions with recursion referenced via nesting are kept."""
|
||||
def test_falls_back_for_circular_refs(self):
|
||||
"""Test that circular references fall back to resolve_root_ref."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
},
|
||||
"$defs": {
|
||||
"foo_def": {
|
||||
"Node": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/foo_def"}},
|
||||
},
|
||||
"unused_def": {"type": "integer"},
|
||||
"properties": {
|
||||
"children": {
|
||||
"type": "array",
|
||||
"items": {"$ref": "#/$defs/Node"},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
"$ref": "#/$defs/Node",
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "foo_def" in result["$defs"]
|
||||
assert "unused_def" not in result["$defs"]
|
||||
result = dereference_refs(schema)
|
||||
|
||||
def test_nested_references_with_recursion_removed(self):
|
||||
"""Test that definitions with recursion referenced via nesting in unused defs are removed."""
|
||||
schema = {
|
||||
"properties": {},
|
||||
"$defs": {
|
||||
"foo_def": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/foo_def"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "$defs" not in result
|
||||
|
||||
def test_multiple_nested_references_with_recursion_kept(self):
|
||||
"""Test that definitions with multiple levels of recursion referenced via nesting are kept."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
},
|
||||
"$defs": {
|
||||
"foo_def": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/nested_def"}},
|
||||
},
|
||||
"nested_def": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/foo_def"}},
|
||||
},
|
||||
"unused_def": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "foo_def" in result["$defs"]
|
||||
assert "nested_def" in result["$defs"]
|
||||
assert "unused_def" not in result["$defs"]
|
||||
|
||||
def test_multiple_nested_references_with_recursion_removed(self):
|
||||
"""Test that definitions with multiple levels of recursion referenced via nesting in unused defs are removed."""
|
||||
schema = {
|
||||
"properties": {},
|
||||
"$defs": {
|
||||
"foo_def": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/nested_def"}},
|
||||
},
|
||||
"nested_def": {
|
||||
"type": "object",
|
||||
"properties": {"nested": {"$ref": "#/$defs/foo_def"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "$defs" not in result
|
||||
|
||||
def test_array_references_kept(self):
|
||||
"""Test that definitions referenced in array items are kept."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"items": {"type": "array", "items": {"$ref": "#/$defs/item_def"}},
|
||||
},
|
||||
"$defs": {
|
||||
"item_def": {"type": "string"},
|
||||
"unused_def": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "item_def" in result["$defs"]
|
||||
assert "unused_def" not in result["$defs"]
|
||||
|
||||
def test_removes_defs_field_when_empty(self):
|
||||
"""Test that $defs field is removed when all definitions are unused."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"type": "string"},
|
||||
},
|
||||
"$defs": {
|
||||
"unused_def": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(
|
||||
schema,
|
||||
prune_defs=True,
|
||||
prune_additional_properties=False,
|
||||
prune_titles=False,
|
||||
)
|
||||
assert "$defs" not in result
|
||||
|
||||
|
||||
class TestPruneAdditionalProperties:
|
||||
"""Tests for the _prune_additional_properties function."""
|
||||
|
||||
def test_removes_when_false(self):
|
||||
"""Test that additionalProperties is removed when it's false."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"foo": {"type": "string"}},
|
||||
"additionalProperties": False,
|
||||
}
|
||||
result = _prune_additional_properties(schema)
|
||||
assert "additionalProperties" not in result
|
||||
|
||||
def test_keeps_when_true(self):
|
||||
"""Test that additionalProperties is kept when it's true."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"foo": {"type": "string"}},
|
||||
"additionalProperties": True,
|
||||
}
|
||||
result = _prune_additional_properties(schema)
|
||||
assert "additionalProperties" in result
|
||||
assert result["additionalProperties"] is True
|
||||
|
||||
def test_keeps_when_object(self):
|
||||
"""Test that additionalProperties is kept when it's an object schema."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"foo": {"type": "string"}},
|
||||
"additionalProperties": {"type": "string"},
|
||||
}
|
||||
result = _prune_additional_properties(schema)
|
||||
assert "additionalProperties" in result
|
||||
assert result["additionalProperties"] == {"type": "string"}
|
||||
# Should fall back to resolve_root_ref behavior
|
||||
# Root should be resolved but nested refs preserved
|
||||
assert result.get("type") == "object"
|
||||
assert "$defs" in result # $defs preserved for circular refs
|
||||
|
||||
|
||||
class TestCompressSchema:
|
||||
"""Tests for the compress_schema function."""
|
||||
|
||||
def test_dereferences_by_default(self):
|
||||
"""Test that compress_schema dereferences $refs by default."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
},
|
||||
"$defs": {
|
||||
"foo_def": {"type": "string"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(schema)
|
||||
|
||||
# $ref should be inlined
|
||||
assert result["properties"]["foo"] == {"type": "string"}
|
||||
# $defs should be removed
|
||||
assert "$defs" not in result
|
||||
|
||||
def test_prune_params(self):
|
||||
"""Test pruning parameters with compress_schema."""
|
||||
schema = {
|
||||
|
|
@ -309,38 +146,6 @@ class TestCompressSchema:
|
|||
assert result["properties"] == {"bar": {"type": "integer"}}
|
||||
assert result["required"] == ["bar"]
|
||||
|
||||
def test_prune_defs(self):
|
||||
"""Test pruning unused definitions with compress_schema."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
"bar": {"type": "integer"},
|
||||
},
|
||||
"$defs": {
|
||||
"foo_def": {"type": "string"},
|
||||
"unused_def": {"type": "number"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(schema)
|
||||
assert "foo_def" in result["$defs"]
|
||||
assert "unused_def" not in result["$defs"]
|
||||
|
||||
def test_disable_prune_defs(self):
|
||||
"""Test disabling pruning of unused definitions."""
|
||||
schema = {
|
||||
"properties": {
|
||||
"foo": {"$ref": "#/$defs/foo_def"},
|
||||
"bar": {"type": "integer"},
|
||||
},
|
||||
"$defs": {
|
||||
"foo_def": {"type": "string"},
|
||||
"unused_def": {"type": "number"},
|
||||
},
|
||||
}
|
||||
result = compress_schema(schema, prune_defs=False)
|
||||
assert "foo_def" in result["$defs"]
|
||||
assert "unused_def" in result["$defs"]
|
||||
|
||||
def test_pruning_additional_properties(self):
|
||||
"""Test pruning additionalProperties when False."""
|
||||
schema = {
|
||||
|
|
@ -382,8 +187,8 @@ class TestCompressSchema:
|
|||
assert "remove" not in result["properties"]
|
||||
# Check that required list was updated
|
||||
assert result["required"] == ["keep"]
|
||||
# Check that unused definitions were removed
|
||||
assert "$defs" not in result # Both defs should be gone
|
||||
# Check that $defs was removed (dereferenced)
|
||||
assert "$defs" not in result
|
||||
# Check that additionalProperties was removed
|
||||
assert "additionalProperties" not in result
|
||||
|
||||
|
|
|
|||
11
uv.lock
generated
11
uv.lock
generated
|
|
@ -689,6 +689,7 @@ dependencies = [
|
|||
{ name = "cyclopts" },
|
||||
{ name = "exceptiongroup" },
|
||||
{ name = "httpx" },
|
||||
{ name = "jsonref" },
|
||||
{ name = "jsonschema-path" },
|
||||
{ name = "mcp" },
|
||||
{ name = "openapi-pydantic" },
|
||||
|
|
@ -745,6 +746,7 @@ requires-dist = [
|
|||
{ name = "cyclopts", specifier = ">=4.0.0" },
|
||||
{ name = "exceptiongroup", specifier = ">=1.2.2" },
|
||||
{ name = "httpx", specifier = ">=0.28.1" },
|
||||
{ name = "jsonref", specifier = ">=1.1.0" },
|
||||
{ name = "jsonschema-path", specifier = ">=0.3.4" },
|
||||
{ name = "mcp", specifier = ">=1.24.0,<2.0" },
|
||||
{ name = "openai", marker = "extra == 'openai'", specifier = ">=1.102.0" },
|
||||
|
|
@ -1100,6 +1102,15 @@ wheels = [
|
|||
{ url = "https://files.pythonhosted.org/packages/2f/9c/6753e6522b8d0ef07d3a3d239426669e984fb0eba15a315cdbc1253904e4/jiter-0.12.0-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c24e864cb30ab82311c6425655b0cdab0a98c5d973b065c66a3f020740c2324c", size = 346110, upload-time = "2025-11-09T20:49:21.817Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jsonref"
|
||||
version = "1.1.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/aa/0d/c1f3277e90ccdb50d33ed5ba1ec5b3f0a242ed8c1b1a85d3afeb68464dca/jsonref-1.1.0.tar.gz", hash = "sha256:32fe8e1d85af0fdefbebce950af85590b22b60f9e95443176adbde4e1ecea552", size = 8814, upload-time = "2023-01-16T16:10:04.455Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/0c/ec/e1db9922bceb168197a558a2b8c03a7963f1afe93517ddd3cf99f202f996/jsonref-1.1.0-py3-none-any.whl", hash = "sha256:590dc7773df6c21cbf948b5dac07a72a251db28b0238ceecce0a2abfa8ec30a9", size = 9425, upload-time = "2023-01-16T16:10:02.255Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jsonschema"
|
||||
version = "4.25.1"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue