TermiNexus
7d147c55e1
- tpc_build/tpc_run_test/tpv_build/tpv_compile_project 的 asan 默认 true - tpv_build 新增 asan 参数（编译器自身 ASan 插桩，崩溃时可输出报告定位） - tpv_bootstrap Phase A 对齐 tpv_build 加 --asan - includes/coro.vp: 裸 null 改 t.CPtr(0)（TPV 兼容，M1 遗留未提交）
596 lines
23 KiB
Python
596 lines
23 KiB
Python
# -*- 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() |