feat(mcp): 实现 transpyc_mcp 服务器——stdio JSON-RPC 驱动 TPC/TPV
- 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 为清理后继续完整编译（非独立清理）
This commit is contained in:
TermiNexus committed
1 parent
e412885a3e
commit
004121de40
5 files changed
+595
-4
No files matched your search
+1
-1
@@ -48,7 +48,7 @@ MCP 服务器（`transpyc-tools`）以 stdio 方式运行于项目根目录：
|
||||
| 编译任意工程 | `tpc_build` | `python Projectrans.py --project <project.json> [--clean] [--run] ...` |
|
||||
| 单文件转 LLVM IR | `tpc_build_file` | `python TransPyC.py -f <in.vp> -o <out.ll>` |
|
||||
| 全量回归 | `tpc_run_regression` | `python Test/regression_test.py` |
|
||||
| 清理工程产物 | `tpc_clean_project` | `python Projectrans.py --project <p> --clean` |
|
||||
| 清理并编译 | `tpc_clean_project` | `python Projectrans.py --project <p> --clean`（清理后继续完整编译） |
|
||||
| 清全局缓存 | `tpc_clear_cache` | `python Projectrans.py --project <p> --clear-cache` |
|
||||
| 重建 includes | `tpc_rebuild_includes` | `python Projectrans.py --project <p> --rebuild-includes` |
|
||||
| 构建 TPV | `tpv_build` | `python Projectrans.py --project TransPyV/project.json --clean` |
|
||||
|
||||
+2
-2
@@ -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 <project> --clean`
|
||||
|
||||
|
||||
@@ -147,7 +147,7 @@
|
||||
},
|
||||
{
|
||||
"name": "tpc_clean_project",
|
||||
"description": "清理工程 output/temp（--clean）",
|
||||
"description": "清理并编译工程（--clean，清理后继续完整编译流程）",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""transpyc_mcp - TransPyC/TransPyV MCP 服务器包
|
||||
|
||||
实现 MCP/ 协议（stdio transport，零第三方依赖）。
|
||||
入口：python -m transpyc_mcp.server
|
||||
"""
|
||||
|
||||
__version__ = '1.0.0'
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user