(self)
| 237 | |
| 238 | @cached_property |
| 239 | def inputs(self) -> list[MCPInput]: |
| 240 | all_deps = self._cb_info.get("inputs", []) + self._cb_info.get("state", []) |
| 241 | callback_map = get_app().mcp_callback_map |
| 242 | |
| 243 | result: list[MCPInput] = [] |
| 244 | for dep, name, annotation in zip( |
| 245 | all_deps, self._param_names, self._param_annotations |
| 246 | ): |
| 247 | comp_id = str(dep.get("id", "unknown")) |
| 248 | comp = find_component(comp_id) |
| 249 | prop = dep.get("property", "unknown") |
| 250 | id_and_prop = f"{comp_id}.{prop}" |
| 251 | |
| 252 | upstream_cb = callback_map.find_by_output(id_and_prop) |
| 253 | upstream_output = None |
| 254 | if upstream_cb is not None and upstream_cb is not self: |
| 255 | if not upstream_cb.prevents_initial_call: |
| 256 | for out in upstream_cb.outputs: |
| 257 | if out["id_and_prop"] == id_and_prop: |
| 258 | upstream_output = out |
| 259 | break |
| 260 | |
| 261 | initial_value = ( |
| 262 | upstream_output["initial_value"] |
| 263 | if upstream_output is not None |
| 264 | else getattr(comp, prop, None) |
| 265 | ) |
| 266 | |
| 267 | if annotation is not None: |
| 268 | required = not is_nullable(annotation) |
| 269 | else: |
| 270 | required = initial_value is not None |
| 271 | |
| 272 | result.append( |
| 273 | { |
| 274 | "name": name, |
| 275 | "id_and_prop": id_and_prop, |
| 276 | "component_id": comp_id, |
| 277 | "property": prop, |
| 278 | "annotation": annotation, |
| 279 | "component_type": getattr(comp, "_type", None), |
| 280 | "component": comp, |
| 281 | "required": required, |
| 282 | "initial_value": initial_value, |
| 283 | "upstream_output": upstream_output, |
| 284 | } |
| 285 | ) |
| 286 | return result |
| 287 | |
| 288 | # ------------------------------------------------------------------- |
| 289 | # Helpers |
nothing calls this directly
no test coverage detected