58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""兄弟技能 CLI 调用工具。
|
||
|
||
通过子进程调用同级其他 skill 的 main.py，并解析 JSON 输出。
|
||
具体调用哪些兄弟技能，由复制后的业务 skill 在 service 层按需决定。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from util.logging_config import subprocess_env_with_trace
|
||
from util.runtime_paths import get_skills_root
|
||
|
||
|
||
def call_sibling_json(skill_slug: str, args: List[str]) -> Optional[Dict[str, Any]]:
|
||
"""通用兄弟技能调用器。
|
||
|
||
Args:
|
||
skill_slug: 兄弟技能的 slug（即同级目录名）。
|
||
args: 传给 main.py 的命令行参数列表。
|
||
|
||
Returns:
|
||
若兄弟技能输出可解析的 JSON 对象则返回 dict；否则返回 None。
|
||
"""
|
||
script_path = os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
|
||
if not os.path.isfile(script_path):
|
||
return None
|
||
proc = subprocess.run(
|
||
[sys.executable, script_path, *args],
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
env=subprocess_env_with_trace(),
|
||
)
|
||
raw = (proc.stdout or "").strip()
|
||
if not raw or raw.startswith("ERROR"):
|
||
return None
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return None
|
||
return data if isinstance(data, dict) else None
|
||
|
||
|
||
def get_sibling_main_path(skill_slug: str) -> str:
|
||
"""获取兄弟技能的 main.py 绝对路径。
|
||
|
||
使用示例（在 task_service.py 中）：
|
||
from service.sibling_bridge import get_sibling_main_path, call_sibling_json
|
||
result = call_sibling_json("account-manager", ["list", "--limit", "10"])
|
||
"""
|
||
return os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
|