Agent Skills for Enterprise KM.

# Repository layout
# .claude/
# └── skills/
#     ├── support/
#     │   ├── refund-resolver.md       # support/refund-resolver
#     │   └── escalation-router.md     # support/escalation-router
#     ├── platform/
#     │   ├── deploy-runbook.md
#     │   └── incident-triage.md
#     ├── data/
#     │   ├── query-builder.md
#     │   └── pii-redactor.md
#     └── finance/
#         └── budget-approval.md       # access_level: sensitive

# Bootstrap script for a fresh repo
import os
TEAMS = ["support", "platform", "data", "growth", "finance"]
for t in TEAMS:
    os.makedirs(f".claude/skills/{t}", exist_ok=True)
    with open(f".claude/skills/{t}/.gitkeep", "w") as f:
        pass
print("namespace-by-team layout ready; commit and start authoring.")

# .claude/skills/_schema.yaml. The frontmatter contract
# Validated by the indexer; PRs that violate this schema fail CI.

required:
  - name              # team/skill-name (e.g. support/refund-resolver)
  - version           # semver: MAJOR.MINOR.PATCH
  - description       # 1-2 sentence description, search-indexed
  - tags              # array, search-indexed
  - access_level      # public | team | role-restricted | sensitive

optional:
  - depends_on        # ['support/case-facts:1.x', ...]
  - deprecated        # 'use support/refund-resolver-v2 instead'
  - owners            # ['@support-team', '@jane.doe']

# Example skill. Support/refund-resolver.md
---
name: support/refund-resolver
version: 1.2.3
description: |
  Resolves customer refund requests up to $500 using the case-facts
  block and escalation queue. For amounts above cap, escalates.
tags: [refund, customer-support, payment]
access_level: team
depends_on:
  - support/case-facts:1.x
  - shared/escalation-queue:2.x
owners:
  - "@support-team"
---

# Body: the actual instructions and examples ...

# scripts/index_skills.py. Runs in CI on push to main
import yaml, json, glob, sys, hashlib, semver
from pathlib import Path

REQUIRED = {"name", "version", "description", "tags", "access_level"}
ACCESS_LEVELS = {"public", "team", "role-restricted", "sensitive"}

def parse(path: Path) -> dict:
    text = path.read_text()
    if not text.startswith("---"):
        raise ValueError(f"{path}: missing frontmatter")
    _, fm, body = text.split("---", 2)
    meta = yaml.safe_load(fm)
    missing = REQUIRED - set(meta)
    if missing:
        raise ValueError(f"{path}: missing keys: {missing}")
    if meta["access_level"] not in ACCESS_LEVELS:
        raise ValueError(f"{path}: bad access_level: {meta['access_level']}")
    semver.VersionInfo.parse(meta["version"])  # raises if invalid
    meta["body_hash"] = hashlib.sha256(body.encode()).hexdigest()[:12]
    meta["path"] = str(path)
    return meta

def build_index() -> list[dict]:
    skills = [parse(Path(p)) for p in glob.glob(".claude/skills/**/*.md", recursive=True)]
    # Resolve dependencies. Every depends_on must exist
    names = {s["name"] for s in skills}
    for s in skills:
        for dep in s.get("depends_on", []):
            dep_name = dep.split(":")[0]
            if dep_name not in names:
                raise ValueError(f"{s['name']}: missing dep {dep_name}")
    return skills

if __name__ == "__main__":
    try:
        index = build_index()
        Path("dist/skill-registry.json").write_text(json.dumps(index, indent=2))
        print(f"indexed {len(index)} skills; pushed to registry")
    except ValueError as e:
        print(f"::error::{e}", file=sys.stderr)
        sys.exit(1)

# scripts/search_service.py
from anthropic import Anthropic
import numpy as np, json
from pathlib import Path

# Index loaded from registry
SKILLS = json.loads(Path("dist/skill-registry.json").read_text())

# Embedding cache keyed by (skill name, body_hash)
_emb_cache: dict[tuple[str, str], list[float]] = {}

def embed_text(text: str) -> list[float]:
    """Stand-in for any embeddings provider (Voyage, OpenAI, etc.)."""
    # In production, batch-embed once at index time and cache:
    # client.embed(text, model="voyage-2-large")
    raise NotImplementedError

def index_skill(s: dict):
    key = (s["name"], s["body_hash"])
    if key not in _emb_cache:
        text = f"{s['description']} {' '.join(s['tags'])}"
        _emb_cache[key] = embed_text(text)

def cosine(a, b):
    a, b = np.array(a), np.array(b)
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

def search(query: str, k: int = 5,
           team: str | None = None,
           access_level: str | None = None) -> list[dict]:
    qv = embed_text(query)
    scored = []
    for s in SKILLS:
        if team and not s["name"].startswith(f"{team}/"):
            continue
        if access_level and s["access_level"] != access_level:
            continue
        index_skill(s)
        sim = cosine(qv, _emb_cache[(s["name"], s["body_hash"])])
        scored.append((sim, s))
    scored.sort(key=lambda x: -x[0])
    return [
        {**s, "score": round(score, 3)}
        for score, s in scored[:k]
    ]

# Example
# results = search("handle a customer refund up to $500", k=3, team="support")

# scripts/resolve_deps.py. Given a Skill, resolve its depends_on graph
import json, semver
from pathlib import Path

SKILLS = json.loads(Path("dist/skill-registry.json").read_text())
INDEX = {s["name"]: [] for s in SKILLS}
for s in SKILLS:
    INDEX[s["name"]].append(s)
