Sprinter-SDCC/toolchain/sdbg_mcp.py

#!/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())