finalize mcp stdio and sse example

Author: Biswajee

.gitignore

@@ -29,7 +29,7 @@ __pycache__/
 *.pyd
 
 # MCP logs
-examples/MCP/*.log
+examples/MCP/**/*.log
 
 # Qt build directory
 __Builds/

examples/MCP/sse/README.md

@@ -1,4 +1,4 @@
-# QTAC MCP Server — SSE
+# QTAC MCP Server - SSE
 
 Persistent server supporting multiple concurrent clients. Each client session gets exclusive
 ownership of a device.
@@ -31,7 +31,7 @@ pip install -r examples/MCP/sse/requirements.txt
 
 ## Configuration
 
-`config.yaml` — all fields optional, defaults shown:
+`config.yaml` - all fields optional, defaults shown:
 
 | Parameter | Default | Description |
 | :-- | :-- | :-- |
@@ -50,14 +50,14 @@ The server must be started before any client connects:
 python examples/MCP/sse/tacdev_mcp_server.py
 ```
 
-### Option 1 — Python client
+### Option 1 - Python client
 
 ```bash
 python examples/MCP/sse/tacdev_mcp_client.py           # first available device
 python examples/MCP/sse/tacdev_mcp_client.py COM41     # specific port
 ```
 
-### Option 2 — Claude CLI
+### Option 2 - Claude CLI
 
 Register once (server must already be running):
 
@@ -71,7 +71,7 @@ Verify:
 claude mcp list
 ```
 
-Then ask Claude naturally — no special syntax needed:
+Then ask Claude naturally - no special syntax needed:
 
 ```
 List connected devices
@@ -100,5 +100,5 @@ python examples/MCP/sse/tacdev_mcp_server.py
 | `TACDev library not found` | Run `build.bat` / `build.sh` first |
 | `Architecture mismatch` | Use Python matching your OS architecture |
 | `No module named 'fastmcp'` | `pip install -r requirements.txt` (requires Python 3.10+) |
-| `cryptography` build failure on ARM64 | Re-run `setup.bat` — downloads OpenSSL automatically. Manual fallback: install [Win64ARMOpenSSL-4_0_1.msi](https://slproweb.com/download/Win64ARMOpenSSL-4_0_1.msi) (full, not Light) and set `OPENSSL_DIR=C:\Program Files\OpenSSL-Win64-ARM` |
+| `cryptography` build failure on ARM64 | Re-run `setup.bat` - downloads OpenSSL automatically. Manual fallback: install [Win64ARMOpenSSL-4_0_1.msi](https://slproweb.com/download/Win64ARMOpenSSL-4_0_1.msi) (full, not Light) and set `OPENSSL_DIR=C:\Program Files\OpenSSL-Win64-ARM` |
 | `get_device_count` returns 0 | Check board is connected. Windows: run `FTDICheck.exe` from `__Builds\x64\Release\bin`. Linux: `sudo cp udev-rules/99-QTAC-USB.rules /etc/udev/rules.d/ && sudo udevadm control --reload` |

examples/MCP/sse/tacdev_mcp_client.py

@@ -299,17 +299,24 @@ async def TACExample(port_name: str | None = None) -> None:
 
             commands = await tac.list_commands(handle)
             if commands:
-                # Extract short command names (e.g. "sedl" from "sedl(30), Secondary EDL")
-                cmd_names = [c.split("(")[0].strip() for c in commands]
-                print(f"\nCommand names: {cmd_names}")
+                cmd_names = [c.split(";")[0].strip() for c in commands]
+                print(f"\nCommands      : {cmd_names}")
 
+            quick_commands = await tac.list_quick_commands(handle)
+            if quick_commands:
+                qc_names = [q.split(";")[1].strip() for q in quick_commands]
+                print(f"Quick commands: {qc_names}")
+
+            if commands:
                 cmd = cmd_names[0]
                 state_before = await tac.get_command_state(handle, cmd)
                 print(f"\nCommand '{cmd}' state before toggle: {state_before}")
                 await tac.send_command(handle, cmd, not state_before)
+                await asyncio.sleep(2)
                 state_after = await tac.get_command_state(handle, cmd)
                 print(f"Command '{cmd}' state after  toggle: {state_after}")
                 await tac.send_command(handle, cmd, state_before)
+                await asyncio.sleep(2)
                 print(f"Command '{cmd}' restored to        : {await tac.get_command_state(handle, cmd)}")
 
             await tac.close_tac_handle(handle)

examples/MCP/sse/tacdev_mcp_server.py

@@ -6,6 +6,7 @@
 import asyncio
 import logging
 import logging.handlers
+import time
 from pathlib import Path
 
 import yaml
@@ -32,8 +33,9 @@ def _setup_logging() -> logging.Logger:
     logger = logging.getLogger("tacdev_mcp")
     logger.setLevel(log_cfg["level"].upper())
 
+    log_path = _CONFIG_PATH.with_name(log_cfg["file"])
     handler = logging.handlers.RotatingFileHandler(
-        log_cfg["file"],
+        log_path,
         maxBytes=log_cfg["max_bytes"],
         backupCount=log_cfg["backup_count"],
     )
@@ -43,6 +45,14 @@ def _setup_logging() -> logging.Logger:
 
 log = _setup_logging()
 
+# Suppress uvicorn SSE shutdown race: starlette sends http.response.start after
+# uvicorn has already torn down the connection on Ctrl+C.
+logging.getLogger("uvicorn.error").addFilter(
+    type("_suppress_sse_shutdown", (logging.Filter,), {
+        "filter": lambda self, r: "Expected ASGI message" not in r.getMessage()
+    })()
+)
+
 # --------------------------------------------------------------------------- #
 # Device ownership registry
 # --------------------------------------------------------------------------- #
@@ -80,6 +90,7 @@ def _release_all() -> None:
         try:
             device.Close()
             log.info("session=%s handle=%d port=%s closed on shutdown", session_id, handle, port)
+            time.sleep(2)
         except Exception as e:
             log.warning("session=%s handle=%d port=%s failed to close on shutdown: %s", session_id, handle, port, e)
     handles.clear()
@@ -594,12 +605,13 @@ async def boot_to_secondary_edl_button(handle: int, ctx: Context) -> dict:
 if __name__ == "__main__":
     server_cfg = cfg["server"]
     log.info("Starting TACDev MCP server on %s:%d", server_cfg["host"], server_cfg["port"])
-    log.info("Tip: if a client crashes and leaves a device locked, press Ctrl+C to restart the server — all devices will be released cleanly on shutdown.")
+    log.info("Tip: if a client crashes and leaves a device locked, press Ctrl+C to restart the server. All devices will be released cleanly on shutdown.")
     try:
         mcp.run(
             transport="sse",
             host=server_cfg["host"],
             port=server_cfg["port"],
+            uvicorn_config={"timeout_graceful_shutdown": 0},
         )
     except KeyboardInterrupt:
         pass

examples/MCP/stdio/README.md

@@ -1,4 +1,4 @@
-# QTAC MCP Server — stdio
+# QTAC MCP Server - stdio
 
 Server is spawned automatically per-client process. No standalone server to manage.
 
@@ -30,16 +30,16 @@ pip install -r examples/MCP/stdio/requirements.txt
 
 ## Running
 
-No separate server process needed — the server is spawned automatically.
+No separate server process needed - the server is spawned automatically.
 
-### Option 1 — Python client
+### Option 1 - Python client
 
 ```bash
 python examples/MCP/stdio/tacdev_mcp_client.py           # first available device
 python examples/MCP/stdio/tacdev_mcp_client.py COM41     # specific port
 ```
 
-### Option 2 — Claude CLI
+### Option 2 - Claude CLI
 
 Register once from repo root:
 
@@ -53,7 +53,7 @@ Verify:
 claude mcp list
 ```
 
