Add shared-session MCP source debugger adapter

toolchain/sdbg_mcp.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""stdio MCP-адаптер к единственной source-debug сессии Sprinter."""
+from __future__ import annotations
+
+import argparse
+from functools import wraps
+import os
+import sys
+from typing import Any
+
+from sdbg.mcp_adapter import McpSession
+from sdbg.session import SessionError
+
+
+def make_server(client: McpSession):
+    try:
+        from mcp.server import MCPServer
+        from mcp.server.mcpserver.exceptions import ToolError
+        from mcp.types import ToolAnnotations
+    except ImportError as error:
+        raise RuntimeError('Нужен официальный MCP SDK 2.x: python -m pip install "mcp>=2.2,<3"') from error
+
+    server = MCPServer('sprinter-source-debug',
+                       instructions='Все команды работают через одну sdbg-сессию. '
+                       'Чтение переменных и памяти требует остановленного CPU. '
+                       'Для длинного шага используйте recent_events или pause_execution.')
+    reading = ToolAnnotations(readOnlyHint=True, openWorldHint=False)
+
+    def tool(*, annotations=None):
+        def register(function):
+            @wraps(function)
+            def guarded(*args, **kwargs):
+                try:
+                    return function(*args, **kwargs)
+                except (SessionError, OSError, ValueError) as error:
+                    raise ToolError(str(error)) from error
+            return server.tool(annotations=annotations)(guarded)
+        return register
+
+    @tool(annotations=reading)
+    def session_status() -> dict[str, Any]:
+        """Получить build ID, generation, состояние CPU и ограничения текущей сессии."""
+        return client.status()
+
+    @tool(annotations=reading)
+    def where() -> dict[str, Any]:
+        """Получить проверенное соответствие PC исходнику C и текущему банку."""
+        return client.where()
+
+    @tool(annotations=reading)
+    def read_registers() -> dict[str, Any]:
+        """Читать регистры Z80 и отображение банков на остановленном CPU."""
+        return client.registers()
+
+    @tool(annotations=reading)
+    def read_memory(address: str, length: int = 16) -> dict[str, Any]:
+        """Читать 1..256 байт logical Z80 memory без side effects; адрес decimal или 0xHEX."""
+        return client.read_memory(address, length)
+
+    @tool(annotations=reading)
+    def list_variables() -> dict[str, Any]:
+        """Перечислить известные global/static переменные и поддержанность типов."""
+        return {'variables': client.variables()}
+
+    @tool(annotations=reading)
+    def read_variable(name: str, module: str | None = None) -> dict[str, Any]:
+        """Читать поддержанную global/static переменную; module разрешает неоднозначность."""
+        return client.read_variable(name, module)
+
+    @tool(annotations=reading)
+    def recent_events(after: int = 0, timeout: float = 0) -> dict[str, Any]:
+        """Читать общий журнал stopped/continued/output с курсором seq; timeout 0..30 с."""
+        return client.events(after, timeout)
+
+    @tool(annotations=reading)
+    def mame_console_tail(count: int = 40) -> dict[str, Any]:
+        """Получить последние строки консоли MAME, если bridge поддерживает её чтение."""
+        return client.console_tail(count)
+
+    @tool()
+    def set_line_breakpoint(file: str, line: int) -> dict[str, Any]:
+        """Поставить личную точку по C-файлу и строке без замены точек VS Code."""
+        return client.break_line(file, line)
+
+    @tool()
+    def set_function_breakpoint(name: str) -> dict[str, Any]:
+        """Поставить личную точку по имени C-функции."""
+        return client.break_function(name)
+
+    @tool()
+    def clear_breakpoint(identifier: int) -> dict[str, Any]:
+        """Удалить свою логическую точку по ID; чужие точки отклоняются."""
+        return client.clear_breakpoint(identifier)
+
+    @tool()
+    def clear_my_breakpoints() -> dict[str, Any]:
+        """Удалить только точки текущего MCP-подключения."""
+        return client.clear_owned_breakpoints()
+
+    @tool()
+    def continue_execution() -> dict[str, Any]:
+        """Продолжить CPU; событие остановки читать через recent_events."""
+        return client.continue_execution()
+
+    @tool()
+    def pause_execution() -> dict[str, Any]:
+        """Остановить CPU и вернуть текущую C-позицию."""
+        return client.pause_execution()
+
+    @tool()
+    def step_instruction() -> dict[str, Any]:
+        """Выполнить одну машинную инструкцию Z80 и вернуть текущую позицию."""
+        return client.step_instruction()
+
+    @tool()
+    def step_source(kind: str = 'into') -> dict[str, Any]:
+        """Начать асинхронный C-шаг: into, over или out; результат в recent_events."""
+        return client.step_source(kind)
+
+    return server
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--socket', default=os.environ.get('SDBG_SOCKET'),
+                        help='Unix socket уже запущенного sdbg_server.py (или SDBG_SOCKET)')
+    args = parser.parse_args()
+    if not args.socket:
+        parser.error('укажите --socket или SDBG_SOCKET')
+    client = McpSession(args.socket)
+    try:
+        client.status()  # Не публикуем инструменты для отсутствующей сессии.
+        server = make_server(client)
+        server.run(transport='stdio')
+        return 0
+    except (SessionError, OSError, RuntimeError) as error:
+        print('sdbg-mcp: ' + str(error), file=sys.stderr)
+        return 1
+    finally:
+        try:
+            client.clear_owned_breakpoints()
+        except (SessionError, OSError):
+            pass
+
+
+if __name__ == '__main__':
+    sys.exit(main())