Skip to content
29 changes: 27 additions & 2 deletions python/packages/core/agent_framework/_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -436,6 +436,28 @@ def _normalize_tool_description_format( # pyright: ignore[reportUnusedFunction]
return normalized


def _canonicalize_schema(schema: Any) -> Any:
"""Recursively sort schema dictionary keys deterministically to preserve prompt cache stability.

Sorts all object-level keyword keys alphabetically, and recursively canonicalizes nested
schemas in mappings (properties, patternProperties, $defs, definitions, dependentSchemas)
and sequences (items, allOf, anyOf, oneOf). Non-dict/list values are returned unmodified.
"""
if isinstance(schema, dict):
schema_dict = cast(dict[str, Any], schema)
result: dict[str, Any] = {}
for key, val in sorted(schema_dict.items(), key=lambda pair: pair[0]):
if isinstance(val, (dict, list)):
result[key] = _canonicalize_schema(val)
else:
result[key] = val
return result
if isinstance(schema, list):
schema_list = cast(list[Any], schema)
return [_canonicalize_schema(item) for item in schema_list]
return schema


def _format_tool_parameters( # pyright: ignore[reportUnusedFunction]
parameters: dict[str, Any],
*,
Expand Down Expand Up @@ -1235,11 +1257,14 @@ def parameters(self) -> dict[str, Any]:
"""Create the JSON schema of the parameters.

Returns:
A dictionary containing the JSON schema for the function's parameters.
A dictionary containing the JSON schema for the function's parameters,
with deterministic recursive sorting to preserve prompt cache stability.
The result is cached after the first call for performance.
"""
if self._cached_parameters is None:
self._cached_parameters = self._input_schema
raw_schema = self._input_schema
canonical = _canonicalize_schema(raw_schema) if isinstance(raw_schema, dict) else raw_schema
self._cached_parameters = canonical if isinstance(canonical, dict) else {}
return self._cached_parameters

@staticmethod
Expand Down
149 changes: 149 additions & 0 deletions python/packages/core/tests/core/test_tools.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import copy
import json
import logging
import threading
from contextvars import ContextVar
Expand Down Expand Up @@ -2029,4 +2030,152 @@ def test_skip_parsing_is_singleton() -> None:
assert repr(SKIP_PARSING) == "SKIP_PARSING"


def test_to_json_schema_spec_canonicalizes_reordered_top_level_keywords() -> None:
"""Verify that to_json_schema_spec canonicalizes top-level and nested keyword order for prompt caching."""
# Two tools with differently ordered top-level keywords and properties
schema_1 = {
"title": "SearchTool",
"type": "object",
"description": "Searches items",
"properties": {
"z_query": {"description": "Query string", "type": "string"},
"a_limit": {"default": 10, "type": "integer"},
},
"required": ["z_query"],
}
schema_2 = {
"description": "Searches items",
"properties": {
"a_limit": {"type": "integer", "default": 10},
"z_query": {"type": "string", "description": "Query string"},
},
"type": "object",
"required": ["z_query"],
"title": "SearchTool",
}

tool_1 = FunctionTool(name="search", description="Search tool", input_model=schema_1)
tool_2 = FunctionTool(name="search", description="Search tool", input_model=schema_2)

spec_1 = tool_1.to_json_schema_spec()
spec_2 = tool_2.to_json_schema_spec()

# Top-level parameters dictionary keys must be sorted deterministically
assert list(spec_1["function"]["parameters"].keys()) == ["description", "properties", "required", "title", "type"]
assert list(spec_2["function"]["parameters"].keys()) == ["description", "properties", "required", "title", "type"]

# Properties dictionary keys must be sorted deterministically
assert list(spec_1["function"]["parameters"]["properties"].keys()) == ["a_limit", "z_query"]
assert list(spec_2["function"]["parameters"]["properties"].keys()) == ["a_limit", "z_query"]

# Wire-level serialization must be identical
wire_1 = json.dumps(spec_1)
wire_2 = json.dumps(spec_2)
assert wire_1 == wire_2


def test_to_json_schema_spec_canonicalizes_root_schema_without_properties() -> None:
"""Regression: root schemas without 'properties' are canonicalized deterministically."""
schema_1 = {
"description": "A scalar or empty tool schema",
"type": "string",
"title": "ScalarInput",
}
schema_2 = {
"title": "ScalarInput",
"description": "A scalar or empty tool schema",
"type": "string",
}

tool_1 = FunctionTool(name="scalar_tool", description="Scalar tool", input_model=schema_1)
tool_2 = FunctionTool(name="scalar_tool", description="Scalar tool", input_model=schema_2)

spec_1 = tool_1.to_json_schema_spec()
spec_2 = tool_2.to_json_schema_spec()

assert list(spec_1["function"]["parameters"].keys()) == ["description", "title", "type"]
assert list(spec_2["function"]["parameters"].keys()) == ["description", "title", "type"]
assert json.dumps(spec_1) == json.dumps(spec_2)


def test_to_json_schema_spec_canonicalizes_nested_and_defs_deterministically() -> None:
"""Verify that nested objects, $defs, and arrays are recursively sorted deterministically."""
schema = {
"type": "object",
"$defs": {
"ZItem": {"type": "string"},
"AItem": {"type": "integer"},
},
"properties": {
"metadata": {
"type": "object",
"properties": {
"tag_z": {"type": "string"},
"tag_a": {"type": "string"},
},
},
},
}
tool = FunctionTool(name="complex_tool", description="Complex tool", input_model=schema)
spec = tool.to_json_schema_spec()
params = spec["function"]["parameters"]

assert list(params.keys()) == ["$defs", "properties", "type"]
assert list(params["$defs"].keys()) == ["AItem", "ZItem"]
assert list(params["properties"]["metadata"]["properties"].keys()) == ["tag_a", "tag_z"]


def test_parameters_canonicalizes_schema_deterministically() -> None:
"""Verify that FunctionTool.parameters() directly canonicalizes schemas deterministically.

This ensures providers that serialize tool.parameters() directly (such as Anthropic,
Bedrock, Gemini, and Mistral) obtain deterministic wire-level outputs without cache churn.
"""
schema_1 = {
"title": "Config",
"description": "Provider tool",
"type": "object",
"properties": {
"z_param": {"type": "string"},
"a_param": {"type": "integer"},
},
}
schema_2 = {
"properties": {
"a_param": {"type": "integer"},
"z_param": {"type": "string"},
},
"type": "object",
"description": "Provider tool",
"title": "Config",
}

tool_1 = FunctionTool(name="config_tool", description="tool", input_model=schema_1)
tool_2 = FunctionTool(name="config_tool", description="tool", input_model=schema_2)

params_1 = tool_1.parameters()
params_2 = tool_2.parameters()

# Direct parameters() must be sorted deterministically
assert list(params_1.keys()) == ["description", "properties", "title", "type"]
assert list(params_2.keys()) == ["description", "properties", "title", "type"]
assert list(params_1["properties"].keys()) == ["a_param", "z_param"]
assert list(params_2["properties"].keys()) == ["a_param", "z_param"]

# Provider wire simulation: Anthropic tool definition shape
anthropic_wire_1 = json.dumps({
"type": "custom",
"name": tool_1.name,
"description": tool_1.description,
"input_schema": tool_1.parameters(),
})
anthropic_wire_2 = json.dumps({
"type": "custom",
"name": tool_2.name,
"description": tool_2.description,
"input_schema": tool_2.parameters(),
})
assert anthropic_wire_1 == anthropic_wire_2


# endregion
Loading