#!/usr/bin/env python3 """stdio MCP-адаптер к единственной source-debug сессии Sprinter.""" from __future__ import annotations import argparse from functools import wraps import os import sys import threading from typing import Any from mame_profile import add_arguments from sdbg.mcp_adapter import McpSession from sdbg.session import SessionError from sdbg.supervisor import ManagedMcpSession, SessionSupervisor def make_server(client: McpSession | ManagedMcpSession, supervisor: SessionSupervisor | None = None): 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() if supervisor is not None: @tool() def start_session() -> dict[str, Any]: """Асинхронно запустить MAME→DSS→EXE→main из --build; готовность читать через session_status.""" return supervisor.start() @tool() def stop_session() -> dict[str, Any]: """Завершить только MAME, созданный этим MCP-сервером.""" return supervisor.stop() @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 read_program_memory(address: str, length: int = 16) -> dict[str, Any]: """Читать 1..4096 байт raw program space 0..0x3ffff; Z80 окна находятся в 0x10000+.""" return client.read_program_memory(address, length) @tool(annotations=reading) def disassemble_logical(address: str, length: int = 32) -> dict[str, Any]: """Дизассемблировать 1..256 байт текущей logical Z80 памяти при stop; ответ содержит bank_pages.""" return client.disassemble_logical(address, length) @tool(annotations=reading) def list_ports() -> dict[str, Any]: """Перечислить порты ввода MAME и поля с масками; результат ограничен по размеру.""" return client.list_ports() @tool(annotations=reading) def list_shares() -> dict[str, Any]: """Перечислить точные tag и размеры memory shares/regions MAME.""" return client.list_shares() @tool(annotations=reading) def read_share(tag: str, address: str, length: int = 16) -> dict[str, Any]: """Читать 1..4096 байт из share по точному tag; адрес decimal или 0xHEX.""" return client.read_share(tag, address, length) @tool(annotations=reading) def read_vram(address: str, length: int = 16) -> dict[str, Any]: """Читать 1..4096 байт VRAM без Z80 bank mapping.""" return client.read_vram(address, length) @tool(annotations=reading) def read_screen_pixels(x: int, y: int, width: int = 1, height: int = 1) -> dict[str, Any]: """Читать до 8192 видимых pen16 пикселей; при stop кадр может быть старым.""" return client.read_screen_pixels(x, y, width, height) @tool() def screenshot() -> dict[str, Any]: """Сохранить PNG последнего кадра в изолированном каталоге текущей сессии.""" return client.screenshot() @tool() def press_key(key: str, frames: int = 3) -> dict[str, Any]: """Нажать один символ/enter/space/tab на 1..60 кадров; отпустить при stop/ошибке.""" return client.press_key(key, frames) @tool() def type_string(value: str) -> dict[str, Any]: """Ввести 1..64 символа физической PC-клавиатурой; '\\n' означает Enter.""" return client.type_string(value) @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(annotations=reading) def list_breakpoints() -> dict[str, Any]: """Читать C-точки всех владельцев с логическими ID и условиями банка.""" return client.list_breakpoints() @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 claim_control() -> dict[str, Any]: """Захватить управление CPU, если оно свободно; конфликт возвращает текущего владельца.""" return client.claim_control() @tool() def release_control() -> dict[str, Any]: """Освободить своё управление CPU для VS Code или другого MCP-клиента.""" return client.release_control() @tool() def pause_execution() -> dict[str, Any]: """Остановить CPU и вернуть текущую C-позицию.""" return client.pause_execution() @tool() def step_instruction(count: int = 1) -> dict[str, Any]: """Выполнить 1..64 машинных инструкций Z80 и вернуть текущую позицию.""" return client.step_instruction(count) @tool() def step_over_instruction(count: int = 1) -> dict[str, Any]: """Перешагнуть 1..64 машинных инструкций Z80 (вызовы — целиком).""" return client.step_over_instruction(count) @tool() def step_out_instruction() -> dict[str, Any]: """Выйти из текущего машинного frame Z80; C source-step не используется.""" return client.step_out_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 сеанса; с --build можно не указывать') parser.add_argument('--build', help='Debug-пакет для автономного MCP запуска') add_arguments(parser) parser.add_argument('--app-hdd') parser.add_argument('--launch-path') parser.add_argument('--data', action='append', default=[]) parser.add_argument('--debugger', default='sdbg') parser.add_argument('--launch-at', type=float) parser.add_argument('--dss-timeout', type=float) args = parser.parse_args() if not args.socket and not args.build: parser.error('укажите --socket или SDBG_SOCKET') supervisor = None if args.build: options = [] for name in ('mame_home', 'mame_bin', 'mame_rompath', 'mame_dss_image', 'mame_system_hdd_image', 'mame_bios', 'app_hdd', 'launch_path', 'debugger', 'launch_at', 'dss_timeout'): value = getattr(args, name) if value is not None: options += ['--' + name.replace('_', '-'), str(value)] for filename in args.data: options += ['--data', filename] supervisor = SessionSupervisor(args.build, args.socket, options) client = ManagedMcpSession(supervisor) else: client = McpSession(args.socket) heartbeat_stop = threading.Event() heartbeat = None try: if supervisor is None: client.status() # Attach-режим требует уже работающий session server. def renew_control(): while not heartbeat_stop.wait(10): try: client.renew_control() except (SessionError, OSError): # Без lease обновлять нечего; следующая мутация захватит его. pass heartbeat = threading.Thread(target=renew_control, name='sdbg-mcp-heartbeat', daemon=True) heartbeat.start() server = make_server(client, supervisor) server.run(transport='stdio') return 0 except (SessionError, OSError, RuntimeError) as error: print('sdbg-mcp: ' + str(error), file=sys.stderr) return 1 finally: heartbeat_stop.set() if heartbeat is not None: heartbeat.join(timeout=1) try: client.release_control() except (SessionError, OSError): pass try: client.clear_owned_breakpoints() except (SessionError, OSError): pass if supervisor is not None: supervisor.close() if __name__ == '__main__': sys.exit(main())