Александр Петров
88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
"""Тонкий MCP-клиент общей source-debug сессии без зависимости от MCP SDK."""
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
import uuid
|
||
|
||
from .server import rpc_call
|
||
from .session import SessionError
|
||
|
||
|
||
class McpSession:
|
||
def __init__(self, socket_path: str, rpc=rpc_call):
|
||
self.socket_path = str(Path(socket_path).expanduser())
|
||
self.owner = 'mcp:' + uuid.uuid4().hex
|
||
self._rpc = rpc
|
||
|
||
def call(self, method: str, arguments: dict | None = None, timeout: float = 10):
|
||
return self._rpc(self.socket_path, method, arguments or {}, timeout=timeout)
|
||
|
||
def status(self):
|
||
return self.call('status')
|
||
|
||
def where(self):
|
||
return self.call('where')
|
||
|
||
def registers(self):
|
||
return self.call('registers')
|
||
|
||
def read_memory(self, address: str, length: int = 16):
|
||
try:
|
||
parsed = int(address, 16 if address.lower().startswith('0x') else 10)
|
||
except ValueError as error:
|
||
raise SessionError('Адрес памяти должен быть десятичным или 0xHEX') from error
|
||
return self.call('read_memory', {'address': parsed, 'length': length})
|
||
|
||
def variables(self):
|
||
return self.call('variables')
|
||
|
||
def read_variable(self, name: str, module: str | None = None):
|
||
return self.call('read_variable', {'name': name, 'module': module})
|
||
|
||
def events(self, after: int = 0, timeout: float = 0):
|
||
if after < 0 or timeout < 0 or timeout > 30:
|
||
raise SessionError('after должен быть неотрицательным, timeout — от 0 до 30 с')
|
||
return self.call('events', {'after': after, 'timeout': timeout,
|
||
'owner': self.owner},
|
||
timeout=max(10, timeout + 2))
|
||
|
||
def console_tail(self, count: int = 40):
|
||
if count < 1 or count > 200:
|
||
raise SessionError('count должен быть от 1 до 200')
|
||
return self.call('mame_console_tail', {'count': count})
|
||
|
||
def break_line(self, file: str, line: int):
|
||
return self.call('break_line', {'file': file, 'line': line, 'owner': self.owner})
|
||
|
||
def break_function(self, name: str):
|
||
return self.call('break_function', {'name': name, 'owner': self.owner})
|
||
|
||
def clear_breakpoint(self, identifier: int):
|
||
return self.call('clear_breakpoint', {'id': identifier, 'owner': self.owner})
|
||
|
||
def clear_owned_breakpoints(self):
|
||
return self.call('clear_owned_breakpoints', {'owner': self.owner})
|
||
|
||
def continue_execution(self):
|
||
return self.call('continue', {'owner': self.owner})
|
||
|
||
def pause_execution(self):
|
||
return self.call('pause', {'owner': self.owner})
|
||
|
||
def step_instruction(self):
|
||
return self.call('step', {'owner': self.owner})
|
||
|
||
def step_source(self, kind: str = 'into'):
|
||
if kind not in ('into', 'over', 'out'):
|
||
raise SessionError('kind должен быть into, over или out')
|
||
return self.call('source_step', {'kind': kind, 'owner': self.owner})
|
||
|
||
def claim_control(self):
|
||
return self.call('claim_control', {'owner': self.owner})
|
||
|
||
def renew_control(self):
|
||
return self.call('renew_control', {'owner': self.owner})
|
||
|
||
def release_control(self):
|
||
return self.call('release_control', {'owner': self.owner})
|