Setting the file. One moment.
Compare Tool Schemas · Eval Engineering · langchain-ai/langchain-skills · Skills Docs
Back to the top of the page
Type
Python scripts/ compare_tool_schemas.py
Python · 391 lines · 14 KB
from
typing
import
Any
15
16
17 class SchemaError ( ValueError ):
18 """Raised when a schema file is malformed."""
19
20
21 def load_json (path: Path) -> Any:
22 try :
23 return json.loads(path.read_text( encoding = "utf-8" ))
24 except ( OSError , json.JSONDecodeError) as error:
25 raise SchemaError( f "cannot read { path } : { error } " ) from error
26
27
28 def tool_list (value: Any) -> list[dict[ str , Any]]:
29 if isinstance (value, dict ) and "tools" in value:
30 value = value[ "tools" ]
31 if not isinstance (value, list ) or not all ( isinstance (item, dict ) for item in value):
32 raise SchemaError( "tool file must be a list or an object with a tools list" )
33 return value
34
35
36 def normalize_tool (value: dict[ str , Any]) -> dict[ str , Any]:
37 function = value.get( "function" ) if isinstance (value.get( "function" ), dict ) else {}
38 name = value.get( "name" ) or function.get( "name" )
39 if not isinstance (name, str ) or not name:
40 raise SchemaError( "every tool must have a name" )
41
42 def present ( * choices: tuple[dict[ str , Any], str ], default: Any = None ) -> Any:
43 for source, key in choices:
44 if key in source:
45 return source[key]
46 return default
47
48 input_schema = present(
49 (value, "input_schema" ),
50 (value, "parameters" ),
51 (function, "parameters" ),
52 )
53 result = {
54 "name" : name,
55 "input" : input_schema,
56 "output" : present((value, "output_schema" ), (function, "output_schema" )),
57 "error" : present((value, "error_schema" ), (function, "error_schema" )),
58 }
59 for schema_name in ( "input" , "output" , "error" ):
60 if result[schema_name] is not None :
61 validate_schema(result[schema_name], f "tool { name } { schema_name } " )
62 return result
63
64
65 def validate_schema (value: Any, path: str ) -> None :
66 """Reject malformed structure used by this conservative comparator."""
67 if isinstance (value, bool ):
68 return
69 if not isinstance (value, dict ):
70 raise SchemaError( f " { path } schema must be an object or boolean" )
71
72 schema_type = value.get( "type" )
73 if schema_type is not None :
74 valid_types = { "array" , "boolean" , "integer" , "null" , "number" , "object" , "string" }
75 types = [schema_type] if isinstance (schema_type, str ) else schema_type
76 if (
77 not isinstance (types, list )
78 or not types
79 or not all ( isinstance (item, str ) and item in valid_types for item in types)
80 or len (types) != len ( set (types))
81 ):
82 raise SchemaError( f " { path } .type is invalid" )
83
84 required = value.get( "required" )
85 if required is not None and (
86 not isinstance (required, list )
87 or not all ( isinstance (item, str ) for item in required)
88 or len (required) != len ( set (required))
89 ):
90 raise SchemaError( f " { path } .required must be a list of unique strings" )
91
92 properties = value.get( "properties" )
93 if properties is not None :
94 if not isinstance (properties, dict ) or not all ( isinstance (name, str ) for name in properties):
95 raise SchemaError( f " { path } .properties must be an object" )
96 for name, child in properties.items():
97 validate_schema(child, f " { path } .properties. { name } " )
98
99 if "items" in value:
100 validate_schema(value[ "items" ], f " { path } .items" )
101 for keyword in ( "additionalProperties" , "contains" , "not" , "propertyNames" , "unevaluatedItems" , "unevaluatedProperties" ):
102 if keyword in value:
103 validate_schema(value[keyword], f " { path } . { keyword } " )
104 for keyword in ( "allOf" , "anyOf" , "oneOf" , "prefixItems" ):
105 if keyword in value:
106 children = value[keyword]
107 if not isinstance (children, list ):
108 raise SchemaError( f " { path } . { keyword } must be a list" )
109 for index, child in enumerate (children):
110 validate_schema(child, f " { path } . { keyword } [ { index } ]" )
111
112 enum = value.get( "enum" )
113 if enum is not None and ( not isinstance (enum, list ) or not enum):
114 raise SchemaError( f " { path } .enum must be a non-empty list" )
115
116
117 def index_tools (value: Any) -> dict[ str , dict[ str , Any]]:
118 result: dict[ str , dict[ str , Any]] = {}
119 for raw in tool_list(value):
120 tool = normalize_tool(raw)
121 if tool[ "name" ] in result:
122 raise SchemaError( f "duplicate tool: { tool[ 'name' ] } " )
123 result[tool[ "name" ]] = tool
124 return result
125
126
127 def compare_schema (
128 required: Any,
129 actual: Any,
130 path: str ,
131 findings: list[dict[ str , str ]],
132 additions: list[dict[ str , str ]],
133 input_schema: bool ,
134 ) -> None :
135 if isinstance (required, bool ) or isinstance (actual, bool ):
136 if required != actual:
137 findings.append(
138 { "path" : path, "issue" : "boolean_schema_changed" , "required" : repr (required), "actual" : repr (actual)}
139 )
140 return
141 if required is None :
142 return
143 if not isinstance (required, dict ) or not isinstance (actual, dict ):
144 if required != actual:
145 findings.append(
146 { "path" : path, "issue" : "value_changed" , "required" : repr (required), "actual" : repr (actual)}
147 )
148 return
149
150 required_type = required.get( "type" )
151 actual_type = actual.get( "type" )
152 if required_type != actual_type:
153 findings.append(
154 { "path" : f " { path } .type" , "issue" : "type_changed" , "required" : str (required_type), "actual" : str (actual_type)}
155 )
156
157 required_enum = required.get( "enum" )
158 actual_enum = actual.get( "enum" )
159 if isinstance (required_enum, list ):
160 required_values = {json.dumps(value, sort_keys = True , separators = ( "," , ":" )) for value in required_enum}
161 actual_values = (
162 {json.dumps(value, sort_keys = True , separators = ( "," , ":" )) for value in actual_enum}
163 if isinstance (actual_enum, list )
164 else set ()
165 )
166 if required_values != actual_values:
167 findings.append(
168 { "path" : f " { path } .enum" , "issue" : "enum_changed" , "required" : repr (required_enum), "actual" : repr (actual_enum)}
169 )
170 elif actual_enum is not None :
171 findings.append(
172 { "path" : f " { path } .enum" , "issue" : "enum_changed" , "required" : "<absent>" , "actual" : repr (actual_enum)}
173 )
174
175 strict_keywords = {
176 "$ref" ,
177 "additionalProperties" ,
178 "allOf" ,
179 "anyOf" ,
180 "const" ,
181 "contains" ,
182 "dependentRequired" ,
183 "dependentSchemas" ,
184 "else" ,
185 "exclusiveMaximum" ,
186 "exclusiveMinimum" ,
187 "format" ,
188 "if" ,
189 "maxContains" ,
190 "maxItems" ,
191 "maxLength" ,
192 "maxProperties" ,
193 "maximum" ,
194 "minContains" ,
195 "minItems" ,
196 "minLength" ,
197 "minProperties" ,
198 "minimum" ,
199 "multipleOf" ,
200 "not" ,
201 "oneOf" ,
202 "pattern" ,
203 "patternProperties" ,
204 "prefixItems" ,
205 "propertyNames" ,
206 "then" ,
207 "unevaluatedItems" ,
208 "unevaluatedProperties" ,
209 "uniqueItems" ,
210 }
211 for keyword in sorted (strict_keywords):
212 if keyword not in required and keyword not in actual:
213 continue
214 required_value = required.get(keyword, "<absent>" )
215 actual_value = actual.get(keyword, "<absent>" )
216 if required_value != actual_value:
217 findings.append(
218 {
219 "path" : f " { path } . { keyword } " ,
220 "issue" : "constraint_changed" ,
221 "required" : repr (required_value),
222 "actual" : repr (actual_value),
223 }
224 )
225
226 annotations = {
227 "$comment" ,
228 "$id" ,
229 "$schema" ,
230 "default" ,
231 "deprecated" ,
232 "description" ,
233 "examples" ,
234 "readOnly" ,
235 "title" ,
236 "writeOnly" ,
237 }
238 handled = { "type" , "enum" , "required" , "properties" , "items" } | strict_keywords | annotations
239 for keyword in sorted (( set (required) | set (actual)) - handled):
240 if required.get(keyword, "<absent>" ) != actual.get(keyword, "<absent>" ):
241 findings.append(
242 {
243 "path" : f " { path } . { keyword } " ,
244 "issue" : "unknown_constraint_changed" ,
245 "required" : repr (required.get(keyword, "<absent>" )),
246 "actual" : repr (actual.get(keyword, "<absent>" )),
247 }
248 )
249
250 required_required = set (required.get( "required" , []))
251 actual_required = set (actual.get( "required" , []))
252 if required_required != actual_required:
253 findings.append(
254 {
255 "path" : f " { path } .required" ,
256 "issue" : "required_fields_changed" ,
257 "required" : repr ( sorted (required_required)),
258 "actual" : repr ( sorted (actual_required)),
259 }
260 )
261
262 required_properties = required.get( "properties" , {})
263 actual_properties = actual.get( "properties" , {})
264 if isinstance (required_properties, dict ):
265 if not isinstance (actual_properties, dict ):
266 actual_properties = {}
267 for name, child in required_properties.items():
268 if name not in actual_properties:
269 findings.append(
270 { "path" : f " { path } .properties. { name } " , "issue" : "property_missing" , "required" : "present" , "actual" : "missing" }
271 )
272 else :
273 compare_schema(
274 child,
275 actual_properties[name],
276 f " { path } .properties. { name } " ,
277 findings,
278 additions,
279 input_schema,
280 )
281 for name in sorted ( set (actual_properties) - set (required_properties)):
282 if input_schema and name not in actual_required:
283 additions.append(
284 { "path" : f " { path } .properties. { name } " , "change" : "optional_property_added" }
285 )
286 else :
287 findings.append(
288 {
289 "path" : f " { path } .properties. { name } " ,
290 "issue" : "property_added" ,
291 "required" : "missing" ,
292 "actual" : "present" ,
293 }
294 )
295
296 if "items" in required or "items" in actual:
297 if "items" not in actual:
298 findings.append(
299 { "path" : f " { path } .items" , "issue" : "items_schema_missing" , "required" : "present" , "actual" : "missing" }
300 )
301 elif "items" not in required:
302 findings.append(
303 { "path" : f " { path } .items" , "issue" : "items_schema_added" , "required" : "missing" , "actual" : "present" }
304 )
305 else :
306 compare_schema(
307 required[ "items" ],
308 actual[ "items" ],
309 f " { path } .items" ,
310 findings,
311 additions,
312 input_schema,
313 )
314
315
316 def compare (required_path: Path, actual_path: Path) -> dict[ str , Any]:
317 required = index_tools(load_json(required_path))
318 actual = index_tools(load_json(actual_path))
319 findings: list[dict[ str , str ]] = []
320 additions: list[dict[ str , str ]] = []
321
322 for name in sorted ( set (actual) - set (required)):
323 additions.append({ "path" : f "tools. { name } " , "change" : "tool_added" })
324
325 for name, tool in required.items():
326 if name not in actual:
327 findings.append(
328 { "path" : f "tools. { name } " , "issue" : "tool_missing" , "required" : "present" , "actual" : "missing" }
329 )
330 continue
331 for schema_name in ( "input" , "output" , "error" ):
332 if (tool[schema_name] is None ) != (actual[name][schema_name] is None ):
333 findings.append(
334 {
335 "path" : f "tools. { name } . { schema_name } " ,
336 "issue" : "schema_presence_changed" ,
337 "required" : "present" if tool[schema_name] is not None else "missing" ,
338 "actual" : "present" if actual[name][schema_name] is not None else "missing" ,
339 }
340 )
341 elif tool[schema_name] is not None :
342 compare_schema(
343 tool[schema_name],
344 actual[name][schema_name],
345 f "tools. { name } . { schema_name } " ,
346 findings,
347 additions,
348 schema_name == "input" ,
349 )
350
351 return {
352 "compatible" : not findings,
353 "required_tools" : sorted (required),
354 "actual_tools" : sorted (actual),
355 "additions" : additions,
356 "findings" : findings,
357 "limitations" : [
358 "external JSON Schema $ref targets are not resolved" ,
359 "constraint changes are reported conservatively and require review" ,
360 ],
361 }
362
363
364 def main () -> int :
365 parser = argparse.ArgumentParser( description = __doc__ )
366 parser.add_argument( "required" , type = Path, help = "Harness tool schema JSON" )
367 parser.add_argument( "actual" , type = Path, help = "Environment tool schema JSON" )
368 parser.add_argument( "--output" , type = Path)
369 args = parser.parse_args()
370
371 try :
372 report = compare(args.required, args.actual)
373 except ( OSError , SchemaError) as error:
374 print ( f "ERROR: { error } " , file = sys.stderr)
375 return 2
376
377 text = json.dumps(report, indent = 2 , sort_keys = True ) + " \n "
378 try :
379 if args.output:
380 args.output.parent.mkdir( parents = True , exist_ok = True )
381 args.output.write_text(text, encoding = "utf-8" )
382 else :
383 print (text, end = "" )
384 except OSError as error:
385 print ( f "ERROR: { error } " , file = sys.stderr)
386 return 2
387 return 0 if report[ "compatible" ] else 1
388
389
390 if __name__ == "__main__" :
391 raise SystemExit (main())