Александр Петров
268 lines
12 KiB
Python
268 lines
12 KiB
Python
#!/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(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() -> 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 сеанса; с --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())
|