Coverage for src/data_api/checks/mcp.py: 61%

1"""Pin public MCP names and input schemas, excluding descriptive metadata.""" 

2 

3import argparse 

4import asyncio 

5import json 

6from pathlib import Path 

7 

8from fastmcp import Client 

9 

10from data_api.checks.openapi import contract_differences 

11from data_api.mcp.server import mcp 

12 

13BASELINE = Path("tests") / "fixtures" / "mcp_contract.json" 

14 

15 

16def structural(value): 

17 if isinstance(value, dict): 

18 return { 

19 key: ( 

20 {name: structural(field) for name, field in sorted(item.items())} 

21 if key in {"properties", "$defs"} 

22 else structural(item) 

23 ) 

24 for key, item in sorted(value.items()) 

25 if key not in {"title", "description", "examples", "example"} 

26 } 

27 if isinstance(value, list): 

28 return [structural(item) for item in value] 

29 return value 

30 

31 

32async def tool_contract() -> dict: 

33 async with Client(mcp) as client: 

34 tools = await client.list_tools() 

35 return { 

36 tool.name: { 

37 "input": structural(tool.input_schema), 

38 "output": structural(tool.output_schema), 

39 } 

40 for tool in sorted(tools, key=lambda tool: tool.name) 

41 } 

42 

43 

44def main() -> int: 

45 parser = argparse.ArgumentParser(description=__doc__) 

46 parser.add_argument("--write", action="store_true") 

47 args = parser.parse_args() 

48 actual = asyncio.run(tool_contract()) 

49 if args.write: 

50 BASELINE.write_text( 

51 json.dumps(actual, ensure_ascii=False, indent=2, sort_keys=True) + "\n", 

52 encoding="utf-8", 

53 ) 

54 print(f"Wrote {BASELINE}") 

55 return 0 

56 changes = contract_differences(json.loads(BASELINE.read_text(encoding="utf-8")), actual) 

57 for change in changes: 

58 print(change) 

59 return int(bool(changes)) 

60 

61 

62if __name__ == "__main__": 

63 raise SystemExit(main())