From 004121de409089fd78e6e694232e01f1a4e71044 Mon Sep 17 00:00:00 2001 From: TermiNexus Date: Fri, 21 Aug 2026 14:05:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20=E5=AE=9E=E7=8E=B0=20transpyc=5Fmc?= =?UTF-8?q?p=20=E6=9C=8D=E5=8A=A1=E5=99=A8=E2=80=94=E2=80=94stdio=20JSON-R?= =?UTF-8?q?PC=20=E9=A9=B1=E5=8A=A8=20TPC/TPV?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - transpyc_mcp/server.py：零依赖 MCP 服务器（initialize/tools/list/tools/call） - 14 个工具按协议实现：tpc_build/tpc_build_file/tpc_run_test/tpc_run_regression/ tpc_clean_project/tpc_clear_cache/tpc_rebuild_includes/tpv_build/tpv_bootstrap/ tpv_compile_project/tpv_run_test/tp_read_ast/tp_read_log/tp_sha1_map - 安全：路径白名单/命令白名单/工程串行锁/TPV 内存保护（>=1GiB） - 输出统一 {ok, exit_code, stdout, stderr, artifacts, elapsed_ms}，日志末尾 50 行 - 冒烟测试通过：init/tools-list/AST 解析/日志/越界拒绝/未知工具/锁 - 修正协议：--clean 为清理后继续完整编译（非独立清理） --- MCP/README.md | 2 +- MCP/protocol.md | 4 +- MCP/tools/tools.json | 2 +- transpyc_mcp/__init__.py | 8 + transpyc_mcp/server.py | 583 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 595 insertions(+), 4 deletions(-) create mode 100644 transpyc_mcp/__init__.py create mode 100644 transpyc_mcp/server.py diff --git a/MCP/README.md b/MCP/README.md index 61aceb5..ec6fb99 100644 --- a/MCP/README.md +++ b/MCP/README.md @@ -48,7 +48,7 @@ MCP 服务器（`transpyc-tools`）以 stdio 方式运行于项目根目录： | 编译任意工程 | `tpc_build` | `python Projectrans.py --project [--clean] [--run] ...` | | 单文件转 LLVM IR | `tpc_build_file` | `python TransPyC.py -f -o ` | | 全量回归 | `tpc_run_regression` | `python Test/regression_test.py` | -| 清理工程产物 | `tpc_clean_project` | `python Projectrans.py --project

--clean` | +| 清理并编译 | `tpc_clean_project` | `python Projectrans.py --project

--clean`（清理后继续完整编译） | | 清全局缓存 | `tpc_clear_cache` | `python Projectrans.py --project

--clear-cache` | | 重建 includes | `tpc_rebuild_includes` | `python Projectrans.py --project

--rebuild-includes` | | 构建 TPV | `tpv_build` | `python Projectrans.py --project TransPyV/project.json --clean` | diff --git a/MCP/protocol.md b/MCP/protocol.md index 7b382ae..c9c6072 100644 --- a/MCP/protocol.md +++ b/MCP/protocol.md @@ -220,9 +220,9 @@ python Test/regression_test.py - 说明：默认全量耗时较长；客户端应按需使用 `name` 过滤或分片调用。 -### 6.5 tpc_clean_project — 清理工程产物 +### 6.5 tpc_clean_project — 清理并编译工程 -- description：清理工程 `output/` 与 `temp/`（`Projectrans.py --clean`）。 +- description：执行 `Projectrans.py --clean`——先清理 `output/` 与 `temp/`，**随后仍继续完整编译流程**（`--clean` 是编译前置步骤，非独立清理命令）。等价 `tpc_build(clean=true)`，仅不含运行。 - inputSchema：`{ "project": {"type":"string"} }`，required: ["project"]。 - command：`python Projectrans.py --project --clean` diff --git a/MCP/tools/tools.json b/MCP/tools/tools.json index 770eebe..5ead1d1 100644 --- a/MCP/tools/tools.json +++ b/MCP/tools/tools.json @@ -147,7 +147,7 @@ }, { "name": "tpc_clean_project", - "description": "清理工程 output/temp（--clean）", + "description": "清理并编译工程（--clean，清理后继续完整编译流程）", "inputSchema": { "type": "object", "properties": { diff --git a/transpyc_mcp/__init__.py b/transpyc_mcp/__init__.py new file mode 100644 index 0000000..b58df4f --- /dev/null +++ b/transpyc_mcp/__init__.py @@ -0,0 +1,8 @@ +# -*- coding: utf-8 -*- +"""transpyc_mcp - TransPyC/TransPyV MCP 服务器包 + +实现 MCP/ 协议（stdio transport，零第三方依赖）。 +入口：python -m transpyc_mcp.server +""" + +__version__ = '1.0.0' \ No newline at end of file diff --git a/transpyc_mcp/server.py b/transpyc_mcp/server.py new file mode 100644 index 0000000..d29122c --- /dev/null +++ b/transpyc_mcp/server.py @@ -0,0 +1,583 @@ +# -*- 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 + +# --------------------------------------------------------------------------- +# 常量 +# --------------------------------------------------------------------------- +_ROOT: str = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_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'): + 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('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']) + 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() \ No newline at end of file