chore(mcp): 支持 Node 和 Python 运行时配置

2026-09-28 18:02:36 +08:00
parent 99fc673e8c
commit 5ebddadf50
10 changed files with 245 additions and 74 deletions
--- a/.codex/README.md
+++ b/.codex/README.md
@@ -13,3 +13,23 @@ TAPD 仅开放配置中的读取工具。查询“我的待办”时，默认查
 `.codex/dbhub_mcp/` 提供数据库 MCP 配置模板和只读启动器。开发者可按 [DBHub 官方说明](https://github.com/bytebase/dbhub)安装 `@bytebase/dbhub@1.3.1` 到个人的 `~/.codex/mcp/dbhub/`，并在 `.env.local` 中填写实际安装入口和自己的只读数据库 DSN。由于示例中没有可安全使用的数据库连接信息，`config.toml` 默认设为 `enabled = false`；填写 DSN 后再改为 `true` 并重启任务。示例 DSN 仅作格式示意，不是项目真实连接。

 启动器通过环境变量传 DSN，不在命令参数中暴露密码。DBHub 1.3.1 的只读限制在 `dbhub.toml` 中对 `execute_sql` 设置；数据库账号本身仍应授予只读权限，因为应用层 SQL 检查不能替代数据库权限。不要用生产或共享环境做配置验证。
+
+## Node / Python 入口
+
+三个启动器共享 `mcp_runtime.py`，支持 `MCP_RUNTIME="auto"`（默认）、`"node"` 和 `"python"`。
+`auto` 识别 `.js/.cjs/.mjs`、`.py` 和带 Node/Python shebang 的入口。
+入口必须填写绝对路径；`NODE_BIN` / `PYTHON_BIN` 支持绝对路径或 PATH 中的命令名。
+桌面进程的 PATH 可能与终端不同，推荐填写解释器绝对路径。Python MCP 应使用安装它的虚拟环境解释器。
+启动配置文件的 `command="python3"` 是外层启动器（要求 Python 3.9+）；`PYTHON_BIN` 只决定内部 MCP 的解释器。
+
+TAPD 支持原 Node 实现，以及 Python 包 `mcp-server-tapd` 的 `bin/mcp-server-tapd` 入口。
+Python 模式将 `TAPD_API_TOKEN` 映射到 `TAPD_ACCESS_TOKEN`，将 `TAPD_CURRENT_USER` 映射到 `CURRENT_USER_NICK`，强制使用 stdio。
+可用 `TAPD_API_BASE_URL`、`TAPD_BASE_URL` 指定服务地址，默认使用 TAPD 官方站点。
+不要指向另行加载个人 credentials.env 的包装脚本，以免覆盖项目凭据。
+配置保留两种实现的只读工具白名单；Python 查询须显式传入本项目 `workspace_id`，不会自动将项目 ID 注入每次调用。
+
+运行时兼容不代表任意同名 MCP 的参数和 API 兼容：Playwright 入口须支持现有 Chrome/profile/output 参数，DBHub 入口须支持现有 `--config` 和只读配置。
+Playwright 和 DBHub 的已验证安装包是 Node 实现，不需要改成 Python。
+TEST_* 字段仍由执行测试的代理按需使用，不自动登录；DBHub 的默认禁用状态保持不变。
+
+离线回归：`python3 -B .codex/tests/test_mcp_runtime.py`，使用临时配置和模拟进程启动，不访问 TAPD 或数据库。
--- a/.codex/config.toml
+++ b/.codex/config.toml
@@ -4,7 +4,7 @@ args = [".codex/tapd_mcp/launch.py"]
 enabled = true
 startup_timeout_sec = 20
 tool_timeout_sec = 60
-enabled_tools = ["tapd_get_current_user", "tapd_parse_url", "tapd_get_story", "tapd_list_stories", "tapd_get_task", "tapd_list_tasks", "tapd_get_bug", "tapd_list_bugs", "tapd_get_iteration", "tapd_list_iterations", "tapd_list_comments", "tapd_get_story_related_bugs", "tapd_get_story_changes", "tapd_list_workspace_users"]
+enabled_tools = ["tapd_get_current_user", "tapd_parse_url", "tapd_get_story", "tapd_list_stories", "tapd_get_task", "tapd_list_tasks", "tapd_get_bug", "tapd_list_bugs", "tapd_get_iteration", "tapd_list_iterations", "tapd_list_comments", "tapd_get_story_related_bugs", "tapd_get_story_changes", "tapd_list_workspace_users", "get_stories_or_tasks", "get_story_or_task_count", "get_bug", "get_bug_count", "get_comments", "get_iterations", "get_related_bugs", "get_workspace_users", "get_entity_custom_fields", "get_entity_attachments", "get_image", "get_workflows_status_map", "get_workitem_types"]

 [mcp_servers.playwright]
 command = "python3"
--- a/.codex/dbhub_mcp/.env.example
+++ b/.codex/dbhub_mcp/.env.example
@@ -1,3 +1,8 @@
 DBHUB_DSN="mysql://readonly_user:password@127.0.0.1:3306/dataease"
 DBHUB_MCP_ENTRY="/absolute/path/to/@bytebase/dbhub/dist/index.js"
 NODE_BIN="node"
+
+# auto 根据扩展名或 shebang 判断；也可指定 node / python
+MCP_RUNTIME="auto"
+# Python 入口请使用安装该 MCP 的虚拟环境解释器
+PYTHON_BIN="python3"
--- a/.codex/dbhub_mcp/launch.py
+++ b/.codex/dbhub_mcp/launch.py
@@ -1,30 +1,19 @@
-"""Launch a locally installed DBHub in read-only mode."""
-import json
+"""Launch DBHub using the existing read-only configuration."""
 import os
-import shutil
 import sys
 from pathlib import Path

+sys.dont_write_bytecode = True
 base = Path(__file__).resolve().parent
-try:
-    local = {}
-    for raw in (base / '.env.local').read_text().splitlines():
-        line = raw.strip()
-        if not line or line.startswith('#'):
-            continue
-        key, sep, value = line.partition('=')
-        if not sep or key.strip() not in {'DBHUB_DSN', 'DBHUB_MCP_ENTRY', 'NODE_BIN'}:
-            raise ValueError()
-        local[key.strip()] = json.loads(value.strip())
-    if not all(isinstance(local.get(k), str) and local[k] for k in ('DBHUB_DSN', 'DBHUB_MCP_ENTRY')):
-        raise ValueError()
-    entry = Path(local['DBHUB_MCP_ENTRY']).expanduser()
-    node = local.get('NODE_BIN') or shutil.which('node')
-    if not entry.is_file() or not node:
-        raise ValueError()
-except (OSError, ValueError, KeyError, TypeError):
-    sys.exit('DBHub MCP 配置无效；请检查 .codex/dbhub_mcp/.env.local')
+sys.path.insert(0, str(base.parent))
+from mcp_runtime import execute, load_config, resolve_command

-env = os.environ.copy()
-env['DSN'] = local['DBHUB_DSN']
-os.execvpe(node, [node, str(entry), '--config', str(base / 'dbhub.toml')], env)
+try:
+    local = load_config(base, {'DBHUB_DSN', 'DBHUB_MCP_ENTRY'}, {'DBHUB_DSN', 'DBHUB_MCP_ENTRY'})
+    _, command = resolve_command(local, 'DBHUB_MCP_ENTRY')
+    env = os.environ.copy()
+    env['DSN'] = local['DBHUB_DSN']
+    command.extend(['--config', str(base / 'dbhub.toml')])
+    execute(command, env)
+except (OSError, ValueError, KeyError, TypeError):
+    sys.exit('DBHub MCP 启动失败；请检查 .codex/dbhub_mcp/.env.local 的字段、入口及解释器')
--- a/.codex/mcp_runtime.py
+++ b/.codex/mcp_runtime.py
@@ -0,0 +1,62 @@
+"""Shared local MCP configuration and runtime resolution (Python 3.9+)."""
+import json
+import os
+import shutil
+from pathlib import Path
+
+RUNTIME_KEYS = {'MCP_RUNTIME', 'NODE_BIN', 'PYTHON_BIN'}
+
+
+def load_config(base, allowed, required):
+    local = {}
+    for number, raw in enumerate((base / '.env.local').read_text().splitlines(), 1):
+        line = raw.strip()
+        if not line or line.startswith('#'):
+            continue
+        key, sep, value = line.partition('=')
+        key = key.strip()
+        if not sep or key not in allowed | RUNTIME_KEYS or key in local:
+            raise ValueError(f'配置第 {number} 行存在未知或重复字段')
+        try:
+            parsed = json.loads(value.strip())
+        except ValueError:
+            raise ValueError(f'配置第 {number} 行必须使用 JSON 字符串') from None
+        if not isinstance(parsed, str):
+            raise ValueError(f'配置第 {number} 行必须使用字符串')
+        local[key] = parsed
+    for key in required:
+        if not local.get(key):
+            raise ValueError(f'缺少配置字段 {key}')
+    return local
+
+
+def resolve_command(local, entry_key):
+    entry = Path(local[entry_key]).expanduser()
+    if not entry.is_absolute() or not entry.is_file():
+        raise ValueError(f'{entry_key} 必须是存在的文件绝对路径')
+    runtime = local.get('MCP_RUNTIME') or 'auto'
+    if runtime == 'auto':
+        if entry.suffix in {'.js', '.cjs', '.mjs'}:
+            runtime = 'node'
+        elif entry.suffix == '.py':
+            runtime = 'python'
+        else:
+            with entry.open('rb') as stream:
+                shebang = stream.readline(512)
+            if shebang.startswith(b'#!') and b'python' in shebang:
+                runtime = 'python'
+            elif shebang.startswith(b'#!') and b'node' in shebang:
+                runtime = 'node'
+            else:
+                raise ValueError('无法判断入口类型，请明确设置 MCP_RUNTIME')
+    if runtime not in {'node', 'python'}:
+        raise ValueError('MCP_RUNTIME 仅支持 auto、node、python')
+    key, fallback = ('NODE_BIN', 'node') if runtime == 'node' else ('PYTHON_BIN', 'python3')
+    binary = shutil.which(str(Path(local.get(key) or fallback).expanduser()))
+    if not binary:
+        raise ValueError(f'{key} 不是可执行文件，请指定实际解释器路径')
+    return runtime, [binary, str(entry)]
+
+
+def execute(command, env):
+    os.execvpe(command[0], command, env)
--- a/.codex/playwright_mcp/.env.example
+++ b/.codex/playwright_mcp/.env.example
@@ -3,3 +3,8 @@ NODE_BIN="node"
 TEST_BASE_URL="https://qa-de3.fit2cloud.cn"
 TEST_USERNAME=""
 TEST_PASSWORD=""
+
+# auto 根据扩展名或 shebang 判断；也可指定 node / python
+MCP_RUNTIME="auto"
+# Python 入口请使用安装该 MCP 的虚拟环境解释器
+PYTHON_BIN="python3"
--- a/.codex/playwright_mcp/launch.py
+++ b/.codex/playwright_mcp/launch.py
@@ -1,30 +1,22 @@
 """Launch Playwright MCP with project-local browser storage."""
-import json
 import os
-import shutil
 import sys
 from pathlib import Path

+sys.dont_write_bytecode = True
 base = Path(__file__).resolve().parent
-try:
-    local = {}
-    for raw in (base / '.env.local').read_text().splitlines():
-        line = raw.strip()
-        if not line or line.startswith('#'):
-            continue
-        key, sep, value = line.partition('=')
-        if not sep or key.strip() not in {'PLAYWRIGHT_MCP_ENTRY', 'TEST_BASE_URL', 'TEST_USERNAME', 'TEST_PASSWORD', 'NODE_BIN'}:
-            raise ValueError()
-        local[key.strip()] = json.loads(value.strip())
-    entry = Path(local['PLAYWRIGHT_MCP_ENTRY']).expanduser()
-    node = local.get('NODE_BIN') or shutil.which('node')
-    if not entry.is_file() or not node:
-        raise ValueError()
-except (OSError, ValueError, KeyError, TypeError):
-    sys.exit('Playwright MCP 配置无效；请检查 .codex/playwright_mcp/.env.local')
+sys.path.insert(0, str(base.parent))
+from mcp_runtime import execute, load_config, resolve_command

-profile = base / 'browser-profile'
-output = base / 'output'
-profile.mkdir(exist_ok=True, mode=0o700)
-output.mkdir(exist_ok=True, mode=0o700)
-os.execvpe(node, [node, str(entry), '--browser', 'chrome', '--user-data-dir', str(profile), '--output-dir', str(output)], os.environ.copy())
+try:
+    local = load_config(base, {'PLAYWRIGHT_MCP_ENTRY', 'TEST_BASE_URL', 'TEST_USERNAME', 'TEST_PASSWORD'},
+                        {'PLAYWRIGHT_MCP_ENTRY'})
+    _, command = resolve_command(local, 'PLAYWRIGHT_MCP_ENTRY')
+    profile = base / 'browser-profile'
+    output = base / 'output'
+    profile.mkdir(exist_ok=True, mode=0o700)
+    output.mkdir(exist_ok=True, mode=0o700)
+    command.extend(['--browser', 'chrome', '--user-data-dir', str(profile), '--output-dir', str(output)])
+    execute(command, os.environ.copy())
+except (OSError, ValueError, KeyError, TypeError):
+    sys.exit('Playwright MCP 启动失败；请检查 .codex/playwright_mcp/.env.local 的字段、入口及解释器')
--- a/.codex/tapd_mcp/.env.example
+++ b/.codex/tapd_mcp/.env.example
@@ -3,3 +3,13 @@ TAPD_CURRENT_USER="your TAPD username"
 TAPD_API_TOKEN="your personal token"
 TAPD_MCP_ENTRY="/absolute/path/to/tapd-mcp/dist/index.js"
 NODE_BIN="node"
+
+# auto 根据扩展名或 shebang 判断；也可指定 node / python
+MCP_RUNTIME="auto"
+# Python 入口请使用安装该 MCP 的虚拟环境解释器
+PYTHON_BIN="python3"
+
+# Python mcp-server-tapd：入口填写虚拟环境 bin/mcp-server-tapd
+# PYTHON_BIN 填写同一虚拟环境 bin/python，令牌仍填写 TAPD_API_TOKEN
+TAPD_API_BASE_URL="https://api.tapd.cn"
+TAPD_BASE_URL="https://www.tapd.cn"
--- a/.codex/tapd_mcp/launch.py
+++ b/.codex/tapd_mcp/launch.py
@@ -1,32 +1,34 @@
-"""Launch TAPD MCP with project-local credentials."""
-import json
+"""Launch Node or Python TAPD MCP using only project-local credentials."""
 import os
-import shutil
 import sys
 from pathlib import Path

+sys.dont_write_bytecode = True
 base = Path(__file__).resolve().parent
-try:
-    local = {}
-    for raw in (base / '.env.local').read_text().splitlines():
-        line = raw.strip()
-        if not line or line.startswith('#'):
-            continue
-        key, sep, value = line.partition('=')
-        if not sep or key.strip() not in {'TAPD_WORKSPACE_ID', 'TAPD_CURRENT_USER', 'TAPD_API_TOKEN', 'TAPD_MCP_ENTRY', 'NODE_BIN'}:
-            raise ValueError()
-        local[key.strip()] = json.loads(value.strip())
-    if not all(isinstance(local.get(k), str) and local[k] for k in ('TAPD_WORKSPACE_ID', 'TAPD_CURRENT_USER', 'TAPD_API_TOKEN', 'TAPD_MCP_ENTRY')):
-        raise ValueError()
-    entry = Path(local['TAPD_MCP_ENTRY']).expanduser()
-    node = local.get('NODE_BIN') or shutil.which('node')
-    if not entry.is_file() or not node:
-        raise ValueError()
-except (OSError, ValueError, KeyError, TypeError):
-    sys.exit('TAPD MCP 配置无效；请检查 .codex/tapd_mcp/.env.local')
+sys.path.insert(0, str(base.parent))
+from mcp_runtime import execute, load_config, resolve_command

-env = os.environ.copy()
-env.pop('TAPD_API_USER', None)
-env.pop('TAPD_API_PASSWORD', None)
-env.update({key: local[key] for key in ('TAPD_WORKSPACE_ID', 'TAPD_CURRENT_USER', 'TAPD_API_TOKEN')})
-os.execvpe(node, [node, str(entry)], env)
+try:
+    keys = {'TAPD_WORKSPACE_ID', 'TAPD_CURRENT_USER', 'TAPD_API_TOKEN',
+            'TAPD_MCP_ENTRY', 'TAPD_API_BASE_URL', 'TAPD_BASE_URL'}
+    local = load_config(base, keys, {'TAPD_WORKSPACE_ID', 'TAPD_CURRENT_USER',
+                                    'TAPD_API_TOKEN', 'TAPD_MCP_ENTRY'})
+    runtime, command = resolve_command(local, 'TAPD_MCP_ENTRY')
+    env = os.environ.copy()
+    for key in ('TAPD_API_USER', 'TAPD_API_PASSWORD', 'TAPD_ACCESS_TOKEN',
+                'TAPD_API_BASE_URL', 'TAPD_BASE_URL', 'CURRENT_USER_NICK'):
+        env.pop(key, None)
+    env.update({key: local[key] for key in ('TAPD_WORKSPACE_ID', 'TAPD_CURRENT_USER', 'TAPD_API_TOKEN')})
+    if runtime == 'python':
+        env['TAPD_ACCESS_TOKEN'] = local['TAPD_API_TOKEN']
+        env['CURRENT_USER_NICK'] = local['TAPD_CURRENT_USER']
+        env['TAPD_API_BASE_URL'] = local.get('TAPD_API_BASE_URL') or 'https://api.tapd.cn'
+        env['TAPD_BASE_URL'] = local.get('TAPD_BASE_URL') or 'https://www.tapd.cn'
+        command.extend(['--mode', 'stdio'])
+    else:
+        for key in ('TAPD_API_BASE_URL', 'TAPD_BASE_URL'):
+            if local.get(key):
+                env[key] = local[key]
+    execute(command, env)
+except (OSError, ValueError, KeyError, TypeError):
+    sys.exit('TAPD MCP 启动失败；请检查 .codex/tapd_mcp/.env.local 的字段、入口及解释器')
--- a/.codex/tests/test_mcp_runtime.py
+++ b/.codex/tests/test_mcp_runtime.py
@@ -0,0 +1,86 @@
+"""Offline launcher regression tests; no real credentials or services."""
+import json
+import os
+from pathlib import Path
+import runpy
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.dont_write_bytecode = True
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+from mcp_runtime import load_config, resolve_command
+
+
+class RuntimeTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.base = Path(self.temp.name)
+
+    def entry(self, name, content=''):
+        p = self.base / name
+        p.write_text(content)
+        return str(p)
+
+    def test_runtime_detection(self):
+        for name, content, expected in [('index.js', '', 'node'), ('index.mjs', '', 'node'),
+                                        ('index.cjs', '', 'node'), ('main.py', '', 'python'),
+                                        ('mcp-server', '#!/venv/bin/python\n', 'python')]:
+            with self.subTest(name=name):
+                cfg = {'ENTRY': self.entry(name, content), 'NODE_BIN': sys.executable, 'PYTHON_BIN': sys.executable}
+                runtime, command = resolve_command(cfg, 'ENTRY')
+                self.assertEqual(runtime, expected)
+                self.assertEqual(command[0], sys.executable)
+
+    def test_explicit_runtime_and_missing_binary(self):
+        cfg = {'ENTRY': self.entry('custom'), 'MCP_RUNTIME': 'python', 'PYTHON_BIN': sys.executable}
+        self.assertEqual(resolve_command(cfg, 'ENTRY')[0], 'python')
+        cfg['PYTHON_BIN'] = '/does-not-exist/python'
+        with self.assertRaises(ValueError):
+            resolve_command(cfg, 'ENTRY')
+
+    def test_bad_configs_do_not_echo_secrets(self):
+        for value in ['TOKEN="secret"\nTOKEN="secret"', 'TOKEN=secret', 'TOKEN=123', 'UNKNOWN="secret"']:
+            (self.base / '.env.local').write_text(value)
+            with self.assertRaises(ValueError) as error:
+                load_config(self.base, {'TOKEN'}, {'TOKEN'})
+            self.assertNotIn('secret', str(error.exception))
+
+    def test_launcher_contracts(self):
+        for service, runtime in [('tapd', 'node'), ('tapd', 'python'), ('playwright', 'node'), ('dbhub', 'node')]:
+            with self.subTest(service=service, runtime=runtime):
+                config = {'TAPD_WORKSPACE_ID': '123', 'TAPD_CURRENT_USER': 'tester',
+                          'TAPD_API_TOKEN': 'test-token', 'DBHUB_DSN': 'test-dsn'}
+                command = [sys.executable, '/fake/entry']
+                with patch('mcp_runtime.load_config', return_value=config), \
+                     patch('mcp_runtime.resolve_command', return_value=(runtime, command)), \
+                     patch('mcp_runtime.execute') as execute, \
+                     patch.object(Path, 'mkdir'), \
+                     patch.dict(os.environ, {'TAPD_API_USER': 'old', 'TAPD_API_PASSWORD': 'old', 'TAPD_ACCESS_TOKEN': 'old'}):
+                    runpy.run_path(str(ROOT / (service + '_mcp') / 'launch.py'), run_name='__main__')
+                args, env = execute.call_args.args
+                self.assertNotIn('test-token', args)
+                self.assertNotIn('test-dsn', args)
+                if service == 'tapd':
+                    self.assertNotIn('TAPD_API_USER', env)
+                    self.assertNotIn('TAPD_API_PASSWORD', env)
+                    self.assertEqual(env['TAPD_API_TOKEN'], 'test-token')
+                    if runtime == 'python':
+                        self.assertEqual(env['TAPD_ACCESS_TOKEN'], 'test-token')
+                        self.assertEqual(env['CURRENT_USER_NICK'], 'tester')
+                        self.assertEqual(args[-2:], ['--mode', 'stdio'])
+                    else:
+                        self.assertNotIn('TAPD_ACCESS_TOKEN', env)
+                elif service == 'dbhub':
+                    self.assertEqual(env['DSN'], 'test-dsn')
+                    self.assertIn('--config', args)
+                else:
+                    self.assertIn('--user-data-dir', args)
+                    self.assertIn('--output-dir', args)
+
+
+if __name__ == '__main__':
+    unittest.main()