(func_param_name, mcp_param_name) per dep, in dep order. Single source of truth for mapping deps to param names. All dict-vs-list branching is confined here.
(self)
| 309 | |
| 310 | @cached_property |
| 311 | def _dep_param_map(self) -> list[tuple[str, str]]: |
| 312 | """(func_param_name, mcp_param_name) per dep, in dep order. |
| 313 | |
| 314 | Single source of truth for mapping deps to param names. |
| 315 | All dict-vs-list branching is confined here. |
| 316 | """ |
| 317 | all_deps = self._cb_info.get("inputs", []) + self._cb_info.get("state", []) |
| 318 | n_deps = len(all_deps) |
| 319 | indices = self._cb_info.get("inputs_state_indices") |
| 320 | |
| 321 | if isinstance(indices, dict): |
| 322 | entries: list[tuple[int, str, str]] = [] |
| 323 | for func_name, idx in indices.items(): |
| 324 | positions = flatten_grouping(idx) |
| 325 | if len(positions) == 1: |
| 326 | entries.append((positions[0], func_name, func_name)) |
| 327 | else: |
| 328 | for pos in positions: |
| 329 | dep = all_deps[pos] if pos < n_deps else {} |
| 330 | comp_id = str(dep.get("id", "unknown")).replace("-", "_") |
| 331 | prop = dep.get("property", "unknown") |
| 332 | entries.append( |
| 333 | (pos, func_name, f"{func_name}_{comp_id}__{prop}") |
| 334 | ) |
| 335 | entries.sort(key=lambda e: e[0]) |
| 336 | result = [(f, m) for _, f, m in entries] |
| 337 | elif self._func_signature is not None: |
| 338 | names = list(self._func_signature.parameters.keys()) |
| 339 | result = [(n, n) for n in names] |
| 340 | else: |
| 341 | result = [] |
| 342 | |
| 343 | while len(result) < n_deps: |
| 344 | fallback = f"param_{len(result)}" |
| 345 | result.append((fallback, fallback)) |
| 346 | return result |
| 347 | |
| 348 | @cached_property |
| 349 | def _param_names(self) -> list[str]: |
nothing calls this directly
no test coverage detected