# -*- coding: utf-8 -*- """transpyc_mcp.server - TransPyC/TransPyV MCP 服务器 实现 MCP/ 协议（v2026-08）： - stdio transport，newline-delimited JSON-RPC 2.0 - 14 个工具（tpc_/tpv_/tp_），schema 从 MCP/tools/tools.json 加载 - 安全：路径白名单 / 命令白名单 / 工程串行锁 / 内存保护 - 输出：统一 {ok, exit_code, stdout, stderr, artifacts, elapsed_ms}，日志末尾 50 行 启动：python -m transpyc_mcp.server（cwd = 项目根） """ from __future__ import annotations import ctypes import hashlib import json import os import shlex import subprocess import sys import time from typing import Any, Callable # --------------------------------------------------------------------------- # 常量 # --------------------------------------------------------------------------- def _find_root() -> str: """向上定位项目根（含 Projectrans.py 的目录），兼容服务器位于 MCP/transpyc_mcp 下。""" d: str = os.path.dirname(os.path.abspath(__file__)) while True: if os.path.isfile(os.path.join(d, "Projectrans.py")): return d parent: str = os.path.dirname(d) if parent == d: return d d = parent _ROOT: str = _find_root() _TOOLS_JSON: str = os.path.join(_ROOT, 'MCP', 'tools', 'tools.json') _COMPILE_TIMEOUT: int = 300 _RUN_TIMEOUT: int = 30 _REGRESSION_TIMEOUT: int = 1800 _LOG_TAIL: int = 50 _PROTOCOL_VERSION: str = '2024-11-05' _SERVER_INFO: dict[str, str] = {'name': 'transpyc-tools', 'version': '1.0.0'} _TPV_PROJECT: str = os.path.join('TransPyV', 'project.json') _TPV_EXE: str = os.path.join('TransPyV', 'output', 'TransPyV.exe') # --------------------------------------------------------------------------- # 安全：路径白名单 # --------------------------------------------------------------------------- class _PathError(ValueError): pass def _resolve(path: str) -> str: """规范化并校验路径必须位于项目根内（拒绝越界）。""" p: str = path if not os.path.isabs(p): p = os.path.join(_ROOT, p) p = os.path.normpath(p) if not (p == _ROOT or p.startswith(_ROOT + os.sep)): raise _PathError(f'path outside root: {path}') return p # --------------------------------------------------------------------------- # 命令执行与输出截断 # --------------------------------------------------------------------------- def _tail(text: str, n: int = _LOG_TAIL) -> str: lines: list[str] = text.splitlines() if len(lines) <= n: return text return '\n'.join(lines[-n:]) def _run(cmd: list[str], timeout: int = _COMPILE_TIMEOUT, cwd: str | None = None) -> dict[str, Any]: """执行白名单命令，返回统一结果结构。""" t0: float = time.perf_counter() try: r: subprocess.CompletedProcess = subprocess.run( cmd, cwd=cwd or _ROOT, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=timeout, ) return { 'ok': r.returncode == 0, 'exit_code': r.returncode, 'stdout': _tail(r.stdout), 'stderr': _tail(r.stderr), 'elapsed_ms': int((time.perf_counter() - t0) * 1000), } except subprocess.TimeoutExpired: return {'ok': False, 'exit_code': 6, 'stdout': '', 'stderr': f'timeout after {timeout}s', 'elapsed_ms': timeout * 1000} except Exception as e: # 捕获启动失败等 return {'ok': False, 'exit_code': 1, 'stdout': '', 'stderr': str(e), 'elapsed_ms': int((time.perf_counter() - t0) * 1000)} def _result_ok(**kw: Any) -> dict[str, Any]: base: dict[str, Any] = {'ok': True, 'exit_code': 0, 'stdout': '', 'stderr': '', 'elapsed_ms': 0} base.update(kw) return base def _result_err(code: int, msg: str) -> dict[str, Any]: return {'ok': False, 'exit_code': code, 'stdout': '', 'stderr': msg, 'elapsed_ms': 0} # --------------------------------------------------------------------------- # 工程串行锁 # --------------------------------------------------------------------------- _BUSY: set[str] = set() def _with_lock(key: str, fn: Callable[[], dict[str, Any]]) -> dict[str, Any]: if key in _BUSY: return _result_err(5, f'project busy: {key}') _BUSY.add(key) try: return fn() finally: _BUSY.discard(key) # --------------------------------------------------------------------------- # 内存保护（TPV 需 >= 1 GiB） # --------------------------------------------------------------------------- def _available_gb() -> float: try: class _M(ctypes.Structure): _fields_ = [('length', ctypes.c_ulong), ('memory_load', ctypes.c_ulong), ('total_phys', ctypes.c_ulonglong), ('avail_phys', ctypes.c_ulonglong), ('total_page', ctypes.c_ulonglong), ('avail_page', ctypes.c_ulonglong), ('total_virtual', ctypes.c_ulonglong), ('avail_virtual', ctypes.c_ulonglong), ('avail_ext', ctypes.c_ulonglong)] m: _M = _M() m.length = ctypes.sizeof(_M) ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(m)) return m.avail_phys / (1024.0 ** 3) except Exception: return 99.0 def _check_memory() -> str | None: if _available_gb() < 1.0: return 'insufficient memory (< 1 GiB) for TPV' return None # --------------------------------------------------------------------------- # 工具：参数构造辅助 # --------------------------------------------------------------------------- def _build_project_cmd(args: dict[str, Any], project: str, exe: str | None = None) -> list[str]: """构造工程编译命令（exe=None 用 TPC Projectrans.py；否则用 TPV exe）。""" cmd: list[str] = [sys.executable, 'Projectrans.py'] if exe is None else [exe] cmd += ['--project', project] if args.get('clean'): cmd.append('--clean') if args.get('run'): cmd.append('--run') phase: str = args.get('phase', 'all') if phase in ('1', '2'): cmd += ['--phase', phase] if args.get('rebuild_includes'): cmd.append('--rebuild-includes') if args.get('clear_cache'): cmd.append('--clear-cache') if args.get('asan', True): cmd.append('--asan') if args.get('nowarn_register'): cmd.append('--nowarn-register') for opt in ('src', 'temp', 'output', 'cc'): if args.get(opt): cmd += [f'--{opt}', str(args[opt])] return cmd def _find_exe(project: str) -> dict[str, str]: """从 project.json 的 linker.output 定位产物 exe（相对根）。""" proj_path: str = _resolve(project) proj_dir: str = os.path.dirname(proj_path) try: with open(proj_path, 'r', encoding='utf-8') as f: cfg: dict[str, Any] = json.load(f) except Exception: return {} out_name: str = str(cfg.get('linker', {}).get('output', '')) out_dir: str = str(cfg.get('output_dir', 'output')) if out_name: exe: str = os.path.join(proj_dir, out_dir, out_name) if os.path.isfile(exe): return {'exe': os.path.relpath(exe, _ROOT)} # 兜底：output 目录下最新 .exe out_abs: str = os.path.join(proj_dir, out_dir) if os.path.isdir(out_abs): cands: list[str] = [os.path.join(out_abs, f) for f in os.listdir(out_abs) if f.lower().endswith('.exe')] if cands: latest: str = max(cands, key=os.path.getmtime) return {'exe': os.path.relpath(latest, _ROOT)} return {} # --------------------------------------------------------------------------- # 工具：tpc_ 系列 # --------------------------------------------------------------------------- def _tpc_build(args: dict[str, Any]) -> dict[str, Any]: project: str = str(args.get('project', '')) if not project: return _result_err(2, 'missing required argument: project') proj_path: str = _resolve(project) if not os.path.isfile(proj_path): return _result_err(4, f'project not found: {project}') def run() -> dict[str, Any]: cmd: list[str] = _build_project_cmd(args, project) res: dict[str, Any] = _run(cmd) if res['ok']: res['artifacts'] = _find_exe(project) return res return _with_lock(project, run) def _tpc_build_file(args: dict[str, Any]) -> dict[str, Any]: input_f: str = str(args.get('input', '')) output_f: str = str(args.get('output', '')) if not input_f or not output_f: return _result_err(2, 'missing required arguments: input, output') in_path: str = _resolve(input_f) out_path: str = _resolve(output_f) if not os.path.isfile(in_path): return _result_err(4, f'input not found: {input_f}') cmd: list[str] = [sys.executable, 'TransPyC.py', '-f', in_path, '-o', out_path] if args.get('cc'): cmd += ['-cc', str(args['cc'])] if args.get('cflags'): cmd += ['-cflags', str(args['cflags'])] if args.get('run'): cmd.append('-run') if args.get('args'): cmd += ['-args', str(args['args'])] if args.get('level'): cmd += ['-level', str(args['level'])] cmd += ['-e', str(args.get('encoding', 'utf-8'))] res: dict[str, Any] = _run(cmd) if res['ok']: res['artifacts'] = {'ir': output_f} return res def _tpc_run_test(args: dict[str, Any]) -> dict[str, Any]: name: str = str(args.get('name', '')) if not name: return _result_err(2, 'missing required argument: name') project: str = os.path.join('Test', name, 'project.json') proj_path: str = _resolve(project) if not os.path.isfile(proj_path): return _result_err(4, f'test project not found: {project}') cargs: dict[str, Any] = dict(args) cargs['clean'] = True cargs['run'] = True def run() -> dict[str, Any]: cmd: list[str] = _build_project_cmd(cargs, project) res: dict[str, Any] = _run(cmd, timeout=_RUN_TIMEOUT + _COMPILE_TIMEOUT) if res['ok']: res['artifacts'] = _find_exe(project) return res return _with_lock(project, run) def _tpc_run_regression(args: dict[str, Any]) -> dict[str, Any]: timeout: int = int(args.get('timeout', _REGRESSION_TIMEOUT)) cmd: list[str] = [sys.executable, os.path.join('Test', 'regression_test.py')] return _run(cmd, timeout=timeout) def _tpc_flag_only(args: dict[str, Any], flag: str) -> dict[str, Any]: project: str = str(args.get('project', '')) if not project: return _result_err(2, 'missing required argument: project') proj_path: str = _resolve(project) if not os.path.isfile(proj_path): return _result_err(4, f'project not found: {project}') def run() -> dict[str, Any]: return _run([sys.executable, 'Projectrans.py', '--project', project, flag]) return _with_lock(project, run) def _tpc_clean_project(args: dict[str, Any]) -> dict[str, Any]: return _tpc_flag_only(args, '--clean') def _tpc_clear_cache(args: dict[str, Any]) -> dict[str, Any]: return _tpc_flag_only(args, '--clear-cache') def _tpc_rebuild_includes(args: dict[str, Any]) -> dict[str, Any]: return _tpc_flag_only(args, '--rebuild-includes') # --------------------------------------------------------------------------- # 工具：tpv_ 系列 # --------------------------------------------------------------------------- def _tpv_build(args: dict[str, Any]) -> dict[str, Any]: proj_path: str = _resolve(_TPV_PROJECT) if not os.path.isfile(proj_path): return _result_err(4, f'project not found: {_TPV_PROJECT}') if args.get('clear_cache'): return _result_err(2, 'clear_cache unsupported on tpv_build; use tpc_clear_cache on TPC side first') def run() -> dict[str, Any]: cmd: list[str] = [sys.executable, 'Projectrans.py', '--project', _TPV_PROJECT, '--clean'] if args.get('asan', True): cmd.append('--asan') if args.get('rebuild_includes'): cmd.append('--rebuild-includes') res: dict[str, Any] = _run(cmd) if res['ok']: exe: str = os.path.join(_ROOT, _TPV_EXE) res['artifacts'] = {'exe': _TPV_EXE if os.path.isfile(exe) else ''} return res return _with_lock(_TPV_PROJECT, run) def _sha1_file(path: str) -> str: h: hashlib._Hash = hashlib.sha1() with open(path, 'rb') as f: for chunk in iter(lambda: f.read(65536), b''): h.update(chunk) return h.hexdigest() def _tpv_bootstrap(args: dict[str, Any]) -> dict[str, Any]: mem_err: str | None = _check_memory() if mem_err: return _result_err(7, mem_err) phases: list[str] = list(args.get('phases') or ['A', 'B', 'C']) compare: bool = bool(args.get('compare', True)) out_dirs: dict[str, str] = {'A': 'TransPyV/output', 'B': 'TransPyV/output_b', 'C': 'TransPyV/output_c'} steps: list[dict[str, Any]] = [] # Phase A：TPC 编译 TransPyV → output/TransPyV.exe if 'A' in phases: res: dict[str, Any] = _run([sys.executable, 'Projectrans.py', '--project', _TPV_PROJECT, '--clean', '--asan']) steps.append({'phase': 'A', 'exe': _TPV_EXE, **res}) if not res['ok']: return {'ok': False, 'exit_code': res['exit_code'], 'stdout': res['stdout'], 'stderr': 'bootstrap Phase A failed\n' + res['stderr'], 'artifacts': {'steps': steps}, 'elapsed_ms': res['elapsed_ms']} else: steps.append({'phase': 'A', 'exe': _TPV_EXE, 'ok': True, 'exit_code': 0, 'stdout': '', 'stderr': '(skipped)'}) # Phase B/C：上一阶段 exe 编译自身到下一 output 目录 for idx, ph in enumerate(('B', 'C')): if ph not in phases: steps.append({'phase': ph, 'ok': True, 'exit_code': 0, 'stdout': '', 'stderr': '(skipped)'}) continue prev_dir: str = out_dirs['A'] if idx == 0 else out_dirs['B'] prev_exe: str = os.path.join(_ROOT, prev_dir, 'TransPyV.exe') if not os.path.isfile(prev_exe): steps.append({'phase': ph, 'ok': False, 'exit_code': 4, 'stdout': '', 'stderr': f'previous exe missing: {prev_exe}'}) return {'ok': False, 'exit_code': 4, 'stdout': '', 'stderr': f'bootstrap Phase {ph} missing input', 'artifacts': {'steps': steps}, 'elapsed_ms': 0} out_dir: str = out_dirs[ph] r: dict[str, Any] = _run([prev_exe, '--project', _TPV_PROJECT, '--clean', '--output', out_dir]) steps.append({'phase': ph, 'exe': os.path.join(out_dir, 'TransPyV.exe'), **r}) if not r['ok']: return {'ok': False, 'exit_code': r['exit_code'], 'stdout': r['stdout'], 'stderr': f'bootstrap Phase {ph} failed\n' + r['stderr'], 'artifacts': {'steps': steps}, 'elapsed_ms': r['elapsed_ms']} # 对比各阶段产物 SHA1（自举一致性） artifacts: dict[str, Any] = {'steps': steps} if compare: hashes: dict[str, str] = {} for ph in phases: exe_rel: str = os.path.join(out_dirs[ph], 'TransPyV.exe') exe_abs: str = os.path.join(_ROOT, exe_rel) hashes[ph] = _sha1_file(exe_abs) if os.path.isfile(exe_abs) else '' artifacts['hashes'] = hashes artifacts['identical'] = len(set(hashes.values()) - {''}) <= 1 return _result_ok(stdout=json.dumps(artifacts, ensure_ascii=False, indent=2), artifacts=artifacts) def _tpv_compile_project(args: dict[str, Any]) -> dict[str, Any]: project: str = str(args.get('project', '')) if not project: return _result_err(2, 'missing required argument: project') proj_path: str = _resolve(project) if not os.path.isfile(proj_path): return _result_err(4, f'project not found: {project}') exe: str = os.path.join(_ROOT, _TPV_EXE) if not os.path.isfile(exe): return _result_err(4, f'TPV not built yet; run tpv_build first: {_TPV_EXE}') mem_err: str | None = _check_memory() if mem_err: return _result_err(7, mem_err) def run() -> dict[str, Any]: cmd: list[str] = _build_project_cmd(args, project, exe=exe) res: dict[str, Any] = _run(cmd) if res['ok']: res['artifacts'] = _find_exe(project) return res return _with_lock(project, run) def _tpv_run_test(args: dict[str, Any]) -> dict[str, Any]: exe: str = os.path.join(_ROOT, _TPV_EXE) if not os.path.isfile(exe): return _result_err(4, f'TPV not built yet; run tpv_build first: {_TPV_EXE}') mem_err: str | None = _check_memory() if mem_err: return _result_err(7, mem_err) suite: str = str(args.get('suite', 'all')) proj: str = os.path.join('TransPyV', 'Test', 'project.vpj') if not os.path.isfile(_resolve(proj)): return _result_err(4, f'TPV test project not found: {proj}') cmd: list[str] = [exe, '--project', proj, '--clean', '--run'] return _with_lock(proj, lambda: _run(cmd)) # --------------------------------------------------------------------------- # 工具：tp_ 系列（只读诊断） # --------------------------------------------------------------------------- def _tp_read_ast(args: dict[str, Any]) -> dict[str, Any]: file: str = str(args.get('file', '')) if not file: return _result_err(2, 'missing required argument: file') fpath: str = _resolve(file) if not os.path.isfile(fpath): return _result_err(4, f'file not found: {file}') if fpath.lower().endswith('.py') or fpath.lower().endswith('.vp'): lib: str = os.path.join(_ROOT, 'lib') if lib not in sys.path: sys.path.insert(0, lib) try: import vast as ast # noqa: PLC0415 except Exception as e: return _result_err(1, f'vast import failed: {e}') with open(fpath, 'r', encoding='utf-8', errors='replace') as f: src: str = f.read() try: tree: Any = ast.parse(src) dump: str = args.get('dump', 'compact') text: str = ast.dump(tree, indent=2) if dump == 'full' else ast.dump(tree) except Exception as e: return _result_err(1, f'parse failed: {e}') return _result_ok(stdout=text) return _result_err(2, f'unsupported file type: {file} (only .vp/.py)') def _tp_read_log(args: dict[str, Any]) -> dict[str, Any]: file: str = str(args.get('file', '')) if not file: return _result_err(2, 'missing required argument: file') fpath: str = _resolve(file) if not os.path.isfile(fpath): return _result_err(4, f'file not found: {file}') lines: int = int(args.get('lines', _LOG_TAIL)) try: with open(fpath, 'r', encoding='utf-8', errors='replace') as f: content: str = f.read() except Exception as e: return _result_err(1, f'read failed: {e}') pattern: str = str(args.get('pattern', '')) if pattern: import re # noqa: PLC0415 try: content = '\n'.join(l for l in content.splitlines() if re.search(pattern, l)) except Exception as e: return _result_err(2, f'invalid pattern: {e}') return _result_ok(stdout=_tail(content, max(1, lines))) def _tp_sha1_map(args: dict[str, Any]) -> dict[str, Any]: project: str = str(args.get('project', '')) if not project: return _result_err(2, 'missing required argument: project') map_rel: str = os.path.join(project, 'temp', '_sha1_map.txt') map_path: str = _resolve(map_rel) if not os.path.isfile(map_path): return _result_err(4, f'sha1 map not found: {map_rel}') with open(map_path, 'r', encoding='utf-8', errors='replace') as f: content: str = f.read() filt: str = str(args.get('filter', '')) if filt: content = '\n'.join(l for l in content.splitlines() if filt in l) return _result_ok(stdout=_tail(content, 200)) # --------------------------------------------------------------------------- # 工具分派表 # --------------------------------------------------------------------------- DISPATCH: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { 'tpc_build': _tpc_build, 'tpc_build_file': _tpc_build_file, 'tpc_run_test': _tpc_run_test, 'tpc_run_regression': _tpc_run_regression, 'tpc_clean_project': _tpc_clean_project, 'tpc_clear_cache': _tpc_clear_cache, 'tpc_rebuild_includes': _tpc_rebuild_includes, 'tpv_build': _tpv_build, 'tpv_bootstrap': _tpv_bootstrap, 'tpv_compile_project': _tpv_compile_project, 'tpv_run_test': _tpv_run_test, 'tp_read_ast': _tp_read_ast, 'tp_read_log': _tp_read_log, 'tp_sha1_map': _tp_sha1_map, } def _load_tools() -> list[dict[str, Any]]: try: with open(_TOOLS_JSON, 'r', encoding='utf-8') as f: tools: list[dict[str, Any]] = json.load(f) if isinstance(tools, list): return tools except Exception: pass # 兜底：从分派表生成最小 schema return [{'name': n, 'description': n, 'inputSchema': {'type': 'object', 'properties': {}, 'required': []}} for n in DISPATCH] # --------------------------------------------------------------------------- # JSON-RPC 处理 # --------------------------------------------------------------------------- def _rpc_result(msg: dict[str, Any], result: Any) -> dict[str, Any]: return {'jsonrpc': '2.0', 'id': msg.get('id'), 'result': result} def _rpc_error(msg: dict[str, Any], code: int, message: str) -> dict[str, Any]: return {'jsonrpc': '2.0', 'id': msg.get('id'), 'error': {'code': code, 'message': message}} def _handle(msg: dict[str, Any]) -> dict[str, Any] | None: method: str = str(msg.get('method', '')) if method == 'initialize': return _rpc_result(msg, { 'protocolVersion': _PROTOCOL_VERSION, 'capabilities': {'tools': {}}, 'serverInfo': _SERVER_INFO, }) if method in ('notifications/initialized', 'notifications/cancelled'): return None if method == 'tools/list': return _rpc_result(msg, {'tools': _load_tools()}) if method == 'tools/call': params: dict[str, Any] = msg.get('params') or {} name: str = str(params.get('name', '')) fn: Callable[[dict[str, Any]], dict[str, Any]] | None = DISPATCH.get(name) if fn is None: return _rpc_error(msg, -32602, f'unknown tool: {name}') args: dict[str, Any] = params.get('arguments') or {} if not isinstance(args, dict): return _rpc_error(msg, -32602, 'arguments must be an object') try: result: dict[str, Any] = fn(args) except _PathError as e: result = _result_err(3, str(e)) except Exception as e: # 工具内部未预期异常 result = _result_err(1, f'{type(e).__name__}: {e}') text: str = json.dumps(result, ensure_ascii=False, indent=2) return _rpc_result(msg, {'content': [{'type': 'text', 'text': text}], 'isError': not result.get('ok', True)}) if msg.get('id') is None: return None # 其它 notification return _rpc_error(msg, -32601, f'method not found: {method}') def main() -> None: """stdio 循环：逐行读 JSON-RPC 请求，逐行写响应。""" for line in sys.stdin: line = line.strip() if not line: continue try: msg: dict[str, Any] = json.loads(line) except Exception: continue resp: dict[str, Any] | None = _handle(msg) if resp is not None: sys.stdout.write(json.dumps(resp, ensure_ascii=False) + '\n') sys.stdout.flush() if __name__ == '__main__': main()