建構企業內部 AI Skill Hub 平台：整合 Public GitHub 與內部自研技能之人機雙軌實戰指南 ¶

#!/usr/bin/env python3
"""
企業內部 AI Skill Hub 編譯器：
1. 驗證 Public GitHub 技能與公司內部自研技能的 Frontmatter 與資安合規性
2. 將 overlays/ 企業本地化規則自動合併至開源 SKILL.md
3. 產出 dist/skills/ 標準技能包、skills-hub-registry.json 與 MkDocs 技能市集頁面
"""
from pathlib import Path
import hashlib
import json
import re
import shutil
import sys
import yaml

FORBIDDEN_PATTERNS = [
    (r"curl\s+.*\|\s*(ba)?sh", "禁止使用 curl pipe to shell 遠端執行"),
    (r"wget\s+.*\|\s*(ba)?sh", "禁止使用 wget pipe to shell 遠端執行"),
    (r"ignore\s+(all\s+)?previous\s+instructions", "偵測到疑似 Prompt Injection 攻擊語句"),
    (r"(ghp_|glpat-|sk-ant-|AKIA)[A-Za-z0-9_\-]{10,}", "偵測到疑似硬編碼機敏 Token / Key"),
]


def extract_frontmatter(md_text: str) -> dict:
    match = re.match(r"^---\s*\n(.*?)\n---\s*\n", md_text, re.DOTALL)
    if not match:
        return {}
    return yaml.safe_load(match.group(1)) or {}


def security_audit(skill_id: str, text: str) -> list[str]:
    violations = []
    for pattern, reason in FORBIDDEN_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            violations.append(f"[{skill_id}] {reason} (Pattern: {pattern})")
    return violations