-Then ask Claude naturally — no special syntax needed:
+Then ask Claude naturally - no special syntax needed:
 
 ```
 List connected devices
@@ -79,5 +79,5 @@ Boot COM3 to fastboot
 | `TACDev library not found` | Run `build.bat` / `build.sh` first |
 | `Architecture mismatch` | Use Python matching your OS architecture |
 | `No module named 'fastmcp'` | `pip install -r requirements.txt` (requires Python 3.10+) |
-| `cryptography` build failure on ARM64 | Re-run `setup.bat` — downloads OpenSSL automatically. Manual fallback: install [Win64ARMOpenSSL-4_0_1.msi](https://slproweb.com/download/Win64ARMOpenSSL-4_0_1.msi) (full, not Light) and set `OPENSSL_DIR=C:\Program Files\OpenSSL-Win64-ARM` |
+| `cryptography` build failure on ARM64 | Re-run `setup.bat` - downloads OpenSSL automatically. Manual fallback: install [Win64ARMOpenSSL-4_0_1.msi](https://slproweb.com/download/Win64ARMOpenSSL-4_0_1.msi) (full, not Light) and set `OPENSSL_DIR=C:\Program Files\OpenSSL-Win64-ARM` |
 | `get_device_count` returns 0 | Check board is connected. Windows: run `FTDICheck.exe` from `__Builds\x64\Release\bin`. Linux: `sudo cp udev-rules/99-QTAC-USB.rules /etc/udev/rules.d/ && sudo udevadm control --reload` |

examples/MCP/stdio/tacdev_mcp_client.py

@@ -293,17 +293,24 @@ async def TACExample(port_name: str | None = None) -> None:
 
             commands = await tac.list_commands(handle)
             if commands:
-                # Extract short command names (e.g. "sedl" from "sedl(30), Secondary EDL")
-                cmd_names = [c.split("(")[0].strip() for c in commands]
-                print(f"\nCommand names: {cmd_names}")
+                cmd_names = [c.split(";")[0].strip() for c in commands]
+                print(f"\nCommands      : {cmd_names}")
 
+            quick_commands = await tac.list_quick_commands(handle)
+            if quick_commands:
+                qc_names = [q.split(";")[1].strip() for q in quick_commands]
+                print(f"Quick commands: {qc_names}")
+
+            if commands:
                 cmd = cmd_names[0]
                 state_before = await tac.get_command_state(handle, cmd)
                 print(f"\nCommand '{cmd}' state before toggle: {state_before}")
                 await tac.send_command(handle, cmd, not state_before)
+                await asyncio.sleep(2)
                 state_after = await tac.get_command_state(handle, cmd)
                 print(f"Command '{cmd}' state after  toggle: {state_after}")
                 await tac.send_command(handle, cmd, state_before)
+                await asyncio.sleep(2)
                 print(f"Command '{cmd}' restored to        : {await tac.get_command_state(handle, cmd)}")
 
             await tac.close_tac_handle(handle)

examples/MCP/stdio/tacdev_mcp_server.py

@@ -494,5 +494,7 @@ def boot_to_secondary_edl_button(handle: int) -> dict:
 if __name__ == "__main__":
     try:
         mcp.run(transport="stdio")
+    except KeyboardInterrupt:
+        pass
     finally:
         _release_all()