Coverage for mcp_chatkit_widget / widget_loader.py: 100%

1"""Widget discovery and loading utilities. 

2 

3This module provides functionality to load curated .widget definitions and 

4expose them in a controlled way so the tooling always targets the same 

5directory. 

6""" 

7 

8import json (empty)

9import logging (empty)

10from dataclasses import dataclass (empty)

11from pathlib import Path (empty)

12from typing import Any (empty)

13 

14 

15LOGGER = logging.getLogger(__name__) (empty)

16 

17 

18@dataclass (empty)

19class WidgetDefinition: (empty)

20 """Represents a loaded widget definition. 

21 

22 Attributes: 

23 name: The widget name 

24 version: Widget format version 

25 json_schema: JSON Schema defining the widget's input structure 

26 output_json_preview: Preview of the rendered widget output 

27 template: Jinja2 template string for rendering the widget 

28 encoded_widget: Base64 encoded widget data (optional) 

29 file_path: Path to the .widget file 

30 """ 

31 

32 name: str (empty)

33 version: str (empty)

34 json_schema: dict[str, Any] (empty)

35 output_json_preview: dict[str, Any] (empty)

36 template: str (empty)

37 encoded_widget: str | None (empty)

38 file_path: Path (empty)

39 

40 

41def _validate_widgets_dir(widgets_dir: Path | None) -> Path: (empty)

42 """Ensure the provided widgets directory exists and is a directory.""" 

43 if widgets_dir is None: (empty)

44 raise ValueError("The widgets_dir argument is required to load widgets.") (empty)

45 resolved_dir = widgets_dir.resolve() (empty)

46 if not resolved_dir.exists(): (empty)

47 raise ValueError(f"Widgets directory does not exist: {resolved_dir}") (empty)

48 if not resolved_dir.is_dir(): (empty)

49 raise ValueError(f"Widgets directory is not a directory: {resolved_dir}") (empty)

50 return resolved_dir (empty)

51 

52 

53def _collect_widget_files(base_dir: Path) -> list[Path]: (empty)

54 """Return .widget files that physically live inside the base directory.""" 

55 base_resolved = base_dir.resolve() (empty)

56 widget_files: list[Path] = [] (empty)

57 seen_files: set[Path] = set() (empty)

58 for widget_file in sorted(base_dir.rglob("*.widget")): (empty)

59 if not widget_file.is_file(): (empty)

60 continue (empty)

61 try: (empty)

62 resolved_path = widget_file.resolve() (empty)

63 except OSError as exc: (empty)

64 LOGGER.warning( (empty)

65 "Skipping widget %s because it could not be resolved (%s)", 

66 widget_file, 

67 exc, 

68 ) 

69 continue (empty)

70 try: (empty)

71 resolved_path.relative_to(base_resolved) (empty)

72 except ValueError: (empty)

73 LOGGER.warning( (empty)

74 "Skipping widget %s because it is outside the widgets directory", 

75 widget_file, 

76 ) 

77 continue (empty)

78 if resolved_path in seen_files: (empty)

79 continue (empty)

80 seen_files.add(resolved_path) (empty)

81 widget_files.append(widget_file) (empty)

82 return widget_files (empty)

83 

84 

85def load_widgets(widgets_dir: Path | None) -> list[WidgetDefinition]: (empty)

86 """Load all .widget files from the curated directory. 

87 

88 Args: 

89 widgets_dir: Path to the curated widgets directory. Must exist. 

90 

91 Returns: 

92 List of loaded WidgetDefinition instances. 

93 """ 

94 base_dir = _validate_widgets_dir(widgets_dir) (empty)

95 widget_files = _collect_widget_files(base_dir) (empty)

96 

97 if not widget_files: (empty)

98 LOGGER.warning("No widget definitions found in %s", base_dir) (empty)

99 

100 return [load_widget(widget_file) for widget_file in widget_files] (empty)

101 

102 

103def discover_widgets(widgets_dir: Path | None = None) -> list[WidgetDefinition]: (empty)

104 """Discover widgets from a curated directory.""" 

105 return load_widgets(widgets_dir) (empty)

106 

107 

108def load_widget(widget_path: Path) -> WidgetDefinition: (empty)

109 """Load a widget definition from a .widget file. 

110 

111 Args: 

112 widget_path: Path to the .widget file 

113 

114 Returns: 

115 WidgetDefinition object containing the parsed widget data 

116 

117 Raises: 

118 ValueError: If the widget file is invalid or missing required fields 

119 json.JSONDecodeError: If the widget file contains invalid JSON 

120 """ 

121 with open(widget_path) as f: (empty)

122 data = json.load(f) (empty)

123 

124 required_fields = ["name", "version", "jsonSchema", "outputJsonPreview", "template"] (empty)

125 missing_fields = [field for field in required_fields if field not in data] (empty)

126 if missing_fields: (empty)

127 raise ValueError( (empty)

128 f"Widget file {widget_path} missing required fields: " 

129 f"{', '.join(missing_fields)}" 

130 ) 

131 

132 template_value = data["template"] (empty)

133 if not isinstance(template_value, str): (empty)

134 raise ValueError(f"Widget template must be a string: {widget_path}") (empty)

135 

136 return WidgetDefinition( (empty)

137 name=data["name"], 

138 version=data["version"], 

139 json_schema=data["jsonSchema"], 

140 output_json_preview=data["outputJsonPreview"], 

141 template=template_value, 

142 encoded_widget=data.get("encodedWidget"), 

143 file_path=widget_path, 

144 ) 

145 

146 

147def get_widget_by_name(name: str, widgets_dir: Path) -> WidgetDefinition | None: (empty)

148 """Get a specific widget by name. 

149 

150 Args: 

151 name: The widget name to search for 

152 widgets_dir: Directory to search for .widget files. 

153 

154 Returns: 

155 WidgetDefinition if found, None otherwise 

156 """ 

157 widgets = load_widgets(widgets_dir) (empty)

158 for widget in widgets: (empty)

159 if widget.name == name: (empty)

160 return widget (empty)

161 return None (empty)