for name in INDEX:
    INDEX[name].sort(key=lambda s: semver.VersionInfo.parse(s["version"]))

def resolve(spec: str) -> dict:
    """spec: 'team/skill:1.x' or 'team/skill:>=2.0.0 <3.0.0'."""
    name, _, constraint = spec.partition(":")
    versions = INDEX.get(name, [])
    if not versions:
        raise LookupError(f"unknown skill: {name}")

    if constraint.endswith(".x"):
        major = int(constraint.split(".")[0])
        candidates = [
            v for v in versions
            if semver.VersionInfo.parse(v["version"]).major == major
        ]
    else:
        candidates = [
            v for v in versions
            if semver.match(v["version"], constraint)
        ]

    if not candidates:
        raise LookupError(f"{name}: no version satisfies {constraint}")
    return candidates[-1]  # latest matching

def topo_resolve(skill_spec: str, seen: set | None = None) -> list[dict]:
    """Resolve full dep graph in topological order."""
    seen = seen or set()
    skill = resolve(skill_spec)
    if skill["name"] in seen:
        return []
    seen.add(skill["name"])
    out = []
    for dep_spec in skill.get("depends_on", []):
        out.extend(topo_resolve(dep_spec, seen))
    out.append(skill)
    return out

# scripts/acl_gate.py
from typing import TypedDict, Literal

class Skill(TypedDict):
    name: str
    access_level: Literal["public", "team", "role-restricted", "sensitive"]
    owners: list[str]

class AgentContext(TypedDict):
    role: str           # e.g. 'support-agent', 'finance-agent'
    teams: list[str]    # ['support', 'shared']
    elevated: bool      # has the user explicitly elevated to invoke sensitive skills?

ROLE_ACL = {
    # Each access_level → which roles may invoke
    "public": lambda ctx, s: True,
    "team": lambda ctx, s: any(s["name"].startswith(f"{t}/") for t in ctx["teams"]),
    "role-restricted": lambda ctx, s: ctx["role"] in s.get("allowed_roles", []),
    "sensitive": lambda ctx, s: ctx["elevated"] and any(
        s["name"].startswith(f"{t}/") for t in ctx["teams"]
    ),
}

def check(ctx: AgentContext, skill: Skill) -> dict:
    """Returns {allowed, reason, request_url?}."""
    rule = ROLE_ACL[skill["access_level"]]
    if rule(ctx, skill):
        return {"allowed": True, "reason": "access_granted"}
    return {
        "allowed": False,
        "reason": f"agent role={ctx['role']} cannot invoke {skill['name']} (access_level={skill['access_level']})",
        "request_url": f"https://internal.example.com/skills/request-access?skill={skill['name']}",
    }

# Usage in the agent loop
def invoke_skill(ctx: AgentContext, skill_spec: str, payload: dict):
    from resolve_deps import resolve
    skill = resolve(skill_spec)
    decision = check(ctx, skill)
    if not decision["allowed"]:
        return {"error": "ACL_DENIED", **decision}
    # ...actually invoke the Skill body...

# Skills are exposed as two tools to every agent
TOOLS = [
    {
        "name": "search_skills",
        "description": (
            "Find a Skill in the enterprise registry by natural-language query. "
            "Returns up to k matches with name, version, description, score. "
            "Use BEFORE invoke_skill so you have a name+version to invoke."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "k": {"type": "integer", "default": 5},
                "team": {"type": "string"},  # optional filter
                "access_level": {"type": "string"},  # optional filter
            },
            "required": ["query"],
        },
    },
    {
        "name": "invoke_skill",
        "description": (
            "Invoke a Skill from the registry. ACL is checked before the "
            "Skill body executes; if denied, returns ACL_DENIED with a "
            "request_url for access. Always pin a major version (e.g. 1.x)."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "version_constraint": {"type": "string", "default": "*"},
                "payload": {"type": "object"},
            },
            "required": ["name", "payload"],
        },
    },
]

# scripts/usage_tracker.py
from datetime import datetime, timedelta
from collections import Counter
import json
from pathlib import Path

# Append-only log of every invoke_skill call
def log_invocation(name: str, version: str, agent_role: str, outcome: str):
    record = {
        "ts": datetime.utcnow().isoformat() + "Z",
        "name": name,
        "version": version,
        "agent_role": agent_role,
        "outcome": outcome,  # 'success' | 'acl_denied' | 'error'
    }
    with open("logs/skill-invocations.jsonl", "a") as f:
        f.write(json.dumps(record) + "\n")

# Nightly: surface deprecation candidates + hot Skills
def nightly_report():
    cutoff = datetime.utcnow() - timedelta(days=180)
    invocations = [
        json.loads(line)
        for line in Path("logs/skill-invocations.jsonl").read_text().splitlines()
    ]
    recent = [r for r in invocations if datetime.fromisoformat(r["ts"][:-1]) > cutoff]
    hot = Counter((r["name"], r["version"]) for r in recent).most_common(20)
    invoked_names = {r["name"] for r in recent}
    all_names = {s["name"] for s in json.loads(Path("dist/skill-registry.json").read_text())}
    cold = sorted(all_names - invoked_names)

    print("=== Hot Skills (last 180d) ===")
    for (name, ver), count in hot:
        print(f"  {name}:{ver}  {count}")
    print(f"\n=== Cold Skills (deprecation candidates) ({len(cold)}) ===")
    for name in cold:
        print(f"  {name}")