def build_hub() -> None:
    manifest = yaml.safe_load(Path("hub-manifest.yaml").read_text(encoding="utf-8"))
    dist_dir = Path("dist/skills")
    if dist_dir.exists():
        shutil.rmtree(dist_dir)
    dist_dir.mkdir(parents=True)

    registry = []
    all_violations = []

    # 合併處理 Public Upstreams 與 Internal Skills
    items = [
        {**item, "origin_type": "public"} for item in manifest.get("public_upstreams", [])
    ] + [
        {**item, "origin_type": "internal"} for item in manifest.get("internal_skills", [])
    ]

    for item in items:
        skill_id = item["id"]
        slug = skill_id.replace("/", "-")
        src_dir = (
            Path("upstream-mirror") / item["source_path"]
            if item["origin_type"] == "public"
            else Path(item["source_path"])
        )
        skill_md_path = src_dir / "SKILL.md"
        raw_md = skill_md_path.read_text(encoding="utf-8")

        # 若為開源技能且配置了企業 Overlay，自動附加至底部
        if item.get("overlay_file"):
            overlay_text = Path(item["overlay_file"]).read_text(encoding="utf-8")
            raw_md = f"{raw_md.rstrip()}\n\n{overlay_text.lstrip()}\n"

        # 執行資安檢疫掃描
        violations = security_audit(skill_id, raw_md)
        if violations:
            all_violations.extend(violations)
            continue

        # 複製附屬目錄 (scripts/, references/, examples/) 並寫入合併後的 SKILL.md
        target_dir = dist_dir / slug
        shutil.copytree(src_dir, target_dir)
        (target_dir / "SKILL.md").write_text(raw_md, encoding="utf-8")

        meta = extract_frontmatter(raw_md)
        sha256_hash = hashlib.sha256(raw_md.encode("utf-8")).hexdigest()[:12]

        registry.append({
            "skill_id": skill_id,
            "slug": slug,
            "name": meta.get("name", slug),
            "origin_type": item["origin_type"],
            "upstream_repo": item.get("upstream_repo", "internal/ai-skill-hub"),
            "locked_commit": item.get("locked_commit", "HEAD"),
            "category": item["category"],
            "governance_tier": item["governance_tier"],
            "owner_team": item["owner_team"],
            "description": str(meta.get("description", "")).strip(),
            "sha256": sha256_hash,
            "dist_path": f"skills/{slug}/SKILL.md",
            "cli_install": f"corp-skills install {skill_id}",
        })

    if all_violations:
        print("🚨 CI 資安檢疫未通過：", file=sys.stderr)
        for v in all_violations:
            print(f"  - {v}", file=sys.stderr)
        sys.exit(1)

    # 1. 輸出供 AI Agent 與內網 CLI 讀取的 skills-hub-registry.json
    Path("docs/skills-hub-registry.json").write_text(
        json.dumps({"version": manifest["hub_version"], "skills": registry}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    # 2. 輸出供人類閱讀的 MkDocs 技能市集總覽頁
    md_lines = [
        "# 企業內部 AI Skill Hub 技能市集目錄",
        "",
        "| 技能 ID | 來源類型 | 分類 | 安全等級 | 負責團隊 | 觸發條件與功能說明 | 內網安裝指令 |",
        "| :--- | :--- | :--- | :--- | :--- | :--- | :--- |",
    ]
    for s in registry:
        badge = "🏢 內部自研" if s["origin_type"] == "internal" else f"🌐 開源認證 (`{s['upstream_repo']}`)"
        md_lines.append(
            f"| **`{s['skill_id']}`** | {badge} | {s['category']} | `{s['governance_tier']}` "
            f"| `{s['owner_team']}` | {s['description']} | `{s['cli_install']}` |"
        )
    Path("docs/marketplace-catalog.md").write_text("\n".join(md_lines) + "\n", encoding="utf-8")
    print(f"✅ 成功編譯 {len(registry)} 個技能包至企業 Skill Hub！")


if __name__ == "__main__":
    build_hub()

"""
Enterprise AI Skill Hub MCP Server (Streamable HTTP / SSE)
部署於內部 K8s 叢集，同時服務全公司工程師 IDE AI 與自動化維運 Agent
"""
from mcp.server.fastmcp import FastMCP
from pathlib import Path
import json

mcp = FastMCP("Enterprise-AI-Skill-Hub")
REGISTRY_FILE = Path("/app/dist/skills-hub-registry.json")
SKILLS_ROOT = Path("/app/dist/skills")


def _get_catalog() -> list[dict]:
    return json.loads(REGISTRY_FILE.read_text(encoding="utf-8")).get("skills", [])


@mcp.tool()
def search_enterprise_skills(
    query: str,
    origin_filter: str = "all",
    max_results: int = 5,
) -> str:
    """
    [Stage 1: Discovery] 搜尋企業內部 AI Skill Hub。
    - origin_filter: 可指定 'internal' (僅公司內部自研)、'public' (僅開源檢疫認證) 或 'all'。
    """
    keywords = [k.lower() for k in query.split() if k.strip()]
    matched = []

    for s in _get_catalog():
        if origin_filter != "all" and s["origin_type"] != origin_filter:
            continue
        haystack = f"{s['skill_id']} {s['category']} {s['description']} {s['owner_team']}".lower()
        score = sum(2 if kw in s["skill_id"].lower() else 1 for kw in keywords if kw in haystack)
        if score > 0:
            matched.append((score, s))

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


@mcp.tool()
def activate_enterprise_skill(skill_id: str) -> str:
    """
    [Stage 2: Activation] 根據 skill_id (如 'corp/sre-victoriametrics-rca' 或 'public/sre-k8s-troubleshooting')，
    載入經過企業資安檢疫與 Overlay 本地化後的完整 SKILL.md 指令內容。
    """
    for s in _get_catalog():
        if s["skill_id"] == skill_id:
            skill_md = SKILLS_ROOT / s["slug"] / "SKILL.md"
            return skill_md.read_text(encoding="utf-8")
    return f"錯誤：在企業 Skill Hub 中找不到技能 '{skill_id}'"


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