Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,20 @@ uvx --from streamdeck-mcp streamdeck-mcp-usb
- Generated icons are stored in `~/.streamdeck-mcp/generated-icons/`.
- Generated shell scripts are stored in `~/StreamDeckScripts/`.

## Editing Workflow (Important)

The Elgato desktop app keeps every profile in memory and rewrites the on-disk manifests from that snapshot when it quits, so any edit made while the app is running is wiped the next time it closes. The profile writer enforces a quit → write → relaunch cycle:

1. Ensure the Elgato app is not running, or pass `auto_quit_app: true` to `streamdeck_write_page` to have it quit the app for you (AppleScript first, `killall` fallback).
2. Make as many `streamdeck_write_page` calls as you need — the app stays quit across them.
3. Call `streamdeck_restart_app` when you are done. The device re-reads the manifests on launch and your changes appear.

`streamdeck_write_page` raises a `StreamDeckAppRunningError` when the app is running and `auto_quit_app` is not set, so you cannot accidentally write changes that will be silently discarded.

If your Elgato app is installed somewhere other than `/Applications/Elgato Stream Deck.app`, set `STREAMDECK_APP_PATH` to the bundle path.

## Usage Notes

- After writing profiles, the Elgato desktop app may need a restart to pick up the new manifests.
- `streamdeck_create_action` is the safest way to build shell-command buttons because it writes a standalone script and returns the native Open action block for it.
- The profile writer does not require exclusive USB access.

Expand Down
153 changes: 130 additions & 23 deletions profile_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import string
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
Expand Down Expand Up @@ -70,6 +71,13 @@
"UI Stream Deck": {KEYPAD: (4, 2)},
}

# The Elgato Stream Deck desktop app caches every profile in memory and rewrites the
# on-disk manifests when it quits, so any edit made while it is running gets clobbered
# the next time the user closes or restarts the app.
STREAM_DECK_APP_PROCESS_NAMES = ("Stream Deck", "Elgato Stream Deck")
DEFAULT_STREAM_DECK_APP_PATH = Path("/Applications/Elgato Stream Deck.app")
STREAM_DECK_APP_PATH_ENV = "STREAMDECK_APP_PATH"

