exo_skills.watchers

Concrete SkillWatcher implementations for monitoring skill sources and yielding change events.

LocalFileWatcher

from exo_skills import LocalFileWatcher

Watches a local directory for skill file changes using the watchfiles library. Detects additions, modifications, and removals of skill.md / SKILL.md files. Changes are debounced and diffed against a snapshot to produce batches of SkillChangeEvent objects.

Constructor

LocalFileWatcher(path: str | Path, debounce_ms: int = 500)

Methods

watch() -> AsyncIterator[list[SkillChangeEvent]]

Yield batches of skill change events as they occur. Takes an initial snapshot on the first iteration, then watches the directory for changes. The iterator terminates when stop() is called.

watcher = LocalFileWatcher("./skills")
async for batch in watcher.watch():
    for event in batch:
        print(event.kind, event.skill_name)

stop() -> None

Signal the watcher to stop. Causes the awatch loop in watch() to terminate, ending the async iterator.

await watcher.stop()

How It Works

1. On first iteration of watch(), scans the directory tree for skill.md / SKILL.md files and builds an initial snapshot
2. Uses watchfiles.awatch() with a filter that only passes skill file changes
3. On each filesystem event batch, re-scans the directory and diffs against the previous snapshot
4. Yields a list of SkillChangeEvent objects for any added, modified, or removed skills
5. Non-semantic changes (e.g. whitespace-only edits) are detected and silently ignored

Example

import asyncio
from exo_skills import LocalFileWatcher

async def main():
    watcher = LocalFileWatcher("./skills", debounce_ms=300)

    async for batch in watcher.watch():
        for event in batch:
            if event.kind == "added":
                print(f"New skill: {event.skill_name}")
            elif event.kind == "modified":
                print(f"Updated: {event.skill_name}")
            elif event.kind == "removed":
                print(f"Removed: {event.skill_name}")

asyncio.run(main())

GitHubPollingWatcher

from exo_skills import GitHubPollingWatcher

Watches a GitHub repository for skill changes by periodically running git pull --ff-only. The repository is shallow-cloned on first use and cached locally. On each poll cycle, upstream changes are fetched and diffed against a snapshot to produce SkillChangeEvent batches.

Constructor

GitHubPollingWatcher(
    source_url: str,
    poll_interval: float = 300.0,
    cache_dir: Path | None = None,
)

Methods

watch() -> AsyncIterator[list[SkillChangeEvent]]

Yield batches of skill change events as they are detected. Performs an initial shallow clone, takes a snapshot, then enters a poll loop that runs git pull --ff-only at the configured interval. The iterator terminates when stop() is called.

watcher = GitHubPollingWatcher(
    "https://github.com/acme/skills/tree/main/agents",
    poll_interval=60.0,
)
async for batch in watcher.watch():
    for event in batch:
        print(event.kind, event.skill_name)

stop() -> None

Signal the watcher to stop. Sets the internal stop event, causing the poll loop in watch() to terminate on its next cycle.

await watcher.stop()

Supported URL Formats

The watcher parses GitHub URLs of the form:

https://github.com/{owner}/{repo}
https://github.com/{owner}/{repo}/tree/{branch}
https://github.com/{owner}/{repo}/tree/{branch}/{subdir}

If no branch is specified, main is assumed. If a subdirectory is specified, only skills within that subdirectory are scanned.

How It Works

1. Parses the GitHub URL into owner, repo, branch, and optional subdirectory
2. Shallow-clones the repository to {cache_dir}/{owner}/{repo}/{branch} (if not already cached)
3. Takes an initial snapshot of all skill files
4. Enters a poll loop: sleeps for poll_interval seconds, then runs git pull --ff-only
5. After each pull, re-scans and diffs against the previous snapshot
6. Yields SkillChangeEvent batches for any detected changes
7. Failed git pull operations are logged as warnings and the cycle continues

Raises

Example

import asyncio
from exo_skills import GitHubPollingWatcher

async def main():
    watcher = GitHubPollingWatcher(
        "https://github.com/myorg/agent-skills/tree/main/skills",
        poll_interval=120.0,  # Check every 2 minutes
    )

    async for batch in watcher.watch():
        for event in batch:
            print(f"[{event.kind}] {event.skill_name} from {event.source_path}")

asyncio.run(main())

SkillChangeEvent

Defined in exo.skills (exo-core). Both watchers yield lists of this dataclass.

from exo.skills import SkillChangeEvent