diff --git a/README.md b/README.md index c166bcb..221e511 100644 --- a/README.md +++ b/README.md @@ -69,24 +69,25 @@ uvx --from streamdeck-mcp streamdeck-mcp-install-skill - **Profile-native authoring** - reads and writes Elgato `ProfilesV3` files directly, with `ProfilesV2` fallback for older installs. - **Hardware inventory** - discovers profile pages, device model names, key geometry, dials, and touch-strip support before writing. -- **Installed plugin discovery** - scans readable Stream Deck plugin manifests with `streamdeck_read_plugins` so agents can find plugin and action UUIDs. -- **Configured plugin action reuse** - reads existing buttons with `streamdeck_read_page` and preserves plugin-specific settings by copying `button.raw`; it does not infer private property-inspector settings. +- **Configured action search** - `streamdeck_find_actions` scans existing profiles/pages for paste-ready plugin buttons and dials (including protected first-party Elgato plugins configured in the app). +- **Installed plugin discovery** - scans readable Stream Deck plugin manifests with `streamdeck_read_plugins` so agents can find plugin and action UUIDs when synthesizing new actions. +- **Configured plugin action reuse** - prefers copying existing actions via `find_actions` or `streamdeck_read_page`'s `button.raw` so private Property Inspector settings stay intact. - **Offline icon generation** - renders button and touch-strip PNGs from about 7,400 bundled Material Design Icons, or from short text labels. -- **Script-backed automations** - creates executable shell scripts in `~/StreamDeckScripts/` and wires them to Stream Deck Open actions. +- **Script-backed automations** - creates executable shell scripts in `~/StreamDeckScripts/` and wires them to Stream Deck Open actions when no plugin action fits. - **Dial and touch-strip support** - installs a minimal bundled Stream Deck plugin when needed so encoder imagery survives app restarts. - **Safe write cycle** - guards against the Elgato app overwriting manifest edits by enforcing a quit, write, relaunch workflow. ## Agentic Workflows -The point is not generic buttons. When your agent also has Slack, Home Assistant, OBS, GitHub, Hue, Spotify, or other MCP servers loaded, it can query those systems first and build around what is actually in your environment. +The point is not generic buttons. Prefer reusing actions the user already configured in the Stream Deck app (Hue, OBS, Home Assistant, Spotify plugins, etc.), then compose them onto a themed page. When your agent also has Slack, Home Assistant, OBS, GitHub, Hue, Spotify, or other MCP servers loaded, it can query those systems to pick the right entities — and still wire the deck through native plugin actions when available, falling back to scripts only when needed. Try prompts like: - **"Make me a control board for Slack."** Query channels, status, and unread state; create channel jumps, status toggles, read-all controls, and dials. -- **"A hello-kitty-themed Home Assistant dashboard for the living room."** Discover living-room entities, then lay out scenes, lights, and media controls in a matching visual style. -- **"OBS control panel based on my actual scenes and audio inputs."** Read scenes, sources, and devices; write scene switches, source toggles, and per-input dial controls. +- **"A hello-kitty-themed Home Assistant dashboard for the living room."** Find existing HA plugin buttons with `streamdeck_find_actions`, then lay them out with a matching visual style — or discover living-room entities via MCP and synthesize/scripts when nothing is configured yet. +- **"OBS control panel based on my actual scenes and audio inputs."** Reuse configured OBS plugin actions when present; otherwise read scenes/sources and write scene switches, source toggles, and per-input dial controls. - **"A dev deck for this repo in Nordic colors."** Read project scripts and GitHub context; create local command buttons, PR links, and CI shortcuts. -- **"A Friday demo deck."** Compose across Zoom, Slack, Hue, and screen recording by generating local scripts and wiring them to one page. +- **"A Friday demo deck."** Compose across Zoom, Slack, Hue, and screen recording by finding configured plugin actions first, then generating local scripts for the gaps. Iteration is cheap: change the prompt, rerun the authoring flow, and get a new profile. @@ -171,10 +172,11 @@ Client config shape: | Tool | What it does | |------|---------------| +| `streamdeck_find_actions` | Searches configured buttons/dials across profiles and returns paste-ready native `action` objects for `streamdeck_write_page`. | | `streamdeck_read_plugins` | Lists installed Stream Deck plugins and declared actions from readable plugin manifests. Protected or binary manifests are reported with diagnostics instead of failing the whole catalog. | | `streamdeck_read_profiles` | Lists desktop profiles, device metadata, page directories, and active profile roots from `ProfilesV3` or `ProfilesV2`. | | `streamdeck_read_page` | Reads a page manifest and returns simplified button details plus raw native action objects. | -| `streamdeck_write_page` | Creates or rewrites a page manifest. Use copied `button.raw` values when reusing configured third-party plugin actions. | +| `streamdeck_write_page` | Creates or rewrites a page manifest. Prefer `action` from `find_actions` / `button.raw` when reusing configured third-party plugin actions. | | `streamdeck_create_icon` | Generates button or touch-strip PNGs from Material Design Icons or text. Icons are bundled offline; unknown names return close-match suggestions. | | `streamdeck_create_action` | Creates an executable shell script in `~/StreamDeckScripts/` and returns an Open action block. | | `streamdeck_restart_app` | Restarts the macOS Stream Deck desktop app after profile changes. | @@ -204,7 +206,7 @@ The skill covers: - Theme palettes, typography strategy, and icon-color guidance. - Dial and touch-strip authoring for Stream Deck + and + XL devices. - Integration recipes for Hue, OBS, Spotify, Home Assistant, Twitch, shell commands, and browser workflows. -- Existing plugin action reuse through `streamdeck_read_page` and `button.raw`. +- Existing plugin action reuse through `streamdeck_find_actions` and `streamdeck_read_page` / `button.raw`. Clients that do not load Claude Code skills can invoke the `design_streamdeck_deck` MCP prompt instead. diff --git a/profile_manager.py b/profile_manager.py index 23ae663..ec9384e 100644 --- a/profile_manager.py +++ b/profile_manager.py @@ -940,14 +940,27 @@ def read_page( layouts[controller_type.lower()] = {"columns": cols, "rows": rows} actions = controller.get("Actions") or {} + if not isinstance(actions, dict): + continue for position, action in sorted( actions.items(), key=lambda item: self._position_sort_key(item[0]), ): - col, row = [int(part) for part in position.split(",")] + if not isinstance(action, dict): + continue + try: + col, row = [int(part) for part in str(position).split(",")] + except ValueError: + continue key = (row * cols + col) if cols else col states = action.get("States") or [{}] - state_index = min(max(int(action.get("State", 0)), 0), max(len(states) - 1, 0)) + try: + state_index = min( + max(int(action.get("State", 0)), 0), + max(len(states) - 1, 0), + ) + except (TypeError, ValueError): + state_index = 0 active_state = states[state_index] if states else {} buttons.append( { @@ -964,7 +977,9 @@ def read_page( "image": active_state.get("Image"), "settings": action.get("Settings", {}), "show_title": active_state.get("ShowTitle"), - "raw": action, + "raw": self._absolutize_action_assets( + copy.deepcopy(action), page_ref.directory_path + ), } ) @@ -985,6 +1000,193 @@ def read_page( "raw_manifest": page_manifest, } + def find_actions( + self, + *, + query: str | None = None, + plugin_uuid: str | None = None, + action_uuid: str | None = None, + plugin_name: str | None = None, + controller: str | None = None, + profile_name: str | None = None, + profile_id: str | None = None, + limit: int = 50, + ) -> dict[str, Any]: + """Search configured buttons/dials across profiles for reuse in write_page. + + Returns paste-ready native ``action`` objects (deep copies with a fresh + ActionID) so agents can place installed plugin actions without guessing + Property Inspector settings. + """ + + if limit < 1: + raise ProfileValidationError("limit must be at least 1.") + capped_limit = min(int(limit), 200) + + controller_filter: str | None = None + if controller is not None and str(controller).strip(): + controller_filter = _normalize_controller(controller).lower() + + query_norm = (query or "").strip().lower() or None + plugin_uuid_norm = (plugin_uuid or "").strip().lower() or None + action_uuid_norm = (action_uuid or "").strip().lower() or None + plugin_name_norm = (plugin_name or "").strip().lower() or None + + if profile_name or profile_id: + profile_dir, profile_manifest = self._resolve_profile( + profile_name=profile_name, profile_id=profile_id + ) + profiles_to_scan = [(profile_dir, profile_manifest)] + else: + profiles_to_scan = [] + if self.profiles_dir.exists(): + for profile_dir in sorted(self.profiles_dir.glob("*.sdProfile")): + try: + profiles_to_scan.append( + (profile_dir, _load_json(profile_dir / "manifest.json")) + ) + except ProfileManagerError: + # One corrupt/half-imported profile must not abort the search. + continue + + matches: list[dict[str, Any]] = [] + truncated = False + seen: set[tuple[str, str, str, str]] = set() + + for profile_dir, profile_manifest in profiles_to_scan: + if truncated: + break + page_refs = self._iter_page_refs_soft(profile_dir, profile_manifest) + profile_label = str(profile_manifest.get("Name", profile_dir.stem)) + for page_ref in page_refs: + if truncated: + break + try: + page_manifest = _load_json(page_ref.manifest_path) + except ProfileManagerError: + continue + + for ctrl in page_manifest.get("Controllers") or []: + if truncated: + break + controller_type = ctrl.get("Type", KEYPAD) + controller_lower = str(controller_type).lower() + if controller_filter and controller_lower != controller_filter: + continue + + cols, _rows = self._resolve_layout( + profile_manifest, page_manifest, controller_type + ) + actions = ctrl.get("Actions") or {} + if not isinstance(actions, dict): + continue + for position, action in sorted( + actions.items(), + key=lambda item: self._position_sort_key(item[0]), + ): + if not isinstance(action, dict): + continue + dedupe_key = ( + profile_dir.stem, + page_ref.directory_id, + controller_lower, + str(position), + ) + if dedupe_key in seen: + continue + try: + col, row = [int(part) for part in str(position).split(",")] + except ValueError: + continue + key = (row * cols + col) if cols else col + + states = action.get("States") or [{}] + try: + state_index = min( + max(int(action.get("State", 0)), 0), + max(len(states) - 1, 0), + ) + except (TypeError, ValueError): + state_index = 0 + active_state = states[state_index] if states else {} + plugin = action.get("Plugin") or {} + hit_plugin_uuid = str(plugin.get("UUID") or "") + hit_plugin_name = str(plugin.get("Name") or "") + hit_action_uuid = str(action.get("UUID") or "") + hit_name = str(action.get("Name") or "") + hit_title = str(active_state.get("Title") or "") + + if plugin_uuid_norm and hit_plugin_uuid.lower() != plugin_uuid_norm: + continue + if action_uuid_norm and hit_action_uuid.lower() != action_uuid_norm: + continue + if plugin_name_norm and hit_plugin_name.lower() != plugin_name_norm: + continue + if query_norm: + haystack = " ".join( + [ + hit_plugin_name, + hit_name, + hit_title, + hit_plugin_uuid, + hit_action_uuid, + ] + ).lower() + if query_norm not in haystack: + continue + + if len(matches) >= capped_limit: + truncated = True + break + + seen.add(dedupe_key) + paste_action = self._absolutize_action_assets( + copy.deepcopy(action), page_ref.directory_path + ) + paste_action["ActionID"] = str(uuid.uuid4()) + matches.append( + { + "profile_name": profile_label, + "profile_id": profile_dir.stem, + "page_name": page_ref.name + or page_manifest.get("Name") + or page_ref.directory_id, + "directory_id": page_ref.directory_id, + "page_index": page_ref.page_index, + "controller": controller_lower, + "key": key, + "position": position, + "title": hit_title or None, + "name": hit_name or None, + "plugin_name": hit_plugin_name or None, + "plugin_uuid": hit_plugin_uuid or None, + "action_uuid": hit_action_uuid or None, + "settings": copy.deepcopy(action.get("Settings") or {}), + "action": paste_action, + # Paste-ready write_page button: includes controller so + # encoder/dial hits land on the Encoder grid, not Keypad. + "button": { + "controller": controller_lower, + "key": key, + "action": paste_action, + }, + } + ) + + result: dict[str, Any] = { + "count": len(matches), + "truncated": truncated, + "actions": matches, + } + if not matches: + result["hint"] = ( + "No configured actions matched. Configure the action once in the " + "Stream Deck app (then re-run find), or synthesize via " + "streamdeck_read_plugins + plugin_uuid/action_uuid/settings on " + "streamdeck_write_page." + ) + return result + def write_page( self, *, @@ -1441,7 +1643,13 @@ def _resolve_profile( matches: list[tuple[Path, dict[str, Any]]] = [] for profile_dir in sorted(self.profiles_dir.glob("*.sdProfile")): - manifest = _load_json(profile_dir / "manifest.json") + try: + manifest = _load_json(profile_dir / "manifest.json") + except ProfileManagerError: + # Targeted profile_id must surface the real JSON error, not "not found". + if profile_id and profile_dir.stem.lower() == profile_id.lower(): + raise + continue if profile_id and profile_dir.stem.lower() == profile_id.lower(): return profile_dir, manifest if profile_name and str(manifest.get("Name", "")).lower() == profile_name.lower(): @@ -1538,6 +1746,128 @@ def _page_refs_v2(self, profiles_path: Path, profile_manifest: dict[str, Any]) - ) return page_refs + def _iter_page_refs_soft( + self, profile_dir: Path, profile_manifest: dict[str, Any] + ) -> list[PageRef]: + """Like ``_page_refs``, but skip missing/corrupt page manifests. + + Used by ``find_actions`` so one bad page cannot abort a multi-page search. + Direct read/write paths keep strict ``_page_refs`` so invalid JSON surfaces + as a clear error for the targeted page. + """ + + profiles_path = profile_dir / "Profiles" + if not profiles_path.exists(): + return [] + + version = str(profile_manifest.get("Version", "2.0")) + if version.startswith("3"): + return self._page_refs_v3_soft(profiles_path, profile_manifest) + return self._page_refs_v2_soft(profiles_path, profile_manifest) + + def _page_refs_v3_soft( + self, profiles_path: Path, profile_manifest: dict[str, Any] + ) -> list[PageRef]: + page_refs: list[PageRef] = [] + ordered_page_ids: list[tuple[str, bool]] = [] + pages = profile_manifest.get("Pages", {}) + default_uuid = pages.get("Default") + if default_uuid: + ordered_page_ids.append((default_uuid, True)) + for page_uuid in pages.get("Pages", []): + ordered_page_ids.append((page_uuid, False)) + + used: set[str] = set() + for page_index, (page_uuid, is_default) in enumerate(ordered_page_ids): + directory_id = str(page_uuid).upper() + manifest_path = profiles_path / directory_id / "manifest.json" + if not manifest_path.exists(): + continue + used.add(directory_id) + page_ref = self._try_build_page_ref( + page_index=page_index, + directory_id=directory_id, + page_uuid=str(page_uuid).lower(), + manifest_path=manifest_path, + version=str(profile_manifest.get("Version", "unknown")), + mapping="page-uuid", + is_default=is_default, + is_current=_normalize_uuid(str(page_uuid)) + == _normalize_uuid(str(pages.get("Current", ""))), + ) + if page_ref is not None: + page_refs.append(page_ref) + + for manifest_path in sorted(profiles_path.glob("*/manifest.json")): + directory_id = manifest_path.parent.name.upper() + if directory_id in used: + continue + page_ref = self._try_build_page_ref( + page_index=len(page_refs), + directory_id=directory_id, + page_uuid=directory_id.lower() if _looks_like_uuid(directory_id) else None, + manifest_path=manifest_path, + version=str(profile_manifest.get("Version", "unknown")), + mapping="unreferenced", + is_default=False, + is_current=False, + ) + if page_ref is not None: + page_refs.append(page_ref) + + return page_refs + + def _page_refs_v2_soft( + self, profiles_path: Path, profile_manifest: dict[str, Any] + ) -> list[PageRef]: + page_refs: list[PageRef] = [] + entries = sorted( + (Path(entry.path) for entry in os.scandir(profiles_path) if entry.is_dir()), + key=lambda path: path.name.lower(), + ) + for page_index, page_dir in enumerate(entries): + page_ref = self._try_build_page_ref( + page_index=page_index, + directory_id=page_dir.name, + page_uuid=None, + manifest_path=page_dir / "manifest.json", + version=str(profile_manifest.get("Version", "unknown")), + mapping="directory-order", + is_default=False, + is_current=False, + ) + if page_ref is not None: + page_refs.append(page_ref) + return page_refs + + def _try_build_page_ref( + self, + *, + page_index: int, + directory_id: str, + page_uuid: str | None, + manifest_path: Path, + version: str, + mapping: str, + is_default: bool, + is_current: bool, + ) -> PageRef | None: + """Build a PageRef, skipping missing/corrupt page manifests.""" + + try: + return self._build_page_ref( + page_index=page_index, + directory_id=directory_id, + page_uuid=page_uuid, + manifest_path=manifest_path, + version=version, + mapping=mapping, + is_default=is_default, + is_current=is_current, + ) + except ProfileManagerError: + return None + def _build_page_ref( self, *, @@ -1602,9 +1932,16 @@ def _resolve_layout( if page_manifest: actions = _controller_actions(page_manifest, controller_type) - if actions: - cols = max(int(position.split(",")[0]) for position in actions) + 1 - rows = max(int(position.split(",")[1]) for position in actions) + 1 + positions: list[tuple[int, int]] = [] + for position in actions: + try: + col_s, row_s = str(position).split(",") + positions.append((int(col_s), int(row_s))) + except (TypeError, ValueError): + continue + if positions: + cols = max(col for col, _row in positions) + 1 + rows = max(row for _col, row in positions) + 1 if cols > 0 and rows > 0: return cols, rows @@ -1670,8 +2007,17 @@ def _materialize_action( else: raise ProfileValidationError("Button action must be an object or JSON string.") + # Raw/pasted actions may be reused across multiple slots; always mint a + # fresh ActionID so multi-copy pastes (including reuse of one find_actions + # hit) never collide on the destination page. + if raw_action is not None: + action["ActionID"] = str(uuid.uuid4()) + states = copy.deepcopy(action.get("States") or [{}]) - state_index = min(max(int(action.get("State", 0)), 0), max(len(states) - 1, 0)) + try: + state_index = min(max(int(action.get("State", 0)), 0), max(len(states) - 1, 0)) + except (TypeError, ValueError): + state_index = 0 state_data = copy.deepcopy(states[state_index] or {}) if button.get("title") is not None: @@ -1732,6 +2078,7 @@ def _materialize_action( states[state_index] = state_data action["States"] = states + self._ensure_action_assets_on_page(action, page_dir) return action def _build_action_from_fields( @@ -2106,9 +2453,97 @@ def _copy_icon_to_page(self, source_path: Path, page_dir: Path) -> str: return f"Images/{target_name}" + def _absolutize_action_assets(self, action: dict[str, Any], page_dir: Path) -> dict[str, Any]: + """Rewrite page-local Images/ paths to absolute source paths for paste reuse.""" + + for state in action.get("States") or []: + if not isinstance(state, dict): + continue + resolved = self._resolve_page_local_asset(state.get("Image"), page_dir) + if resolved is not None: + state["Image"] = resolved + + encoder = action.get("Encoder") + if isinstance(encoder, dict): + for field in ("Icon", "background"): + resolved = self._resolve_page_local_asset(encoder.get(field), page_dir) + if resolved is not None: + encoder[field] = resolved + return action + + def _resolve_page_local_asset(self, value: Any, page_dir: Path) -> str | None: + if not isinstance(value, str) or not value.strip(): + return None + path = Path(value).expanduser() + page_root = page_dir.resolve() + images_root = (page_dir / "Images").resolve() + + if path.is_absolute(): + if not path.exists(): + return None + resolved = path.resolve() + try: + resolved.relative_to(images_root) + except ValueError: + return None + return str(resolved) + + candidate = (page_dir / value).resolve() + if not candidate.exists(): + return None + try: + candidate.relative_to(page_root) + candidate.relative_to(images_root) + except ValueError: + # Reject ../ escapes and anything outside the page Images/ tree. + return None + return str(candidate) + + def _ensure_action_assets_on_page(self, action: dict[str, Any], page_dir: Path) -> None: + """Copy absolute (or foreign) asset paths into the destination page Images/.""" + + for state in action.get("States") or []: + if not isinstance(state, dict): + continue + relocated = self._ensure_asset_on_page(state.get("Image"), page_dir) + if relocated is not None: + state["Image"] = relocated + + encoder = action.get("Encoder") + if isinstance(encoder, dict): + for field in ("Icon", "background"): + relocated = self._ensure_asset_on_page(encoder.get(field), page_dir) + if relocated is not None: + encoder[field] = relocated + + def _ensure_asset_on_page(self, value: Any, page_dir: Path) -> str | None: + if not isinstance(value, str) or not value.strip(): + return None + path = Path(value).expanduser() + page_images = (page_dir / "Images").resolve() + + if path.is_absolute(): + if not path.exists(): + return None + resolved = path.resolve() + try: + relative = resolved.relative_to(page_images) + return f"Images/{relative.as_posix()}" + except ValueError: + return self._copy_icon_to_page(resolved, page_dir) + + # Relative path already on the destination page — keep as-is. + if (page_dir / value).exists(): + return None + return None + def _position_sort_key(self, position: str) -> tuple[int, int]: - col, row = [int(part) for part in position.split(",")] - return (row, col) + try: + col, row = [int(part) for part in str(position).split(",")] + return (row, col) + except (TypeError, ValueError): + # Sort malformed keys last so one bad slot cannot abort a scan/read. + return (10**9, 10**9) def _generate_directory_id(self, *, length: int = 27) -> str: alphabet = string.ascii_uppercase + string.digits diff --git a/profile_server.py b/profile_server.py index 17a8570..b4841c8 100644 --- a/profile_server.py +++ b/profile_server.py @@ -17,6 +17,7 @@ from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( + CallToolResult, GetPromptResult, Prompt, PromptArgument, @@ -49,6 +50,10 @@ ) +def _error_result(text: str) -> CallToolResult: + return CallToolResult(content=[TextContent(type="text", text=text)], isError=True) + + def _scalar_or_string(base_type: str) -> dict[str, Any]: """Schema helper: field accepts either a native JSON value or a string form of it. Works around MCP clients (notably Claude Code's tool-call @@ -290,6 +295,68 @@ async def list_tools() -> list[Tool]: } return [ + Tool( + name="streamdeck_find_actions", + description=( + "Search configured buttons and dials across Stream Deck profiles/pages " + "and return paste-ready native action objects for streamdeck_write_page. " + "Prefer this FIRST when building layouts that reuse installed plugins " + "(Home Assistant, Hue, OBS, Spotify, Wave Link, etc.): each hit includes " + "a 'button' object ready for write_page (controller + key + action) so " + "encoder/dial hits stay on the Encoder grid. You can also pass " + "hit.action with controller=hit.controller yourself. Filters: " + "query (substring), plugin_uuid, action_uuid, plugin_name, controller " + "(keypad|encoder), optional profile scope. Returns a fresh ActionID per " + "hit so multi-copy pastes are safe. For protected Elgato manifests that " + "streamdeck_read_plugins cannot enumerate, this is the reliable path — " + "configure once in the Stream Deck app, then find and reuse." + ), + inputSchema={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": ( + "Case-insensitive substring matched against plugin name, " + "action name, title, plugin_uuid, and action_uuid." + ), + }, + "plugin_uuid": { + "type": "string", + "description": "Exact plugin UUID filter (case-insensitive).", + }, + "action_uuid": { + "type": "string", + "description": "Exact action UUID filter (case-insensitive).", + }, + "plugin_name": { + "type": "string", + "description": ("Exact plugin display name filter (case-insensitive)."), + }, + "controller": { + "type": "string", + "description": ( + "Limit to 'keypad' (buttons) or 'encoder' (dials). Omit to search both." + ), + }, + "profile_name": { + "type": "string", + "description": "Limit the scan to one profile by display name.", + }, + "profile_id": { + "type": "string", + "description": ("Limit the scan to one profile by .sdProfile folder stem."), + }, + "limit": { + **_scalar_or_string("integer"), + "description": ( + "Max hits to return (default 50, hard cap 200). " + "When more match, truncated=true." + ), + }, + }, + }, + ), Tool( name="streamdeck_read_plugins", description=( @@ -297,13 +364,12 @@ async def list_tools() -> list[Tool]: "the user's Elgato Plugins directory, including each action's UUID, " "Name, States, and — best-effort — the Settings fields its Property " "Inspector expects (parsed from the action's PI HTML/JS). Use this " - "before authoring a third-party plugin action with " - "streamdeck_write_page: it gives you the plugin_uuid, action_uuid, " - "state_count, and settings_fields you need. Notes: some first-party " - "Elgato plugins (OBS, Home Assistant, Hue, Spotify, Zoom, …) ship " - "ELGATO-protected/binary manifests and are reported with " - "parse_status instead of action metadata — for those, copy a " - "configured button via streamdeck_read_page's raw action. " + "to synthesize a NEW third-party action when streamdeck_find_actions " + "finds no configured instance. Prefer find_actions (or read_page.raw) " + "when the user already configured the button/dial in the Stream Deck " + "app — especially for first-party Elgato plugins (OBS, Hue, Spotify, " + "Zoom, …) that ship ELGATO-protected/binary manifests and are " + "reported with parse_status instead of action metadata. " "settings_fields are heuristic: a missing field is not proof it is " "unused, and a listed field is not guaranteed required." ), @@ -349,7 +415,13 @@ async def list_tools() -> list[Tool]: ), Tool( name="streamdeck_read_page", - description="Read a profile page by profile name or ID and page index or directory ID.", + description=( + "Read a profile page by profile name or ID and page index or " + "directory ID. Each button/dial includes a 'raw' native action " + "object — copy it into streamdeck_write_page's button 'action' " + "field to reuse configured plugin settings without guessing. For " + "searching across many pages, prefer streamdeck_find_actions." + ), inputSchema={ "type": "object", "properties": { @@ -385,15 +457,14 @@ async def list_tools() -> list[Tool]: name="streamdeck_write_page", description=( "Create a new page or replace/update an existing Stream Deck desktop " - "page manifest. Buttons can be page-nav (action_type), script-backed " - "Open actions (path, from streamdeck_create_action), the bundled MCP " - "dial, OR a native action for ANY installed third-party plugin — pass " - "plugin_uuid + action_uuid + settings (discover these with " - "streamdeck_read_plugins) and the tool writes the same native action " - "shape Stream Deck itself uses (Plugin metadata, State, States, " - "Settings), defaulting the States count and Plugin name/version from " - "the installed manifest so the button survives Elgato's " - "overwrite-on-quit. Validation: an unknown plugin_uuid/action_uuid " + "page manifest. Authoring preference order: (1) paste a native " + "'action' from streamdeck_find_actions or streamdeck_read_page's " + "button.raw — best for installed plugins with private Settings; " + "(2) synthesize via plugin_uuid + action_uuid + settings from " + "streamdeck_read_plugins when no configured instance exists; " + "(3) script-backed Open actions via path / streamdeck_create_action " + "for custom automation; also supports page-nav (action_type) and " + "the bundled MCP dial. Validation: unknown plugin_uuid/action_uuid " "raises a clear error listing what is installed; missing settings " "fields are reported non-fatally in the result's 'warnings'. " "IMPORTANT: the Elgato desktop app overwrites profile manifests from " @@ -645,7 +716,7 @@ async def list_tools() -> list[Tool]: @server.call_tool() -async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: +async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CallToolResult: """Handle profile writer tool calls.""" # Normalize stringified args from MCP clients that serialize non-string @@ -654,7 +725,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: # the values back to the types the handlers expect. arguments = _coerce_arguments( arguments, - ints=("page_index", "font_size", "state", "key"), + ints=("page_index", "font_size", "state", "key", "limit"), nums=("icon_scale",), bools=( "clear_existing", @@ -670,6 +741,19 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: ) try: + if name == "streamdeck_find_actions": + payload = manager.find_actions( + query=arguments.get("query"), + plugin_uuid=arguments.get("plugin_uuid"), + action_uuid=arguments.get("action_uuid"), + plugin_name=arguments.get("plugin_name"), + controller=arguments.get("controller"), + profile_name=arguments.get("profile_name"), + profile_id=arguments.get("profile_id"), + limit=arguments.get("limit", 50), + ) + return [TextContent(type="text", text=json.dumps(payload, indent=2))] + if name == "streamdeck_read_plugins": plugins = manager.list_plugins( plugin_id=arguments.get("plugin_id"), @@ -696,16 +780,11 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: # handler iterate the string one char at a time. raw_buttons = arguments.get("buttons") if isinstance(raw_buttons, str): - return [ - TextContent( - type="text", - text=( - "❌ 'buttons' was a string but not a valid JSON " - "array. Pass either a JSON array or a JSON-encoded " - "string of an array." - ), - ) - ] + return _error_result( + "❌ 'buttons' was a string but not a valid JSON " + "array. Pass either a JSON array or a JSON-encoded " + "string of an array." + ) result = manager.write_page( profile_name=arguments.get("profile_name"), profile_id=arguments.get("profile_id"), @@ -729,16 +808,11 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: try: batch = json.loads(batch) except json.JSONDecodeError as exc: - return [ - TextContent( - type="text", - text=( - "❌ 'icons' was a string but not valid JSON: " - f"{exc}. Pass either a JSON array or a " - "JSON-encoded string." - ), - ) - ] + return _error_result( + "❌ 'icons' was a string but not valid JSON: " + f"{exc}. Pass either a JSON array or a " + "JSON-encoded string." + ) if batch is not None: icons_result = manager.create_icons(batch) return [ @@ -779,20 +853,20 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: result = manager.restart_app() return [TextContent(type="text", text=json.dumps(result, indent=2))] - return [TextContent(type="text", text=f"❌ Unknown tool: {name}")] + return _error_result(f"❌ Unknown tool: {name}") except StreamDeckAppRunningError as exc: - return [TextContent(type="text", text=f"⚠️ {exc}")] + return _error_result(f"⚠️ {exc}") except ( ProfileManagerError, ProfileNotFoundError, PageNotFoundError, ProfileValidationError, ) as exc: - return [TextContent(type="text", text=f"❌ {exc}")] + return _error_result(f"❌ {exc}") except Exception as exc: # pragma: no cover logger.exception("Unexpected error in %s", name) - return [TextContent(type="text", text=f"❌ Unexpected error: {exc}")] + return _error_result(f"❌ Unexpected error: {exc}") @server.list_prompts() diff --git a/pyproject.toml b/pyproject.toml index a830ac4..16291a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ classifiers = [ "Programming Language :: Python :: 3.12", ] dependencies = [ - "mcp[cli]>=1.6.0", + "mcp[cli]>=1.25.0", "streamdeck>=0.9.0", "pillow>=10.0.0", ] diff --git a/skills/streamdeck-profile/SKILL.md b/skills/streamdeck-profile/SKILL.md index d2cab66..99a28ee 100644 --- a/skills/streamdeck-profile/SKILL.md +++ b/skills/streamdeck-profile/SKILL.md @@ -6,11 +6,26 @@ Use this skill when you need to configure Elgato Stream Deck desktop profiles wi 1. Call `streamdeck_read_profiles` to discover the active profiles root and page `directory_id` values. 2. Call `streamdeck_read_page` before editing an existing page so you can inspect the current native action objects. -3. Call `streamdeck_read_plugins` before authoring installed plugin actions. It discovers plugin/action UUIDs from readable manifests, but does not infer property inspector settings schemas. Prefer copying configured plugin actions from `streamdeck_read_page` via `button.raw`. -4. Use `streamdeck_create_icon` for button art. Pass `icon` with a Material Design Icons name like `mdi:cpu-64-bit`, `mdi:volume-high`, or `mdi:github` (~7400 glyphs bundled) plus `icon_color` and `bg_color` for a glyph; or pass `text` alone for a text-only icon. `icon` and `text` are mutually exclusive — set the button's `title` on `streamdeck_write_page` for labels, since Elgato overlays titles on images. -5. Use `streamdeck_create_action` when you want a shell-command button. It creates an executable script in `~/StreamDeckScripts/` and returns a ready-to-insert Open action block. -6. Call `streamdeck_write_page` with `directory_id` for the safest updates on existing pages. -7. Call `streamdeck_restart_app` on macOS if the Stream Deck desktop app does not pick up changes immediately. +3. **Reuse installed plugin actions first.** Call `streamdeck_find_actions` (by `query`, `plugin_name`, or `plugin_uuid`) to locate configured buttons/dials across profiles. Paste each hit's `action` into `streamdeck_write_page`. If you already know the source page, copy `button.raw` from `streamdeck_read_page` instead. +4. Only when nothing is configured: call `streamdeck_read_plugins` for plugin/action UUIDs and best-effort `settings_fields`, then synthesize with `plugin_uuid` + `action_uuid` + `settings` on write. +5. Use `streamdeck_create_icon` for button art. Pass `icon` with a Material Design Icons name like `mdi:cpu-64-bit`, `mdi:volume-high`, or `mdi:github` (~7400 glyphs bundled) plus `icon_color` and `bg_color` for a glyph; or pass `text` alone for a text-only icon. `icon` and `text` are mutually exclusive — set the button's `title` on `streamdeck_write_page` for labels, since Elgato overlays titles on images. +6. Use `streamdeck_create_action` when you need a custom shell-command button (no matching plugin action). It creates an executable script in `~/StreamDeckScripts/` and returns a ready-to-insert Open action block. +7. Call `streamdeck_write_page` with `directory_id` for the safest updates on existing pages. +8. Call `streamdeck_restart_app` on macOS if the Stream Deck desktop app does not pick up changes immediately. + +## Reusing a found action + +```json +{ + "key": 0, + "title": "Office Light", + "action": { "...paste hit.action from streamdeck_find_actions..." } +} +``` + +For dials/encoders, prefer the hit's ready-made `button` object (includes `controller`) or pass `controller: hit.controller` explicitly — otherwise write_page defaults to keypad. + +Example find call: `streamdeck_find_actions(query="home assistant", limit=20)`. ## Practical Notes @@ -19,6 +34,7 @@ Use this skill when you need to configure Elgato Stream Deck desktop profiles wi - `ProfilesV2` uses opaque page directory names, so treat `directory_id` as the source of truth for updates. - `streamdeck_write_page` replaces the page by default because `clear_existing` defaults to `true`. - Button inputs can use `key` for linear indexing or `position` for native `col,row` coordinates. +- `streamdeck_find_actions` returns a fresh `ActionID` on each hit so multi-copy pastes are safe. ## Button Shape @@ -33,7 +49,7 @@ Each `buttons[]` item can use: } ``` -Or a raw native action object: +Or a raw native action object (preferred for plugin reuse): ```json { diff --git a/streamdeck_assets/skill/streamdeck-designer/SKILL.md b/streamdeck_assets/skill/streamdeck-designer/SKILL.md index 5e9a92b..89e4990 100644 --- a/streamdeck_assets/skill/streamdeck-designer/SKILL.md +++ b/streamdeck_assets/skill/streamdeck-designer/SKILL.md @@ -61,13 +61,11 @@ For **Original / MK.2 / Neo / Mini**: fewer slots, so every slot matters more. O ### 5. Discover integrations at authoring time, don't guess endpoints -When the user names an integration, check what you have access to **right now** in this session: +When the user names an integration, prefer **reuse of already-configured Stream Deck plugin actions**, then synthesize, then scripts: -- If the user wants to use an **installed Stream Deck plugin action**, call `streamdeck_read_plugins` first to discover installed plugins and action UUIDs. This is a manifest catalog only: it does not infer property-inspector settings schemas. Prefer copying configured actions from `streamdeck_read_page` via `button.raw`; only synthesize new third-party actions when you already have a known-good `plugin_uuid`, `action_uuid`, and `settings` shape. -- If a matching MCP is available (Hue MCP, Home Assistant MCP, Spotify MCP, OBS MCP), **use it** to enumerate the user's real devices, scenes, rooms, playlists. The goal is that when the user presses "Chill Scene" on the deck, their actual living-room Hue scene named "Chill" activates — not a placeholder. -- If no matching MCP is available, **ask the user** for the endpoints, credentials, bridge IPs, auth tokens you need. Tell them plainly what you're going to do with each (e.g. "I'll store your Hue API key in ~/StreamDeckScripts/.env — nothing outside your machine"). - -Bake the discovered invocations into **shell scripts** via `streamdeck_create_action`. Those scripts run standalone — they don't need Claude or the MCP server to be present at press time. That's the whole point of the static authoring model. +1. **Find configured actions first.** Call `streamdeck_find_actions` with a `query` / `plugin_name` / `plugin_uuid` for the integration (e.g. `"Home Assistant"`, `"Hue"`, `"OBS"`). Each hit's `action` field is paste-ready for `streamdeck_write_page` — it preserves private Property Inspector `Settings`. Also works for dials (`controller: "encoder"`). If you already know the source page, `streamdeck_read_page` and copy `button.raw` instead. +2. **Synthesize only when nothing is configured.** Call `streamdeck_read_plugins` for plugin/action UUIDs and best-effort `settings_fields` (inferred from PI HTML/JS when the manifest is readable). Pass `plugin_uuid` + `action_uuid` + `settings` on write. Protected first-party Elgato plugins often lack catalog metadata — for those, ask the user to configure one button in the Stream Deck app, then re-run `find_actions`. +3. **Scripts as last resort** for custom automation or when no Stream Deck plugin exists. If a matching MCP is available (Hue, Home Assistant, Spotify, OBS), use it to enumerate real devices/scenes, then bake invocations into shell scripts via `streamdeck_create_action`. Those scripts run standalone — they don't need Claude or the MCP server at press time. **Never inline credentials into the action scripts.** Put secrets in `~/StreamDeckScripts/.env`; source it in each generated script: @@ -109,19 +107,19 @@ Icon param conventions: If an MDI name misses, the tool returns close-match suggestions. Take them — the exact right glyph for a concept is often named something slightly different (e.g. `mdi:cog` vs `mdi:gear`). `references/icons.md` has 30+ categorized exemplars. -### 7. Wire actions with shell scripts +### 7. Wire actions — reuse, then synthesize, then scripts + +**Preference order:** -For commands, use `streamdeck_create_action(name, command)`. It: -- Writes an executable script to `~/StreamDeckScripts/.sh`. -- Returns a native `com.elgato.streamdeck.system.open` action block you pass into `streamdeck_write_page`. +1. **Reuse configured plugin actions.** `streamdeck_find_actions(query="…")` → pass each hit's `action` into `streamdeck_write_page` buttons (override `title` / `icon_path` as needed). Or copy `button.raw` from `streamdeck_read_page` when you already know the source page. +2. **Synthesize** with `plugin_uuid` + `action_uuid` + `settings` from `streamdeck_read_plugins` when no configured instance exists and you know the settings shape. +3. **Scripts** via `streamdeck_create_action(name, command)` for custom shell automation. It writes `~/StreamDeckScripts/.sh` and returns a native Open action block. For page navigation between pages in the same profile, use `action_type: "next_page"` or `action_type: "previous_page"` on the button. For opening a URL, `command="open 'https://...'"` works on macOS; Windows uses `start`. For switching to a **different profile** on press, that's not exposed as a convenience field — build the action object explicitly with `plugin_uuid: "com.elgato.streamdeck.profile"` and the profile-switch action UUID. Most decks don't need this; if you need it, check the Elgato SDK reference. -For installed third-party plugins, discover action UUIDs with `streamdeck_read_plugins`. If the user already configured the action somewhere in the Stream Deck app, use `streamdeck_read_page` on that page and copy the button's `raw` action object into the new location; this preserves plugin-specific `Settings` without guessing. If you only have the plugin manifest, you still need the correct settings shape from the user, the plugin docs, or an existing configured action. - -No other action types exist standalone in Phase 1. In particular: there is no "call back to Claude" action yet — pressing a key fires a shell script or switches a page, nothing else. See Phase 2 in the repo roadmap if the user asks about live/dynamic behavior. +No other action types exist standalone in Phase 1. In particular: there is no "call back to Claude" action yet — pressing a key fires a plugin action, shell script, or page switch. See Phase 2 in the repo roadmap if the user asks about live/dynamic behavior. ### 8. Pick encoder layouts with the Phase 1 constraint in mind diff --git a/tests/test_profile_manager.py b/tests/test_profile_manager.py index d2fa057..be21fb8 100644 --- a/tests/test_profile_manager.py +++ b/tests/test_profile_manager.py @@ -478,6 +478,492 @@ def test_plugin_catalog_supports_reusing_configured_raw_plugin_actions( assert copied["title"] == "Office Copy" +def test_find_actions_matches_keypad_by_plugin_uuid_and_query( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + page_path = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + page_manifest = json.loads(page_path.read_text()) + configured_action = { + "ActionID": "ha-light-action", + "LinkedTitle": False, + "Name": "Entity", + "Plugin": { + "Name": "Home Assistant", + "UUID": "de.perdoctus.streamdeck.homeassistant", + "Version": "3.7.1", + }, + "Settings": {"entityId": "light.office", "mode": "toggle"}, + "State": 0, + "States": [{"Title": "Office", "Image": "Images/office.png"}], + "UUID": "de.perdoctus.streamdeck.homeassistant.entity", + } + page_manifest["Controllers"][0]["Actions"]["2,0"] = configured_action + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + by_uuid = manager.find_actions(plugin_uuid="de.perdoctus.streamdeck.homeassistant") + assert by_uuid["count"] == 1 + assert by_uuid["truncated"] is False + hit = by_uuid["actions"][0] + assert hit["plugin_name"] == "Home Assistant" + assert hit["title"] == "Office" + assert hit["controller"] == "keypad" + assert hit["position"] == "2,0" + assert hit["settings"] == {"entityId": "light.office", "mode": "toggle"} + assert hit["action"]["Settings"] == {"entityId": "light.office", "mode": "toggle"} + assert hit["action"]["ActionID"] != "ha-light-action" + assert hit["action"]["UUID"] == "de.perdoctus.streamdeck.homeassistant.entity" + # Page-local Images/ paths are absolutized so paste into another page can copy them. + assert Path(hit["action"]["States"][0]["Image"]).is_absolute() or hit["action"]["States"][0][ + "Image" + ].startswith("Images/") + + by_query = manager.find_actions(query="office") + assert by_query["count"] == 1 + assert by_query["actions"][0]["action_uuid"] == ("de.perdoctus.streamdeck.homeassistant.entity") + + +def test_find_actions_absolutizes_existing_page_assets( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + page_dir = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + ) + images_dir = page_dir / "Images" + images_dir.mkdir(parents=True, exist_ok=True) + icon_path = images_dir / "office.png" + icon_path.write_bytes( + b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" + b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f" + b"\x00\x00\x01\x01\x00\x05\x18\xd8N\x00\x00\x00\x00IEND\xaeB`\x82" + ) + + page_path = page_dir / "manifest.json" + page_manifest = json.loads(page_path.read_text()) + page_manifest["Controllers"][0]["Actions"]["2,0"] = { + "ActionID": "ha-light-action", + "Name": "Entity", + "Plugin": { + "Name": "Home Assistant", + "UUID": "de.perdoctus.streamdeck.homeassistant", + "Version": "3.7.1", + }, + "Settings": {"entityId": "light.office"}, + "State": 0, + "States": [{"Title": "Office", "Image": "Images/office.png"}], + "UUID": "de.perdoctus.streamdeck.homeassistant.entity", + } + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + hit = manager.find_actions(plugin_uuid="de.perdoctus.streamdeck.homeassistant")["actions"][0] + image = hit["action"]["States"][0]["Image"] + assert Path(image).is_absolute() + assert Path(image).exists() + + manager.write_page( + profile_name="Default Profile", + directory_id="CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC", + buttons=[{"key": 0, "action": hit["action"]}], + ) + target_page = manager.read_page( + profile_name="Default Profile", + directory_id="CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC", + ) + target_image = target_page["buttons"][0]["image"] + assert target_image.startswith("Images/") + target_dir = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC" + ) + assert (target_dir / target_image).exists() + + +def test_find_actions_skips_corrupt_profile_and_continues( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + bad_profile = sample_profiles_v3 / "BROKEN.sdProfile" + bad_profile.mkdir() + (bad_profile / "manifest.json").write_text("{not-json") + + page_path = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + page_manifest = json.loads(page_path.read_text()) + page_manifest["Controllers"][0]["Actions"]["0,1"] = { + "ActionID": "alive", + "Name": "Open", + "Plugin": { + "Name": "Open", + "UUID": "com.elgato.streamdeck.system.open", + "Version": "1.0", + }, + "Settings": {}, + "State": 0, + "States": [{"Title": "Alive"}], + "UUID": "com.elgato.streamdeck.system.open", + } + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + result = manager.find_actions(query="alive") + assert result["count"] >= 1 + assert any(hit["title"] == "Alive" for hit in result["actions"]) + + +def test_resolve_layout_tolerates_malformed_action_keys( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + profile_manifest = { + "Device": {"Model": "UNKNOWN-MODEL"}, + "Name": "Weird", + "Version": "3.0", + } + page_manifest = { + "Controllers": [ + { + "Type": "Keypad", + "Actions": { + "0,0": {"UUID": "x"}, + "bad-slot": {"UUID": "y"}, + "3,2": {"UUID": "z"}, + }, + } + ] + } + cols, rows = manager._resolve_layout(profile_manifest, page_manifest, "Keypad") + assert cols == 4 + assert rows == 3 + + +def test_find_actions_skips_corrupt_page_and_continues( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + bad_page = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "AAAAAAAA-AAAA-AAAA-AAAA-AAAAAAAAAAAA" + / "manifest.json" + ) + bad_page.write_text("{not-json") + + page_path = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + page_manifest = json.loads(page_path.read_text()) + page_manifest["Controllers"][0]["Actions"]["1,0"] = { + "ActionID": "keep-me", + "Name": "Open", + "Plugin": { + "Name": "Open", + "UUID": "com.elgato.streamdeck.system.open", + "Version": "1.0", + }, + "Settings": {}, + "State": 0, + "States": [{"Title": "Keep"}], + "UUID": "com.elgato.streamdeck.system.open", + } + # Malformed position key must not abort the scan. + page_manifest["Controllers"][0]["Actions"]["bad-slot"] = { + "ActionID": "broken", + "Name": "Broken", + "Plugin": {"Name": "X", "UUID": "com.example.x", "Version": "1"}, + "Settings": {}, + "State": 0, + "States": [{}], + "UUID": "com.example.x.action", + } + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + result = manager.find_actions(query="keep") + assert result["count"] >= 1 + assert any(hit["title"] == "Keep" for hit in result["actions"]) + + +def test_write_page_mints_unique_action_ids_for_repeated_raw_paste( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + shared_action = { + "ActionID": "shared-id", + "Name": "Open", + "Plugin": { + "Name": "Open", + "UUID": "com.elgato.streamdeck.system.open", + "Version": "1.0", + }, + "Settings": {"path": '"/tmp/a.sh"'}, + "State": 0, + "States": [{"Title": "A"}], + "UUID": "com.elgato.streamdeck.system.open", + } + manager.write_page( + profile_name="Default Profile", + directory_id="CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC", + buttons=[ + {"key": 0, "action": shared_action}, + {"key": 1, "action": shared_action}, + ], + ) + page = manager.read_page( + profile_name="Default Profile", + directory_id="CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC", + ) + ids = [button["action_id"] for button in page["buttons"]] + assert len(ids) == 2 + assert ids[0] != ids[1] + assert "shared-id" not in ids + + +def test_find_actions_finds_encoder_dials(sample_profiles_plus_xl: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_plus_xl, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + result = manager.find_actions(controller="encoder", query="wave") + assert result["count"] == 1 + hit = result["actions"][0] + assert hit["controller"] == "encoder" + assert hit["plugin_uuid"] == "com.elgato.wave-link" + assert hit["settings"] == {"channelId": "mic"} + assert hit["action"]["ActionID"] != "dial-volume" + assert hit["button"]["controller"] == "encoder" + assert hit["button"]["action"]["UUID"] == hit["action"]["UUID"] + + +def test_find_actions_scopes_profile_and_respects_limit( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + page_path = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + page_manifest = json.loads(page_path.read_text()) + actions = page_manifest["Controllers"][0]["Actions"] + for col in range(3): + actions[f"{col},1"] = { + "ActionID": f"extra-{col}", + "Name": "Open", + "Plugin": { + "Name": "Open", + "UUID": "com.elgato.streamdeck.system.open", + "Version": "1.0", + }, + "Settings": {"path": f'"/tmp/{col}.sh"'}, + "State": 0, + "States": [{"Title": f"Extra {col}"}], + "UUID": "com.elgato.streamdeck.system.open", + } + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + scoped = manager.find_actions( + profile_name="Default Profile", + plugin_uuid="com.elgato.streamdeck.system.open", + ) + assert scoped["count"] == 4 # original Deploy + 3 extras + assert all(hit["profile_name"] == "Default Profile" for hit in scoped["actions"]) + + limited = manager.find_actions( + profile_name="Default Profile", + plugin_uuid="com.elgato.streamdeck.system.open", + limit=2, + ) + assert limited["count"] == 2 + assert limited["truncated"] is True + + +def test_find_actions_paste_into_write_page(sample_profiles_v3: Path, tmp_path: Path) -> None: + page_path = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + page_manifest = json.loads(page_path.read_text()) + configured_action = { + "ActionID": "ha-light-action", + "LinkedTitle": False, + "Name": "Entity", + "Plugin": { + "Name": "Home Assistant", + "UUID": "de.perdoctus.streamdeck.homeassistant", + "Version": "3.7.1", + }, + "Settings": {"entityId": "light.office", "mode": "toggle"}, + "State": 0, + "States": [{"Title": "Office"}], + "UUID": "de.perdoctus.streamdeck.homeassistant.entity", + } + page_manifest["Controllers"][0]["Actions"]["2,0"] = configured_action + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + found = manager.find_actions(plugin_name="Home Assistant") + assert found["count"] == 1 + hit = found["actions"][0] + + manager.write_page( + profile_name="Default Profile", + directory_id="CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC", + buttons=[ + { + "key": 3, + "title": "Office Paste", + "action": hit["action"], + } + ], + ) + + target = manager.read_page( + profile_name="Default Profile", + directory_id="CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC", + ) + pasted = target["buttons"][0] + assert pasted["key"] == 3 + assert pasted["plugin_uuid"] == "de.perdoctus.streamdeck.homeassistant" + assert pasted["settings"] == {"entityId": "light.office", "mode": "toggle"} + assert pasted["title"] == "Office Paste" + # write_page mints a fresh ActionID even when reusing a find_actions hit. + assert pasted["action_id"] != "ha-light-action" + assert pasted["action_id"] != hit["action"]["ActionID"] + + +def test_read_page_skips_malformed_action_keys(sample_profiles_v3: Path, tmp_path: Path) -> None: + page_path = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + page_manifest = json.loads(page_path.read_text()) + page_manifest["Controllers"][0]["Actions"]["bad-slot"] = { + "ActionID": "broken", + "Name": "Broken", + "Plugin": {"Name": "X", "UUID": "com.example.x", "Version": "1"}, + "Settings": {}, + "State": 0, + "States": [{}], + "UUID": "com.example.x.action", + } + page_path.write_text(json.dumps(page_manifest)) + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + page = manager.read_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + ) + assert all(button["position"] != "bad-slot" for button in page["buttons"]) + assert any(button["title"] == "Deploy" for button in page["buttons"]) + + +def test_read_page_surfaces_corrupt_manifest_for_direct_lookup( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + bad_page = ( + sample_profiles_v3 + / "PROFILE-ONE.sdProfile" + / "Profiles" + / "BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" + / "manifest.json" + ) + bad_page.write_text("{not-json") + + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + with pytest.raises(ProfileManagerError, match="Invalid JSON|Missing file"): + manager.read_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + ) + + +def test_find_actions_empty_returns_hint(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + result = manager.find_actions(query="definitely-not-a-real-plugin-xyz") + assert result["count"] == 0 + assert result["actions"] == [] + assert "hint" in result + assert "find" in result["hint"].lower() or "configure" in result["hint"].lower() + + def test_write_page_updates_existing_v3_page(sample_profiles_v3: Path, tmp_path: Path) -> None: manager = ProfileManager( profiles_dir=sample_profiles_v3, diff --git a/tests/test_profile_server.py b/tests/test_profile_server.py index 3004e4e..03202d8 100644 --- a/tests/test_profile_server.py +++ b/tests/test_profile_server.py @@ -15,6 +15,7 @@ async def test_profile_server_exposes_read_plugins_tool() -> None: names = {tool.name for tool in tools} assert "streamdeck_read_plugins" in names + assert "streamdeck_find_actions" in names @pytest.mark.asyncio @@ -41,3 +42,87 @@ def list_plugins(self, *, plugin_id=None, include_raw_manifest=False): payload = json.loads(response[0].text) assert calls == {"plugin_id": "com.example.plugin", "include_raw_manifest": True} assert payload["plugins"][0]["plugin_uuid"] == "com.example.plugin" + + +@pytest.mark.asyncio +async def test_profile_server_find_actions_calls_manager(monkeypatch: pytest.MonkeyPatch) -> None: + calls = {} + + class StubManager: + def find_actions( + self, + *, + query=None, + plugin_uuid=None, + action_uuid=None, + plugin_name=None, + controller=None, + profile_name=None, + profile_id=None, + limit=50, + ): + calls.update( + { + "query": query, + "plugin_uuid": plugin_uuid, + "action_uuid": action_uuid, + "plugin_name": plugin_name, + "controller": controller, + "profile_name": profile_name, + "profile_id": profile_id, + "limit": limit, + } + ) + return { + "count": 1, + "truncated": False, + "actions": [{"plugin_uuid": "com.example.plugin", "action": {"ActionID": "x"}}], + } + + monkeypatch.setattr(profile_server, "manager", StubManager()) + + response = await profile_server.call_tool( + "streamdeck_find_actions", + { + "query": "hue", + "controller": "keypad", + "limit": "10", + }, + ) + + payload = json.loads(response[0].text) + assert calls["query"] == "hue" + assert calls["controller"] == "keypad" + assert calls["limit"] == 10 + assert payload["count"] == 1 + assert payload["actions"][0]["plugin_uuid"] == "com.example.plugin" + + +@pytest.mark.asyncio +async def test_profile_server_validation_errors_are_mcp_errors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class StubManager: + def read_page(self, **_kwargs): + raise profile_server.ProfileNotFoundError("Profile not found: ") + + monkeypatch.setattr(profile_server, "manager", StubManager()) + + response = await profile_server.call_tool( + "streamdeck_read_page", + {"directory_id": "missing-page"}, + ) + + assert response.isError is True + assert response.content[0].text == "❌ Profile not found: " + + +@pytest.mark.asyncio +async def test_profile_server_invalid_buttons_payload_is_mcp_error() -> None: + response = await profile_server.call_tool( + "streamdeck_write_page", + {"buttons": "not json"}, + ) + + assert response.isError is True + assert "'buttons' was a string but not a valid JSON array" in response.content[0].text diff --git a/uv.lock b/uv.lock index 0db9842..6b25ed7 100644 --- a/uv.lock +++ b/uv.lock @@ -1007,7 +1007,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "mcp", extras = ["cli"], specifier = ">=1.6.0" }, + { name = "mcp", extras = ["cli"], specifier = ">=1.25.0" }, { name = "pillow", specifier = ">=10.0.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.24.0" },