HEX_COLOR_PATTERN = re.compile(r"^#[0-9a-fA-F]{6}$")
POSITION_PATTERN = re.compile(r"^\d+,\d+$")
UUID_PATTERN = re.compile(
Expand Down Expand Up @@ -102,6 +110,16 @@ class ProfileValidationError(ProfileManagerError):
"""Raised when inputs for profile operations are invalid."""


class StreamDeckAppRunningError(ProfileManagerError):
"""Raised when a write is attempted while the Elgato desktop app is running.

The app rewrites every profile manifest from its in-memory snapshot on quit, so
writes made while it is running are silently discarded. Callers must quit the
app first (pass `auto_quit_app=True` to `write_page`) and then call
`restart_app` once their edits are complete to see the changes.
"""


@dataclass
class PageRef:
"""Resolved page directory metadata."""
Expand Down Expand Up @@ -277,6 +295,79 @@ def _count_icons(page_dir: Path) -> int:
return len([path for path in images_dir.iterdir() if path.is_file()])


def _resolve_app_path() -> Path:
override = os.environ.get(STREAM_DECK_APP_PATH_ENV)
if override:
return Path(override).expanduser()
return DEFAULT_STREAM_DECK_APP_PATH


def is_stream_deck_app_running() -> bool:
"""Return True if the Elgato Stream Deck desktop app is currently running."""

if sys.platform != "darwin":
return False

for name in STREAM_DECK_APP_PROCESS_NAMES:
result = subprocess.run(
["pgrep", "-x", name],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0 and result.stdout.strip():
return True
return False


def stop_stream_deck_app(*, graceful_timeout: float = 3.0) -> dict[str, Any]:
"""Quit the Elgato Stream Deck desktop app.

Tries an AppleScript quit first so the app can persist any unrelated state, then
falls back to `killall` if it does not exit in time. Returns a small report about
which path was taken.
"""

if sys.platform != "darwin":
return {"stopped": False, "graceful": [], "forced": [], "reason": "non-darwin platform"}

if not is_stream_deck_app_running():
return {"stopped": False, "graceful": [], "forced": [], "reason": "not running"}

graceful: list[str] = []
for name in STREAM_DECK_APP_PROCESS_NAMES:
result = subprocess.run(
["osascript", "-e", f'tell application "{name}" to quit'],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
graceful.append(name)

deadline = time.monotonic() + graceful_timeout
while time.monotonic() < deadline and is_stream_deck_app_running():
time.sleep(0.2)

forced: list[str] = []
if is_stream_deck_app_running():
for name in STREAM_DECK_APP_PROCESS_NAMES:
result = subprocess.run(
["killall", name],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
forced.append(name)

return {
"stopped": not is_stream_deck_app_running(),
"graceful": graceful,
"forced": forced,
}


class ProfileManager:
"""Read and write Elgato Stream Deck profiles."""

Expand Down Expand Up @@ -415,9 +506,30 @@ def write_page(
clear_existing: bool = True,
create_new: bool = False,
make_current: bool = False,
auto_quit_app: bool = False,
) -> dict[str, Any]:
"""Create a page or rewrite an existing page manifest."""

app_stop_report: dict[str, Any] | None = None
if is_stream_deck_app_running():
if not auto_quit_app:
raise StreamDeckAppRunningError(
"The Elgato Stream Deck app is running and will overwrite this "
"edit on quit. Retry with auto_quit_app=True to quit it first, "
"then call streamdeck_restart_app once your edits are complete "
"to apply the changes."
)
app_stop_report = stop_stream_deck_app()
Comment thread
jack-arturo marked this conversation as resolved.
stop_failed = not app_stop_report.get("stopped", False)
still_running = is_stream_deck_app_running()
if stop_failed or still_running:
reason = app_stop_report.get("reason", "")
detail = f" Reason: {reason}." if reason else ""
raise StreamDeckAppRunningError(
f"The Elgato Stream Deck app could not be stopped.{detail} Aborting "
"page write because the running app may overwrite these edits on quit."
)

profile_dir, profile_manifest = self._resolve_profile(
profile_name=profile_name, profile_id=profile_id
)
Expand Down Expand Up @@ -518,6 +630,7 @@ def write_page(
"button_count": total_button_count,
"page_name": page_manifest.get("Name", ""),
"manifest_path": str(page_dir / "manifest.json"),
"app_quit": app_stop_report,
}

def create_icon(
Expand Down Expand Up @@ -599,40 +712,34 @@ def restart_app(self) -> dict[str, Any]:
"streamdeck_restart_app is currently only supported on macOS."
)

killed = []
for app_name in ("Elgato Stream Deck", "Stream Deck"):
result = subprocess.run(
["killall", app_name],
capture_output=True,
text=True,
check=False,
app_path = _resolve_app_path()
if not app_path.exists():
raise ProfileManagerError(
f"Stream Deck app not found at {app_path}. "
f"Set {STREAM_DECK_APP_PATH_ENV} to override the default install path."
)
if result.returncode == 0:
killed.append(app_name)

open_result = subprocess.run(
["open", "-a", "Stream Deck"],
stop_report = stop_stream_deck_app()

# `open -a <name>` relies on LaunchServices name lookup, which returns error
# -600 on some systems even when the bundle is present. Launching by explicit
# path bypasses that lookup.
result = subprocess.run(
["open", str(app_path)],
capture_output=True,
text=True,
check=False,
)
if open_result.returncode != 0:
fallback = subprocess.run(
["open", "-a", "Elgato Stream Deck"],
capture_output=True,
text=True,
check=False,
)
open_result = fallback

if open_result.returncode != 0:
if result.returncode != 0:
message = result.stderr.strip() or result.stdout.strip()
raise ProfileManagerError(
open_result.stderr.strip() or "Failed to relaunch Stream Deck app."
f"Failed to relaunch Stream Deck ({app_path}): {message or 'unknown error'}"
)

return {
"killed": killed,
"restarted": True,
"app_path": str(app_path),
"stop": stop_report,
}

def _resolve_profile(
Expand Down
21 changes: 20 additions & 1 deletion profile_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
ProfileManagerError,
ProfileNotFoundError,
ProfileValidationError,
StreamDeckAppRunningError,
)

logging.basicConfig(
Expand Down Expand Up @@ -189,7 +190,12 @@ async def list_tools() -> list[Tool]:
Tool(
name="streamdeck_write_page",
description=(
"Create a new page or replace/update an existing Stream Deck desktop page manifest."
"Create a new page or replace/update an existing Stream Deck desktop "
"page manifest. IMPORTANT: the Elgato desktop app overwrites profile "
"manifests from its in-memory state on quit, so writes made while the "
"app is running are lost. This tool refuses to write when the app is "
"running unless auto_quit_app=True is passed. Call streamdeck_restart_app "
"once your edits are complete to make the changes visible on the device."
),
inputSchema={
"type": "object",
Expand Down Expand Up @@ -227,6 +233,16 @@ async def list_tools() -> list[Tool]:
"When true, make the page the active current page after writing."
),
},
"auto_quit_app": {
"type": "boolean",
"description": (
"If true and the Elgato Stream Deck desktop app is "
"running, quit it (graceful AppleScript first, then "
"killall) before writing. Required when the app is "
"running or the write will raise an error. Defaults to "
"false so callers must explicitly consent to quitting it."
),
},
},
},
),
Expand Down Expand Up @@ -322,6 +338,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
clear_existing=arguments.get("clear_existing", True),
create_new=arguments.get("create_new", False),
make_current=arguments.get("make_current", False),
auto_quit_app=arguments.get("auto_quit_app", False),
)
return [TextContent(type="text", text=json.dumps(result, indent=2))]

Expand Down Expand Up @@ -350,6 +367,8 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:

return [TextContent(type="text", text=f"❌ Unknown tool: {name}")]

except StreamDeckAppRunningError as exc:
return [TextContent(type="text", text=f"⚠️ {exc}")]
except (
ProfileManagerError,
ProfileNotFoundError,
Expand Down
Loading