建構「多 GitHub Repo AI Skills 聚合知識庫」：人機雙軌查詢架構與實戰指南 ¶

#!/usr/bin/env python3
"""
多 GitHub Repo AI Skills 自動聚合、資安掃描與雙軌索引編譯器
同時產出供人類閱讀之 MkDocs Markdown 目錄與供 AI 查詢之 skills-registry.json
"""
from pathlib import Path
import json
import re
import subprocess
import tempfile
import yaml

SUSPICIOUS_PATTERNS = [
    r"curl\s+.*\|\s*(ba)?sh",
    r"wget\s+.*\|\s*(ba)?sh",
    r"ignore\s+all\s+previous\s+instructions",
    r"<\!--.*(system|override|exfiltrate).*-->",
]


def parse_skill_frontmatter(md_content: str) -> dict:
    """解析 SKILL.md 頂部的 YAML Frontmatter"""
    match = re.match(r"^---\s*\n(.*?)\n---\s*\n", md_content, re.DOTALL)
    if not match:
        return {}
    try:
        data = yaml.safe_load(match.group(1))
        return data if isinstance(data, dict) else {}
    except yaml.YAMLError:
        return {}


def scan_security_risks(md_content: str) -> list[str]:
    """靜態掃描 SKILL.md 是否包含可疑的 Prompt Injection 或高風險遠端執行指令"""
    warnings = []
    for pattern in SUSPICIOUS_PATTERNS:
        if re.search(pattern, md_content, re.IGNORECASE | re.DOTALL):
            warnings.append(f"Matched high-risk pattern: {pattern}")
    return warnings


def sync_and_compile_catalog(manifest_path: str, output_json: str, output_md: str) -> None:
    manifest = yaml.safe_load(Path(manifest_path).read_text(encoding="utf-8"))
    registry_entries = []

    with tempfile.TemporaryDirectory() as tmpdir:
        for repo_cfg in manifest.get("repositories", []):
            repo_name = repo_cfg["repo"]
            repo_dir = Path(tmpdir) / repo_cfg["id"]

            # 淺層複製最新分支
            subprocess.run(
                ["git", "clone", "--depth", "1", "--branch", repo_cfg["branch"], repo_cfg["url"], str(repo_dir)],
                check=True,
                capture_output=True,
            )
            commit_sha = subprocess.check_output(
                ["git", "-C", str(repo_dir), "rev-parse", "--short", "HEAD"],
                text=True,
            ).strip()

            # 掃描倉庫內所有的 SKILL.md
            for skill_file in sorted(repo_dir.rglob("SKILL.md")):
                rel_path = skill_file.relative_to(repo_dir).as_posix()
                content = skill_file.read_text(encoding="utf-8", errors="ignore")
                meta = parse_skill_frontmatter(content)
                sec_warnings = scan_security_risks(content)

                skill_name = meta.get("name") or skill_file.parent.name
                description = str(meta.get("description", "No description provided.")).strip().replace("\n", " ")

                registry_entries.append({
                    "skill_id": f"{repo_cfg['id']}/{skill_name}",
                    "name": skill_name,
                    "repo": repo_name,
                    "commit_sha": commit_sha,
                    "category": repo_cfg["category"],
                    "governance_tier": repo_cfg["default_tier"],
                    "relative_path": rel_path,
                    "raw_url": f"https://raw.githubusercontent.com/{repo_name}/{commit_sha}/{rel_path}",
                    "install_cmd": f"npx skills add {repo_name} --skill {skill_name}",
                    "description": description,
                    "has_scripts": (skill_file.parent / "scripts").is_dir(),
                    "has_references": (skill_file.parent / "references").is_dir(),
                    "security_verified": len(sec_warnings) == 0,
                    "security_warnings": sec_warnings,
                })

    # 1. 輸出供 AI Agent 與 MCP Server 查詢的結構化 JSON 索引
    Path(output_json).write_text(
        json.dumps({"total_skills": len(registry_entries), "skills": registry_entries}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    # 2. 輸出供人類閱讀與 MkDocs 全文檢索的 Markdown 表格
    lines = [
        "# 自動編譯跨儲存庫 AI Skills 總覽表",
        "",
        "| 技能名稱 | 所屬倉庫 | 分類 | 治理層級 | 功能與觸發條件摘要 | 安裝指令 |",
        "| :--- | :--- | :--- | :--- | :--- | :--- |",
    ]
    for item in registry_entries:
        if not item["security_verified"]:
            continue
        lines.append(
            f"| **`{item['name']}`** | `{item['repo']}` (`{item['commit_sha']}`) "
            f"| {item['category']} | `{item['governance_tier']}` "
            f"| {item['description'][:120]} | `{item['install_cmd']}` |"
        )
    Path(output_md).write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    sync_and_compile_catalog(
        manifest_path="repos.yaml",
        output_json="docs/skills-registry.json",
        output_md="docs/notes/compiled-skills-index.md",
    )

"""
Multi-Repo AI Skill Hub MCP Server
讓 AI 代理跨多個 GitHub Skill 儲存庫執行三階段漸進式查詢與按需載入 (On-Demand Skill Loading)
"""
from mcp.server.fastmcp import FastMCP
import json
import urllib.request

mcp = FastMCP("Windcold-MultiRepo-Skill-Hub")
REGISTRY_URL = "https://windcoldchen.github.io/windcold-kb/skills-registry.json"


def _load_registry() -> list[dict]:
    with urllib.request.urlopen(REGISTRY_URL, timeout=5) as resp:
        data = json.loads(resp.read().decode("utf-8"))
        return data.get("skills", [])


@mcp.tool()
def search_ai_skills(query: str, category: str = "", max_results: int = 5) -> str:
    """
    [Stage 1: Discovery] 跨多個開源與內部 GitHub Skill 儲存庫搜尋匹配的 AI 技能。
    回傳精簡的 skill_id、repo、觸發條件摘要與治理等級，僅消耗極少 Token。
    """
    keywords = [k.lower() for k in query.split() if k.strip()]
    results = []

    for skill in _load_registry():
        if not skill.get("security_verified", False):
            continue
        if category and category.lower() not in skill["category"].lower():
            continue

        searchable_text = f"{skill['skill_id']} {skill['description']} {skill['category']}".lower()
        score = sum(1 for kw in keywords if kw in searchable_text)
        if score > 0:
            results.append((score, skill))

    results.sort(key=lambda x: x[0], reverse=True)
    matched = [
        {
            "skill_id": s["skill_id"],
            "repo": s["repo"],
            "governance_tier": s["governance_tier"],
            "description": s["description"],
            "install_cmd": s["install_cmd"],
        }
        for _, s in results[:max_results]
    ]
    return json.dumps(matched, ensure_ascii=False, indent=2)


@mcp.tool()
def load_skill_instructions(skill_id: str) -> str:
    """
    [Stage 2: Activation] 根據 search_ai_skills 查到的 skill_id，
    動態從經過 Commit SHA 安全鎖定的儲存庫拉取完整 SKILL.md 指令本體供 AI 執行。
    """
    for skill in _load_registry():
        if skill["skill_id"] == skill_id and skill.get("security_verified", False):
            with urllib.request.urlopen(skill["raw_url"], timeout=5) as resp:
                return resp.read().decode("utf-8")
    return f"找不到或未通過安全驗證之技能: {skill_id}"


if __name__ == "__main__":
    mcp.run()