Skip to content

Registries API Reference

A registry materializes skill packages into a local directory that SkillsCapability hands to harness. See Skill Registries for the guide.

Bases: ABC

Abstract base for skill registries.

Implement sync to fetch skill packages and lay them out as immediate child directories of a returned library path. Nothing else is required — parsing, validation and instruction rendering all belong to harness.

Convenience methods :meth:filtered, :meth:prefixed, and :meth:renamed return lightweight wrapper views; the underlying registry is never modified.

Source code in pydantic_ai_skills/registries/_base.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class SkillRegistry(ABC):
    """Abstract base for skill registries.

    Implement [`sync`][pydantic_ai_skills.SkillRegistry.sync] to fetch skill packages and
    lay them out as immediate child directories of a returned library path. Nothing else
    is required — parsing, validation and instruction rendering all belong to harness.

    Convenience methods :meth:`filtered`, :meth:`prefixed`, and :meth:`renamed` return
    lightweight wrapper views; the underlying registry is never modified.
    """

    @abstractmethod
    def sync(self) -> Path:
        """Materialize this registry's skills and return the local library directory.

        The returned path is a *library*: its immediate children are skill package
        directories, each holding a `SKILL.md`. It is passed straight to harness's
        `Skills`, so it must satisfy harness's rules — in particular the library itself
        must not contain a `SKILL.md`.

        Implementations should be idempotent and safe to call repeatedly: a second call
        refreshes the local copy (a `git pull`, a re-sync) rather than starting over.

        Returns:
            Path to the local skill-library directory.
        """

    def skill_infos(self) -> list[SkillInfo]:
        """Return the catalog fields of every skill package in this registry.

        Syncs first, then reads each immediate child's `SKILL.md`. Used by
        [`filtered`][pydantic_ai_skills.SkillRegistry.filtered] and by callers that want
        to know what a registry holds without constructing an agent.

        Returns:
            One [`SkillInfo`][pydantic_ai_skills._parsing.SkillInfo] per package, sorted
            by name.
        """
        library = self.sync()
        infos = [read_skill_info(child) for child in sorted(library.iterdir()) if child.is_dir()]
        return [info for info in infos if info is not None]

    def skill_names(self) -> list[str]:
        """Return the names of every skill package in this registry, sorted."""
        return [info.name for info in self.skill_infos()]

    def filtered(self, predicate: Callable[[SkillInfo], bool]) -> FilteredRegistry:
        """Return a view of this registry limited to skills matching ``predicate``.

        Args:
            predicate: A callable that accepts a
                [`SkillInfo`][pydantic_ai_skills._parsing.SkillInfo] and returns ``True``
                if the skill should be included.

        Returns:
            A :class:`~pydantic_ai_skills.registries.filtered.FilteredRegistry`
            view backed by the same underlying source.
        """
        from pydantic_ai_skills.registries.filtered import FilteredRegistry as _Filtered

        return _Filtered(wrapped=self, predicate=predicate)

    def prefixed(self, prefix: str) -> PrefixedRegistry:
        """Return a view of this registry with ``prefix`` prepended to every skill name.

        Args:
            prefix: String to prepend to every skill name. The result must still be a
                valid skill name, so a prefix normally ends with a hyphen.

        Returns:
            A :class:`~pydantic_ai_skills.registries.prefixed.PrefixedRegistry`
            view backed by the same underlying source.
        """
        from pydantic_ai_skills.registries.prefixed import PrefixedRegistry as _Prefixed

        return _Prefixed(wrapped=self, prefix=prefix)

    def renamed(self, name_map: dict[str, str]) -> RenamedRegistry:
        """Return a view of this registry with skills renamed per ``name_map``.

        Args:
            name_map: Mapping of ``{new_name: original_name}``.

        Returns:
            A :class:`~pydantic_ai_skills.registries.renamed.RenamedRegistry`
            view backed by the same underlying source.
        """
        from pydantic_ai_skills.registries.renamed import RenamedRegistry as _Renamed

        return _Renamed(wrapped=self, name_map=name_map)

    def __or__(self, other: SkillRegistry) -> CombinedRegistry:
        """Return a registry that merges this one with ``other``.

        Earlier registries win on a duplicate skill name, matching
        :class:`~pydantic_ai_skills.registries.combined.CombinedRegistry`.
        """
        from pydantic_ai_skills.registries.combined import CombinedRegistry as _Combined

        return _Combined(registries=[self, other])

sync abstractmethod 

sync() -> Path

Materialize this registry's skills and return the local library directory.

The returned path is a library: its immediate children are skill package directories, each holding a SKILL.md. It is passed straight to harness's Skills, so it must satisfy harness's rules — in particular the library itself must not contain a SKILL.md.

Implementations should be idempotent and safe to call repeatedly: a second call refreshes the local copy (a git pull, a re-sync) rather than starting over.

Returns:

Type Description
Path

Path to the local skill-library directory.
Source code in pydantic_ai_skills/registries/_base.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@abstractmethod
def sync(self) -> Path:
    """Materialize this registry's skills and return the local library directory.

    The returned path is a *library*: its immediate children are skill package
    directories, each holding a `SKILL.md`. It is passed straight to harness's
    `Skills`, so it must satisfy harness's rules — in particular the library itself
    must not contain a `SKILL.md`.

    Implementations should be idempotent and safe to call repeatedly: a second call
    refreshes the local copy (a `git pull`, a re-sync) rather than starting over.

    Returns:
        Path to the local skill-library directory.
    """

skill_infos 

skill_infos() -> list[SkillInfo]

Return the catalog fields of every skill package in this registry.

Syncs first, then reads each immediate child's SKILL.md. Used by filtered and by callers that want to know what a registry holds without constructing an agent.

Returns:

Type Description
list[SkillInfo]

One SkillInfo per package, sorted
list[SkillInfo]

by name.
Source code in pydantic_ai_skills/registries/_base.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def skill_infos(self) -> list[SkillInfo]:
    """Return the catalog fields of every skill package in this registry.

    Syncs first, then reads each immediate child's `SKILL.md`. Used by
    [`filtered`][pydantic_ai_skills.SkillRegistry.filtered] and by callers that want
    to know what a registry holds without constructing an agent.

    Returns:
        One [`SkillInfo`][pydantic_ai_skills._parsing.SkillInfo] per package, sorted
        by name.
    """
    library = self.sync()
    infos = [read_skill_info(child) for child in sorted(library.iterdir()) if child.is_dir()]
    return [info for info in infos if info is not None]

skill_names 

skill_names() -> list[str]

Return the names of every skill package in this registry, sorted.

Source code in pydantic_ai_skills/registries/_base.py 
82
83
84
def skill_names(self) -> list[str]:
    """Return the names of every skill package in this registry, sorted."""
    return [info.name for info in self.skill_infos()]

filtered 

filtered(predicate: Callable[[SkillInfo], bool]) -> FilteredRegistry

Return a view of this registry limited to skills matching predicate.

Parameters:

Name Type Description Default
predicate Callable[[SkillInfo], bool]

A callable that accepts a SkillInfo and returns True if the skill should be included.

 required

Returns:

Name Type Description
A FilteredRegistry
FilteredRegistry

view backed by the same underlying source.
Source code in pydantic_ai_skills/registries/_base.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def filtered(self, predicate: Callable[[SkillInfo], bool]) -> FilteredRegistry:
    """Return a view of this registry limited to skills matching ``predicate``.

    Args:
        predicate: A callable that accepts a
            [`SkillInfo`][pydantic_ai_skills._parsing.SkillInfo] and returns ``True``
            if the skill should be included.

    Returns:
        A :class:`~pydantic_ai_skills.registries.filtered.FilteredRegistry`
        view backed by the same underlying source.
    """
    from pydantic_ai_skills.registries.filtered import FilteredRegistry as _Filtered

    return _Filtered(wrapped=self, predicate=predicate)

prefixed 

prefixed(prefix: str) -> PrefixedRegistry

Return a view of this registry with prefix prepended to every skill name.

Parameters:

Name Type Description Default
prefix str

String to prepend to every skill name. The result must still be a valid skill name, so a prefix normally ends with a hyphen.

 required

Returns:

Name Type Description
A PrefixedRegistry
PrefixedRegistry

view backed by the same underlying source.
Source code in pydantic_ai_skills/registries/_base.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def prefixed(self, prefix: str) -> PrefixedRegistry:
    """Return a view of this registry with ``prefix`` prepended to every skill name.

    Args:
        prefix: String to prepend to every skill name. The result must still be a
            valid skill name, so a prefix normally ends with a hyphen.

    Returns:
        A :class:`~pydantic_ai_skills.registries.prefixed.PrefixedRegistry`
        view backed by the same underlying source.
    """
    from pydantic_ai_skills.registries.prefixed import PrefixedRegistry as _Prefixed

    return _Prefixed(wrapped=self, prefix=prefix)

renamed 

renamed(name_map: dict[str, str]) -> RenamedRegistry

Return a view of this registry with skills renamed per name_map.

Parameters:

Name Type Description Default
name_map dict[str, str]

Mapping of {new_name: original_name}.

 required

Returns:

Name Type Description
A RenamedRegistry
RenamedRegistry

view backed by the same underlying source.
Source code in pydantic_ai_skills/registries/_base.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
def renamed(self, name_map: dict[str, str]) -> RenamedRegistry:
    """Return a view of this registry with skills renamed per ``name_map``.

    Args:
        name_map: Mapping of ``{new_name: original_name}``.

    Returns:
        A :class:`~pydantic_ai_skills.registries.renamed.RenamedRegistry`
        view backed by the same underlying source.
    """
    from pydantic_ai_skills.registries.renamed import RenamedRegistry as _Renamed

    return _Renamed(wrapped=self, name_map=name_map)

__or__ 

__or__(other: SkillRegistry) -> CombinedRegistry

Return a registry that merges this one with other.

Earlier registries win on a duplicate skill name, matching :class:~pydantic_ai_skills.registries.combined.CombinedRegistry.

Source code in pydantic_ai_skills/registries/_base.py 
131
132
133
134
135
136
137
138
139
def __or__(self, other: SkillRegistry) -> CombinedRegistry:
    """Return a registry that merges this one with ``other``.

    Earlier registries win on a duplicate skill name, matching
    :class:`~pydantic_ai_skills.registries.combined.CombinedRegistry`.
    """
    from pydantic_ai_skills.registries.combined import CombinedRegistry as _Combined

    return _Combined(registries=[self, other])

The catalog fields of one skill package, as seen before harness validates it.

This is what a FilteredRegistry predicate receives. It is deliberately shallow — no bundled files, no instructions body — because filtering happens while staging directories, well before any skill is handed to an agent.

Attributes:

Name Type Description
name str

The package's directory name, NFKC-normalized. This, not the frontmatter name, is what harness will call the skill.
description str

The frontmatter description, or an empty string when the file has none. harness rejects a missing description later; filtering does not.
directory Path

The package directory.
Source code in pydantic_ai_skills/_parsing.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@dataclass(frozen=True)
class SkillInfo:
    """The catalog fields of one skill package, as seen before harness validates it.

    This is what a [`FilteredRegistry`][pydantic_ai_skills.registries.FilteredRegistry]
    predicate receives. It is deliberately shallow — no bundled files, no instructions
    body — because filtering happens while staging directories, well before any skill is
    handed to an agent.

    Attributes:
        name: The package's directory name, NFKC-normalized. This, not the frontmatter
            `name`, is what harness will call the skill.
        description: The frontmatter `description`, or an empty string when the file has
            none. harness rejects a missing description later; filtering does not.
        directory: The package directory.
    """

    name: str
    description: str
    directory: Path

Bases: SkillRegistry

Skills registry backed by a Git repository cloned with GitPython.

:meth:sync clones the repository on first call and performs a git pull on subsequent ones (or a full re-clone if the local copy is corrupted or missing), then returns the directory holding the skill packages.

The registry only reads the filesystem after cloning — it never calls any hosting platform's REST/GraphQL API — so it works with any git host accessible over HTTPS or SSH (GitHub, GitLab, Bitbucket, self-hosted, etc.).

It does not parse SKILL.md: the directory it produces is handed to :class:~pydantic_ai_skills.SkillsCapability, and validating and rendering the packages inside it is pydantic-ai-harness's job.

Parameters:

Name Type Description Default
repo_url str

Full URL of the Git repository to clone (e.g. "https://github.com/anthropics/skills"). Works with any Git host accessible over HTTPS or SSH (GitHub, GitLab, Bitbucket, self-hosted, etc.).

 required
target_dir str | Path | None

Local directory where the repository is cloned. Defaults to a temporary directory scoped to the registry instance. A directory you pass persists across :meth:sync calls and is not cleaned up automatically — callers own the lifecycle.

 None
path str

Sub-path inside the repository that contains the skill directories. Defaults to the repository root (""). For example, pass "skills" when skills live at owner/name/skills/<skill>/.

 ''
token str | None

Personal access token (or any HTTPS password) used for authentication. When None the registry falls back to the GITHUB_TOKEN environment variable. Anonymous access is used when neither is set (rate-limited for public repos, fails for private ones).

 None
ssh_key_file str | Path | None

Path to a private SSH key for SSH-based authentication. When provided, GIT_SSH_COMMAND is injected into clone_options.env.

 None
clone_options GitCloneOptions | None

Fine-grained GitPython configuration. See :class:GitCloneOptions for the full list of knobs. Any value set here is forwarded verbatim to git.Repo.clone_from / repo.remotes.origin.pull.

 None
auto_install bool

When True (default), :meth:sync clones or pulls so the local copy is up to date. Set to False to read only what is already on disk, which is what offline or air-gapped environments want.

 True

Examples:

Basic usage — clone a repository and expose all its skills:

from pydantic_ai_skills import GitSkillsRegistry, SkillsCapability

capability = SkillsCapability(
    registries=[
        GitSkillsRegistry(
            repo_url="https://github.com/anthropics/skills",
            path="skills",
            target_dir="./cached-skills",
        ),
    ]
)

Blobless shallow clone with a PAT, only the pdf sub-path:

from pydantic_ai_skills.registries.git import GitSkillsRegistry, GitCloneOptions

registry = GitSkillsRegistry(
    repo_url="https://github.com/anthropics/skills",
    path="skills/pdf",
    token="ghp_...",
    clone_options=GitCloneOptions(
        depth=1,
        single_branch=True,
        sparse_paths=["skills/pdf"],
        multi_options=["--filter=blob:none"],
    ),
)

Filter to only PDF-related skills:

pdf_registry = registry.filtered(lambda info: "pdf" in info.name)

Prefix all skill names from this registry:

prefixed_registry = registry.prefixed("anthropic-")
# "pdf" skill is now accessible as "anthropic-pdf"

SSH authentication with a custom key:

registry = GitSkillsRegistry(
    repo_url="git@github.com:my-org/private-skills.git",
    ssh_key_file="~/.ssh/id_ed25519_skills",
)

Offline / air-gapped — pre-clone manually, disable auto-install so :meth:sync never reaches the network:

registry = GitSkillsRegistry(
    repo_url="https://github.com/anthropics/skills",
    target_dir="/opt/skills-mirror",
    auto_install=False,
)
Source code in pydantic_ai_skills/registries/git.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
class GitSkillsRegistry(SkillRegistry):
    """Skills registry backed by a Git repository cloned with GitPython.

    :meth:`sync` clones the repository on first call and performs a ``git pull`` on
    subsequent ones (or a full re-clone if the local copy is corrupted or missing), then
    returns the directory holding the skill packages.

    The registry only reads the filesystem after cloning — it never calls any
    hosting platform's REST/GraphQL API — so it works with any git host
    accessible over HTTPS or SSH (GitHub, GitLab, Bitbucket, self-hosted, etc.).

    It does not parse ``SKILL.md``: the directory it produces is handed to
    :class:`~pydantic_ai_skills.SkillsCapability`, and validating and rendering the
    packages inside it is `pydantic-ai-harness`'s job.

    Args:
        repo_url: Full URL of the Git repository to clone (e.g.
            ``"https://github.com/anthropics/skills"``). Works with any Git host
            accessible over HTTPS or SSH (GitHub, GitLab, Bitbucket,
            self-hosted, etc.).
        target_dir: Local directory where the repository is cloned. Defaults to
            a temporary directory scoped to the registry instance. A directory you
            pass persists across :meth:`sync` calls and is **not** cleaned up
            automatically — callers own the lifecycle.
        path: Sub-path inside the repository that contains the skill directories.
            Defaults to the repository root (``""``). For example, pass
            ``"skills"`` when skills live at ``owner/name/skills/<skill>/``.
        token: Personal access token (or any HTTPS password) used for
            authentication. When ``None`` the registry falls back to the
            ``GITHUB_TOKEN`` environment variable. Anonymous access is used when
            neither is set (rate-limited for public repos, fails for private ones).
        ssh_key_file: Path to a private SSH key for SSH-based authentication.
            When provided, ``GIT_SSH_COMMAND`` is injected into
            ``clone_options.env``.
        clone_options: Fine-grained GitPython configuration. See
            :class:`GitCloneOptions` for the full list of knobs. Any value set
            here is forwarded verbatim to ``git.Repo.clone_from`` /
            ``repo.remotes.origin.pull``.
        auto_install: When ``True`` (default), :meth:`sync` clones or pulls so the local
            copy is up to date. Set to ``False`` to read only what is already on disk,
            which is what offline or air-gapped environments want.

    Examples:
        Basic usage — clone a repository and expose all its skills:

        ```python
        from pydantic_ai_skills import GitSkillsRegistry, SkillsCapability

        capability = SkillsCapability(
            registries=[
                GitSkillsRegistry(
                    repo_url="https://github.com/anthropics/skills",
                    path="skills",
                    target_dir="./cached-skills",
                ),
            ]
        )
        ```

        Blobless shallow clone with a PAT, only the ``pdf`` sub-path:

        ```python
        from pydantic_ai_skills.registries.git import GitSkillsRegistry, GitCloneOptions

        registry = GitSkillsRegistry(
            repo_url="https://github.com/anthropics/skills",
            path="skills/pdf",
            token="ghp_...",
            clone_options=GitCloneOptions(
                depth=1,
                single_branch=True,
                sparse_paths=["skills/pdf"],
                multi_options=["--filter=blob:none"],
            ),
        )
        ```

        Filter to only PDF-related skills:

        ```python
        pdf_registry = registry.filtered(lambda info: "pdf" in info.name)
        ```

        Prefix all skill names from this registry:

        ```python
        prefixed_registry = registry.prefixed("anthropic-")
        # "pdf" skill is now accessible as "anthropic-pdf"
        ```

        SSH authentication with a custom key:

        ```python
        registry = GitSkillsRegistry(
            repo_url="git@github.com:my-org/private-skills.git",
            ssh_key_file="~/.ssh/id_ed25519_skills",
        )
        ```

        Offline / air-gapped — pre-clone manually, disable auto-install so
        :meth:`sync` never reaches the network:

        ```python
        registry = GitSkillsRegistry(
            repo_url="https://github.com/anthropics/skills",
            target_dir="/opt/skills-mirror",
            auto_install=False,
        )
        ```
    """

    def __init__(
        self,
        repo_url: str,
        *,
        target_dir: str | Path | None = None,
        path: str = '',
        token: str | None = None,
        ssh_key_file: str | Path | None = None,
        clone_options: GitCloneOptions | None = None,
        auto_install: bool = True,
    ) -> None:
        try:
            import git as _git  # noqa: F401
        except ImportError as exc:
            raise ImportError(
                'GitPython is required for GitSkillsRegistry. Install it with: pip install pydantic-ai-skills[git]'
            ) from exc

        self._repo_url = repo_url
        self._path = path.strip('/')
        self._auto_install = auto_install
        self._clone_options = clone_options or GitCloneOptions()
        self._tmp_dir: tempfile.TemporaryDirectory[str] | None = None

        # Resolve effective token (explicit arg beats env var)
        effective_token = token or os.environ.get('GITHUB_TOKEN')
        self._token: str | None = effective_token  # kept private for masking

        # Build the URL used for cloning (with token embedded if available)
        if effective_token:
            self._clone_url = _inject_token_into_url(repo_url, effective_token)
        else:
            self._clone_url = repo_url

        # Resolve target directory
        if target_dir is None:
            self._tmp_dir = tempfile.TemporaryDirectory()
            self._target_dir = Path(self._tmp_dir.name)
        else:
            self._target_dir = Path(target_dir).expanduser().resolve()

        # SSH key handling
        if ssh_key_file is not None:
            key_path = Path(ssh_key_file).expanduser().resolve()
            # Warn if permissions are wider than 0o600
            try:
                key_stat = key_path.stat()
                if key_stat.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
                    warnings.warn(
                        f"SSH key file '{key_path}' has permissions wider than 0o600. "
                        'Consider restricting with: chmod 600 '
                        f'{key_path}',
                        UserWarning,
                        stacklevel=2,
                    )
            except OSError:
                pass
            # Use accept-new to avoid disabling host key checking entirely while still
            # allowing non-interactive first-time connections.
            self._clone_options.env['GIT_SSH_COMMAND'] = f'ssh -i {key_path} -o StrictHostKeyChecking=accept-new'

        # Clean repo URL (no credentials) for display and errors
        self._clean_repo_url = _sanitize_url(repo_url)

    # ------------------------------------------------------------------
    # repr — never expose the token
    # ------------------------------------------------------------------

    def __repr__(self) -> str:
        return (
            f'{type(self).__name__}('
            f'repo_url={self._clean_repo_url!r}, '
            f'path={self._path!r}, '
            f'target_dir={str(self._target_dir)!r})'
        )

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _skills_root(self) -> Path:
        """Return the path inside the clone where skill directories live."""
        if self._path:
            return self._target_dir / self._path
        return self._target_dir

    def _is_cloned(self) -> bool:
        """Return True if a valid git repository already exists in the target dir."""
        import git

        if not self._target_dir.exists():
            return False
        try:
            git.Repo(str(self._target_dir))
            return True
        except git.exc.InvalidGitRepositoryError:
            return False

    def _clone(self) -> None:
        """Clone the repository into the target directory."""
        import git

        opts = self._clone_options
        clone_kwargs: dict[str, Any] = {}

        if opts.depth is not None:
            clone_kwargs['depth'] = opts.depth
        if opts.branch is not None:
            clone_kwargs['branch'] = opts.branch
        if opts.single_branch:
            clone_kwargs['single_branch'] = True
        if opts.multi_options:
            clone_kwargs['multi_options'] = opts.multi_options
        if opts.env:
            clone_kwargs['env'] = opts.env

        clone_kwargs.update(opts.git_options)

        self._target_dir.mkdir(parents=True, exist_ok=True)

        try:
            repo = git.Repo.clone_from(
                self._clone_url,
                str(self._target_dir),
                **clone_kwargs,
            )
        except git.exc.GitCommandError as exc:
            sanitized = _sanitize_error_message(exc, self._clone_url, self._clean_repo_url)
            raise RuntimeError(f'Failed to clone repository {self._clean_repo_url!r}: {sanitized}') from exc

        # Apply sparse checkout if requested
        if opts.sparse_paths:
            try:
                repo.git.sparse_checkout('init')
                repo.git.sparse_checkout('set', *opts.sparse_paths)
            except git.exc.GitCommandError as exc:
                sanitized = _sanitize_error_message(exc, self._clone_url, self._clean_repo_url)
                raise RuntimeError(f'Failed to configure sparse checkout: {sanitized}') from exc

    def _pull(self) -> None:
        """Perform ``git pull`` on the existing clone."""
        import git

        pull_kwargs: dict[str, Any] = {}
        if self._clone_options.env:
            pull_kwargs['env'] = self._clone_options.env
        pull_kwargs.update(self._clone_options.git_options)

        try:
            repo = git.Repo(str(self._target_dir))
            repo.remotes.origin.pull(**pull_kwargs)
        except git.exc.InvalidGitRepositoryError:
            # Clone is corrupted or missing — start fresh
            shutil.rmtree(str(self._target_dir), ignore_errors=True)
            self._clone()
        except git.exc.GitCommandError as exc:
            sanitized = _sanitize_error_message(exc, self._clone_url, self._clean_repo_url)
            raise RuntimeError(f'Failed to pull latest changes from {self._clean_repo_url!r}: {sanitized}') from exc

    def _ensure_cloned(self) -> None:
        """Clone or pull the repository to ensure the local cache is up to date."""
        if self._is_cloned():
            self._pull()
        else:
            self._clone()

    def _revision(self) -> str | None:
        """Return the current HEAD commit SHA, or None on failure."""
        import git

        try:
            repo = git.Repo(str(self._target_dir))
            return repo.head.commit.hexsha
        except (OSError, ValueError, git.exc.InvalidGitRepositoryError, git.exc.GitCommandError):
            return None

    # ------------------------------------------------------------------
    # SkillRegistry interface
    # ------------------------------------------------------------------

    def sync(self) -> Path:
        """Clone or pull the repository and return its skill-library directory.

        The returned path is ``target_dir`` joined with ``path``, whose immediate children
        are the skill packages. With ``auto_install=False`` nothing is fetched and
        whatever is already on disk is returned, which is what an air-gapped deployment
        wants.

        Returns:
            Path to the local skill-library directory.

        Raises:
            RuntimeError: On git or network errors.
            ValueError: When the configured ``path`` does not exist in the clone -- the
                usual cause is a ``path`` that does not match the repository's layout.
        """
        if self._auto_install:
            self._ensure_cloned()

        skills_root = self._skills_root()
        if not skills_root.is_dir():
            # Distinguish the two causes: a clone that never happened, versus a clone that
            # did but has no such sub-path. Reporting the first for both would send a
            # caller with a mistyped `path` looking for a network problem.
            if not self._target_dir.is_dir():
                detail = 'the repository has not been cloned yet and auto_install is disabled'
            else:
                detail = f'path={self._path!r} does not exist in the repository'
            raise ValueError(f'No skill library at {skills_root} for {self._clean_repo_url!r}: {detail}.')
        return skills_root

    def revision(self) -> str | None:
        """Return the commit SHA the local clone is on, or None if it is not cloned.

        Useful for recording exactly which version of a remote skill library an agent ran
        with, since :meth:`sync` otherwise tracks a moving branch.
        """
        return self._revision()

sync 

sync() -> Path

Clone or pull the repository and return its skill-library directory.

The returned path is target_dir joined with path, whose immediate children are the skill packages. With auto_install=False nothing is fetched and whatever is already on disk is returned, which is what an air-gapped deployment wants.

Returns:

Type Description
Path

Path to the local skill-library directory.

Raises:

Type Description
RuntimeError

On git or network errors.
ValueError

When the configured path does not exist in the clone -- the usual cause is a path that does not match the repository's layout.
Source code in pydantic_ai_skills/registries/git.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
def sync(self) -> Path:
    """Clone or pull the repository and return its skill-library directory.

    The returned path is ``target_dir`` joined with ``path``, whose immediate children
    are the skill packages. With ``auto_install=False`` nothing is fetched and
    whatever is already on disk is returned, which is what an air-gapped deployment
    wants.

    Returns:
        Path to the local skill-library directory.

    Raises:
        RuntimeError: On git or network errors.
        ValueError: When the configured ``path`` does not exist in the clone -- the
            usual cause is a ``path`` that does not match the repository's layout.
    """
    if self._auto_install:
        self._ensure_cloned()

    skills_root = self._skills_root()
    if not skills_root.is_dir():
        # Distinguish the two causes: a clone that never happened, versus a clone that
        # did but has no such sub-path. Reporting the first for both would send a
        # caller with a mistyped `path` looking for a network problem.
        if not self._target_dir.is_dir():
            detail = 'the repository has not been cloned yet and auto_install is disabled'
        else:
            detail = f'path={self._path!r} does not exist in the repository'
        raise ValueError(f'No skill library at {skills_root} for {self._clean_repo_url!r}: {detail}.')
    return skills_root

revision 

revision() -> str | None

Return the commit SHA the local clone is on, or None if it is not cloned.

Useful for recording exactly which version of a remote skill library an agent ran with, since :meth:sync otherwise tracks a moving branch.

Source code in pydantic_ai_skills/registries/git.py 
445
446
447
448
449
450
451
def revision(self) -> str | None:
    """Return the commit SHA the local clone is on, or None if it is not cloned.

    Useful for recording exactly which version of a remote skill library an agent ran
    with, since :meth:`sync` otherwise tracks a moving branch.
    """
    return self._revision()

Low-level GitPython configuration for clone and fetch operations.

All fields map directly to arguments accepted by git.Repo.clone_from or git.Remote.fetch / git.Remote.pull, so developers who know GitPython can use the full API without any wrapper layer.

Parameters:

Name Type Description Default
depth int | None

Create a shallow clone with history truncated to this many commits. Passed as --depth to git. None means a full clone. Useful for large repositories where only the latest snapshot is needed.

 None
branch str | None

Name of the remote branch, tag, or ref to check out after cloning (--branch flag). Defaults to the repository's default branch when None.

 None
single_branch bool

When True, clone only the branch specified by branch (--single-branch). Has no effect when branch is None.

 False
sparse_paths list[str]

List of path patterns to include in a sparse checkout (--sparse + git sparse-checkout set). An empty list disables sparse checkout and fetches the full tree.

 list()
env dict[str, str]

Mapping of environment variables forwarded to every git sub-process (e.g. GIT_SSH_COMMAND, GIT_ASKPASS). These override the process environment for git calls only.

 dict()
multi_options list[str]

Extra --option strings passed verbatim to git.Repo.clone_from(multi_options=...). Use for git options not exposed by other fields (e.g. ['--filter=blob:none'] for a partial/blobless clone).

 list()
git_options dict[str, Any]

Mapping forwarded as keyword arguments directly to git.Repo.clone_from or repo.remotes.origin.pull. This is the escape hatch for any GitPython kwarg not covered above (e.g. {'allow_unsafe_protocols': True}).

 dict()
Source code in pydantic_ai_skills/registries/git.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass
class GitCloneOptions:
    """Low-level GitPython configuration for clone and fetch operations.

    All fields map directly to arguments accepted by ``git.Repo.clone_from`` or
    ``git.Remote.fetch`` / ``git.Remote.pull``, so developers who know GitPython can
    use the full API without any wrapper layer.

    Args:
        depth: Create a shallow clone with history truncated to this many commits.
            Passed as ``--depth`` to git. ``None`` means a full clone.
            Useful for large repositories where only the latest snapshot is needed.
        branch: Name of the remote branch, tag, or ref to check out after cloning
            (``--branch`` flag). Defaults to the repository's default branch when
            ``None``.
        single_branch: When ``True``, clone only the branch specified by ``branch``
            (``--single-branch``). Has no effect when ``branch`` is ``None``.
        sparse_paths: List of path patterns to include in a sparse checkout
            (``--sparse`` + ``git sparse-checkout set``). An empty list disables
            sparse checkout and fetches the full tree.
        env: Mapping of environment variables forwarded to every git sub-process
            (e.g. ``GIT_SSH_COMMAND``, ``GIT_ASKPASS``). These override the
            process environment for git calls only.
        multi_options: Extra ``--option`` strings passed verbatim to
            ``git.Repo.clone_from(multi_options=...)``. Use for git options not
            exposed by other fields (e.g. ``['--filter=blob:none']`` for a
            partial/blobless clone).
        git_options: Mapping forwarded as keyword arguments directly to
            ``git.Repo.clone_from`` or ``repo.remotes.origin.pull``. This is the
            escape hatch for any GitPython kwarg not covered above
            (e.g. ``{'allow_unsafe_protocols': True}``).
    """

    depth: int | None = None
    branch: str | None = None
    single_branch: bool = False
    sparse_paths: list[str] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    multi_options: list[str] = field(default_factory=list)
    git_options: dict[str, Any] = field(default_factory=dict)

Bases: SkillRegistry

Skills registry backed by an S3 bucket, downloaded with boto3.

:meth:sync lists and downloads every object under bucket/prefix into a local cache directory, then returns the directory holding the skill packages. Each sync mirrors the remote prefix — the cached subtree is cleared first, so skills removed from the bucket no longer appear locally.

Works with Amazon S3 and any S3-compatible store (MinIO, Ceph, Cloudflare R2, etc.). All connection details — credentials, endpoint_url, region, TLS, and path-style addressing — are configured on the boto3 client you pass via boto3_client. When omitted, a default boto3.client("s3") is built, which uses boto3's standard credential resolution chain.

It does not parse SKILL.md: the directory it produces is handed to :class:~pydantic_ai_skills.SkillsCapability, and validating and rendering the packages inside it is pydantic-ai-harness's job.

Parameters:

Name Type Description Default
bucket str

Name of the S3 bucket containing the skills.

 required
prefix str

Key prefix inside the bucket where skill directories live. Defaults to the bucket root (""). For example, pass "skills" when skills live at s3://bucket/skills/<skill>/.

 ''
target_dir str | Path | None

Local directory where objects are downloaded. Defaults to a temporary directory scoped to the registry instance. A directory you pass persists across :meth:sync calls and is not cleaned up automatically — callers own the lifecycle.

 None
boto3_client Any | None

A pre-built boto3 S3 client. Use this to configure credentials, endpoint_url (for MinIO/Ceph/R2), region, TLS, and path-style addressing. When None, a default boto3.client("s3") is created (requires the s3 extra: pip install pydantic-ai-skills[s3]).

 None
auto_install bool

When True (default), :meth:sync contacts S3 so the local copy is up to date. Set to False to read only what already exists in target_dir, which is what offline or air-gapped environments want.

 True

Examples:

Amazon S3 with the ambient credential chain:

from pydantic_ai_skills import S3SkillsRegistry, SkillsCapability

capability = SkillsCapability(
    registries=[S3SkillsRegistry(bucket="my-skills", prefix="skills")]
)

MinIO (or any S3-compatible store) with a custom client:

import boto3
from botocore.config import Config
from pydantic_ai_skills.registries.s3 import S3SkillsRegistry

client = boto3.client(
    "s3",
    endpoint_url="http://localhost:9000",
    aws_access_key_id="minioadmin",
    aws_secret_access_key="minioadmin",
    config=Config(s3={"addressing_style": "path"}),
)
registry = S3SkillsRegistry(bucket="skills", boto3_client=client)
Source code in pydantic_ai_skills/registries/s3.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
class S3SkillsRegistry(SkillRegistry):
    """Skills registry backed by an S3 bucket, downloaded with boto3.

    :meth:`sync` lists and downloads every object under ``bucket/prefix`` into a local
    cache directory, then returns the directory holding the skill packages. Each sync
    mirrors the remote prefix — the cached subtree is cleared first, so skills removed
    from the bucket no longer appear locally.

    Works with Amazon S3 and any S3-compatible store (MinIO, Ceph, Cloudflare R2,
    etc.). All connection details — credentials, ``endpoint_url``, region, TLS,
    and path-style addressing — are configured on the boto3 client you pass via
    ``boto3_client``. When omitted, a default ``boto3.client("s3")`` is built,
    which uses boto3's standard credential resolution chain.

    It does not parse ``SKILL.md``: the directory it produces is handed to
    :class:`~pydantic_ai_skills.SkillsCapability`, and validating and rendering the
    packages inside it is `pydantic-ai-harness`'s job.

    Args:
        bucket: Name of the S3 bucket containing the skills.
        prefix: Key prefix inside the bucket where skill directories live.
            Defaults to the bucket root (``""``). For example, pass ``"skills"``
            when skills live at ``s3://bucket/skills/<skill>/``.
        target_dir: Local directory where objects are downloaded. Defaults to a
            temporary directory scoped to the registry instance. A directory you pass
            persists across :meth:`sync` calls and is **not** cleaned up automatically —
            callers own the lifecycle.
        boto3_client: A pre-built boto3 S3 client. Use this to configure
            credentials, ``endpoint_url`` (for MinIO/Ceph/R2), region, TLS, and
            path-style addressing. When ``None``, a default ``boto3.client("s3")``
            is created (requires the ``s3`` extra: ``pip install pydantic-ai-skills[s3]``).
        auto_install: When ``True`` (default), :meth:`sync` contacts S3 so the local copy
            is up to date. Set to ``False`` to read only what already exists in
            ``target_dir``, which is what offline or air-gapped environments want.

    Examples:
        Amazon S3 with the ambient credential chain:

        ```python
        from pydantic_ai_skills import S3SkillsRegistry, SkillsCapability

        capability = SkillsCapability(
            registries=[S3SkillsRegistry(bucket="my-skills", prefix="skills")]
        )
        ```

        MinIO (or any S3-compatible store) with a custom client:

        ```python
        import boto3
        from botocore.config import Config
        from pydantic_ai_skills.registries.s3 import S3SkillsRegistry

        client = boto3.client(
            "s3",
            endpoint_url="http://localhost:9000",
            aws_access_key_id="minioadmin",
            aws_secret_access_key="minioadmin",
            config=Config(s3={"addressing_style": "path"}),
        )
        registry = S3SkillsRegistry(bucket="skills", boto3_client=client)
        ```
    """

    def __init__(
        self,
        bucket: str,
        *,
        prefix: str = '',
        target_dir: str | Path | None = None,
        boto3_client: Any | None = None,
        auto_install: bool = True,
    ) -> None:
        if boto3_client is None:
            try:
                import boto3
            except ImportError as exc:
                raise ImportError(
                    'boto3 is required to build a default S3 client for S3SkillsRegistry. '
                    'Install it with: pip install pydantic-ai-skills[s3], or pass a pre-built '
                    'boto3_client.'
                ) from exc
            self._client = boto3.client('s3')
        else:
            self._client = boto3_client

        self._bucket = bucket
        self._prefix = prefix.strip('/')
        self._auto_install = auto_install
        self._tmp_dir: Any | None = None
        # Cache of the most recent object listing (Key -> LastModified), populated by _sync.
        self._object_modified: dict[str, datetime | None] = {}

        if target_dir is None:
            import tempfile

            self._tmp_dir = tempfile.TemporaryDirectory()
            self._target_dir = Path(self._tmp_dir.name)
        else:
            self._target_dir = Path(target_dir).expanduser().resolve()

    def __repr__(self) -> str:
        return (
            f'{type(self).__name__}('
            f'bucket={self._bucket!r}, '
            f'prefix={self._prefix!r}, '
            f'target_dir={str(self._target_dir)!r})'
        )

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _skills_root(self) -> Path:
        """Return the path inside the cache where skill directories live."""
        if self._prefix:
            return self._target_dir / self._prefix
        return self._target_dir

    def _list_objects(self) -> list[dict[str, Any]]:
        """Return all object summaries under ``bucket/prefix`` via pagination."""
        list_prefix = f'{self._prefix}/' if self._prefix else ''
        try:
            paginator = self._client.get_paginator('list_objects_v2')
            objects: list[dict[str, Any]] = []
            for page in paginator.paginate(Bucket=self._bucket, Prefix=list_prefix):
                objects.extend(page.get('Contents', []))
            return objects
        except Exception as exc:  # surface any boto3/botocore error with context
            raise RuntimeError(
                f"Failed to list objects in bucket '{self._bucket}' (prefix '{self._prefix}'): {exc}"
            ) from exc

    def _sync(self) -> None:
        """Mirror all objects under ``bucket/prefix`` into ``target_dir``.

        Clears the cached prefix subtree first so skills removed from the bucket
        do not linger locally, then downloads the current objects. The listing is
        fetched once and cached for metadata enrichment.
        """
        objects = self._list_objects()
        self._object_modified = {obj['Key']: obj.get('LastModified') for obj in objects}

        # Mirror the remote: drop the previously synced subtree before re-downloading.
        skills_root = self._skills_root()
        if skills_root.exists():
            shutil.rmtree(skills_root)

        self._target_dir.mkdir(parents=True, exist_ok=True)
        target_resolved = self._target_dir.resolve()

        for obj in objects:
            key = obj['Key']
            if key.endswith('/'):
                # Directory marker — nothing to download.
                continue

            dest = self._target_dir / key
            # Path-traversal guard: the resolved destination must stay inside target_dir.
            if not dest.resolve().is_relative_to(target_resolved):
                raise ValueError(f"Object key '{key}' escapes target directory '{target_resolved}'.")

            dest.parent.mkdir(parents=True, exist_ok=True)
            try:
                self._client.download_file(self._bucket, key, str(dest))
            except Exception as exc:  # surface any boto3/botocore error with context
                raise RuntimeError(f"Failed to download '{key}' from bucket '{self._bucket}': {exc}") from exc

    def _latest_modified(self, skill_name: str) -> str | None:
        """Return the newest ``LastModified`` across one skill's objects, ISO-formatted.

        Reads the object listing cached by the most recent :meth:`sync`, so it performs no
        additional S3 calls.
        """
        key_prefix = f'{self._prefix}/{skill_name}/'.lstrip('/')
        latest: datetime | None = None
        for key, modified in self._object_modified.items():
            if key.startswith(key_prefix) and modified is not None and (latest is None or modified > latest):
                latest = modified
        return latest.isoformat() if latest is not None else None

    # ------------------------------------------------------------------
    # SkillRegistry interface
    # ------------------------------------------------------------------

    def sync(self) -> Path:
        """Download the bucket prefix and return its skill-library directory.

        The returned path is ``target_dir`` joined with ``prefix``, whose immediate
        children are the skill packages. With ``auto_install=False`` nothing is
        downloaded and whatever is already on disk is returned.

        Returns:
            Path to the local skill-library directory.

        Raises:
            RuntimeError: On S3 listing or download errors.
            ValueError: When the prefix holds no synced skill library — usually a
                ``prefix`` that does not match the bucket's layout, or
                ``auto_install=False`` with nothing downloaded yet.
        """
        if self._auto_install:
            self._sync()

        skills_root = self._skills_root()
        if not skills_root.is_dir():
            detail = (
                'nothing has been downloaded yet and auto_install is disabled'
                if not self._auto_install
                else f'prefix={self._prefix!r} matched no objects'
            )
            raise ValueError(f"No skill library at {skills_root} for bucket '{self._bucket}': {detail}.")
        return skills_root

    def revision(self, skill_name: str) -> str | None:
        """Return the newest object modification time for one skill, ISO-formatted.

        Useful for recording which version of a remote skill an agent ran with, since
        :meth:`sync` otherwise tracks a moving prefix. Returns None before the first sync
        or when the skill has no objects in the cached listing.
        """
        return self._latest_modified(skill_name)

sync 

sync() -> Path

Download the bucket prefix and return its skill-library directory.

The returned path is target_dir joined with prefix, whose immediate children are the skill packages. With auto_install=False nothing is downloaded and whatever is already on disk is returned.

Returns:

Type Description
Path

Path to the local skill-library directory.

Raises:

Type Description
RuntimeError

On S3 listing or download errors.
ValueError

When the prefix holds no synced skill library — usually a prefix that does not match the bucket's layout, or auto_install=False with nothing downloaded yet.
Source code in pydantic_ai_skills/registries/s3.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def sync(self) -> Path:
    """Download the bucket prefix and return its skill-library directory.

    The returned path is ``target_dir`` joined with ``prefix``, whose immediate
    children are the skill packages. With ``auto_install=False`` nothing is
    downloaded and whatever is already on disk is returned.

    Returns:
        Path to the local skill-library directory.

    Raises:
        RuntimeError: On S3 listing or download errors.
        ValueError: When the prefix holds no synced skill library — usually a
            ``prefix`` that does not match the bucket's layout, or
            ``auto_install=False`` with nothing downloaded yet.
    """
    if self._auto_install:
        self._sync()

    skills_root = self._skills_root()
    if not skills_root.is_dir():
        detail = (
            'nothing has been downloaded yet and auto_install is disabled'
            if not self._auto_install
            else f'prefix={self._prefix!r} matched no objects'
        )
        raise ValueError(f"No skill library at {skills_root} for bucket '{self._bucket}': {detail}.")
    return skills_root

revision 

revision(skill_name: str) -> str | None

Return the newest object modification time for one skill, ISO-formatted.

Useful for recording which version of a remote skill an agent ran with, since :meth:sync otherwise tracks a moving prefix. Returns None before the first sync or when the skill has no objects in the cached listing.

Source code in pydantic_ai_skills/registries/s3.py 
234
235
236
237
238
239
240
241
def revision(self, skill_name: str) -> str | None:
    """Return the newest object modification time for one skill, ISO-formatted.

    Useful for recording which version of a remote skill an agent ran with, since
    :meth:`sync` otherwise tracks a moving prefix. Returns None before the first sync
    or when the skill has no objects in the cached listing.
    """
    return self._latest_modified(skill_name)

Bases: SkillRegistry

A registry backed by a skill library already present on the filesystem.

Passing a local directory straight to SkillsCapability(directories=...) is simpler and does the same thing. Use this when a local library needs to be composed — merged with a remote one, prefixed, or filtered — since composition operates on registries.

Attributes:

Name Type Description
path str | Path

The skill-library directory. Its immediate children are skill packages.
Example 
from pydantic_ai_skills import GitSkillsRegistry
from pydantic_ai_skills.registries import LocalSkillsRegistry

# Local skills take precedence over the ones published upstream.
combined = LocalSkillsRegistry('./skills') | GitSkillsRegistry(
    'https://github.com/anthropics/skills', path='skills'
)
Source code in pydantic_ai_skills/registries/local.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class LocalSkillsRegistry(SkillRegistry):
    """A registry backed by a skill library already present on the filesystem.

    Passing a local directory straight to `SkillsCapability(directories=...)` is simpler
    and does the same thing. Use this when a local library needs to be *composed* — merged
    with a remote one, prefixed, or filtered — since composition operates on registries.

    Attributes:
        path: The skill-library directory. Its immediate children are skill packages.

    Example:
        ```python
        from pydantic_ai_skills import GitSkillsRegistry
        from pydantic_ai_skills.registries import LocalSkillsRegistry

        # Local skills take precedence over the ones published upstream.
        combined = LocalSkillsRegistry('./skills') | GitSkillsRegistry(
            'https://github.com/anthropics/skills', path='skills'
        )
        ```
    """

    path: str | Path

    def sync(self) -> Path:
        """Return the library directory, checking that it exists.

        Raises:
            ValueError: When the path does not exist or is not a directory.
        """
        library = Path(self.path).expanduser()
        if not library.exists():
            raise ValueError(f'Skill library directory does not exist: {library}')
        if not library.is_dir():
            raise ValueError(f'Skill library path is not a directory: {library}')
        return library

sync 

sync() -> Path

Return the library directory, checking that it exists.

Raises:

Type Description
ValueError

When the path does not exist or is not a directory.
Source code in pydantic_ai_skills/registries/local.py 
42
43
44
45
46
47
48
49
50
51
52
53
def sync(self) -> Path:
    """Return the library directory, checking that it exists.

    Raises:
        ValueError: When the path does not exist or is not a directory.
    """
    library = Path(self.path).expanduser()
    if not library.exists():
        raise ValueError(f'Skill library directory does not exist: {library}')
    if not library.is_dir():
        raise ValueError(f'Skill library path is not a directory: {library}')
    return library

Composition Wrappers

Each wrapper syncs the registry it wraps, then stages a new library holding the packages it wants under the names it wants. The wrapped registry is never modified.

Bases: SkillRegistry

A registry that wraps another registry and delegates to it.

:meth:sync is forwarded to wrapped. Subclasses that present a different library than the one they wrap override it to stage their own.

Attributes:

Name Type Description
wrapped SkillRegistry

The registry being decorated.
target_dir str | Path | None

Where a subclass stages its composed library. When None, a process-lifetime temporary directory is used.
Source code in pydantic_ai_skills/registries/wrapper.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@dataclass
class WrapperRegistry(SkillRegistry):
    """A registry that wraps another registry and delegates to it.

    :meth:`sync` is forwarded to ``wrapped``. Subclasses that present a different library
    than the one they wrap override it to stage their own.

    Attributes:
        wrapped: The registry being decorated.
        target_dir: Where a subclass stages its composed library. When None, a
            process-lifetime temporary directory is used.
    """

    wrapped: SkillRegistry
    target_dir: str | Path | None = field(default=None, kw_only=True)

    def sync(self) -> Path:
        """Delegate sync to the wrapped registry."""
        return self.wrapped.sync()

sync 

sync() -> Path

Delegate sync to the wrapped registry.

Source code in pydantic_ai_skills/registries/wrapper.py 
34
35
36
def sync(self) -> Path:
    """Delegate sync to the wrapped registry."""
    return self.wrapped.sync()

Bases: WrapperRegistry

A registry that exposes only the skills matching a predicate.

Syncs the wrapped registry, then stages a library containing just the packages for which predicate(info) is True. The wrapped registry's own copy is never modified.

Example 
pdf_only = registry.filtered(lambda info: 'pdf' in info.name)
Source code in pydantic_ai_skills/registries/filtered.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@dataclass
class FilteredRegistry(WrapperRegistry):
    """A registry that exposes only the skills matching a predicate.

    Syncs the wrapped registry, then stages a library containing just the packages for
    which ``predicate(info)`` is ``True``. The wrapped registry's own copy is never
    modified.

    Example:
        ```python
        pdf_only = registry.filtered(lambda info: 'pdf' in info.name)
        ```
    """

    predicate: Callable[[SkillInfo], bool]

    def sync(self) -> Path:
        """Stage a library holding only the skills that pass the predicate."""
        source = self.wrapped.sync()
        staged = staging_directory(self.target_dir)

        for child in sorted(source.iterdir()):
            if not child.is_dir():
                continue
            info = read_skill_info(child)
            if info is None or not self.predicate(info):
                continue
            copy_skill_directory(child, staged, info.name)

        return staged

sync 

sync() -> Path

Stage a library holding only the skills that pass the predicate.

Source code in pydantic_ai_skills/registries/filtered.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def sync(self) -> Path:
    """Stage a library holding only the skills that pass the predicate."""
    source = self.wrapped.sync()
    staged = staging_directory(self.target_dir)

    for child in sorted(source.iterdir()):
        if not child.is_dir():
            continue
        info = read_skill_info(child)
        if info is None or not self.predicate(info):
            continue
        copy_skill_directory(child, staged, info.name)

    return staged

Bases: WrapperRegistry

A registry that prepends a prefix to every skill name.

Because harness derives a skill's name from its directory — and rejects a SKILL.md whose frontmatter name disagrees with it — renaming means staging the package under the new directory name and rewriting that key. Both happen here.

Example 
anthropic = registry.prefixed('anthropic-')
# the "pdf" skill is exposed to the model as "anthropic-pdf"
Source code in pydantic_ai_skills/registries/prefixed.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass
class PrefixedRegistry(WrapperRegistry):
    """A registry that prepends a prefix to every skill name.

    Because harness derives a skill's name from its directory — and rejects a `SKILL.md`
    whose frontmatter `name` disagrees with it — renaming means staging the package under
    the new directory name *and* rewriting that key. Both happen here.

    Example:
        ```python
        anthropic = registry.prefixed('anthropic-')
        # the "pdf" skill is exposed to the model as "anthropic-pdf"
        ```
    """

    prefix: str

    def sync(self) -> Path:
        """Stage a library whose skills are all renamed with the prefix.

        Raises:
            ValueError: When the prefix yields a name harness would reject.
        """
        source = self.wrapped.sync()
        staged = staging_directory(self.target_dir)

        for child in sorted(source.iterdir()):
            if not child.is_dir():
                continue
            info = read_skill_info(child)
            if info is None:
                continue

            new_name = validate_skill_name(
                f'{self.prefix}{info.name}',
                context=f'Prefixing {info.name!r} with {self.prefix!r}',
            )
            staged_skill = copy_skill_directory(child, staged, new_name)
            rewrite_skill_name(staged_skill / 'SKILL.md', new_name)

        return staged

sync 

sync() -> Path

Stage a library whose skills are all renamed with the prefix.

Raises:

Type Description
ValueError

When the prefix yields a name harness would reject.
Source code in pydantic_ai_skills/registries/prefixed.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def sync(self) -> Path:
    """Stage a library whose skills are all renamed with the prefix.

    Raises:
        ValueError: When the prefix yields a name harness would reject.
    """
    source = self.wrapped.sync()
    staged = staging_directory(self.target_dir)

    for child in sorted(source.iterdir()):
        if not child.is_dir():
            continue
        info = read_skill_info(child)
        if info is None:
            continue

        new_name = validate_skill_name(
            f'{self.prefix}{info.name}',
            context=f'Prefixing {info.name!r} with {self.prefix!r}',
        )
        staged_skill = copy_skill_directory(child, staged, new_name)
        rewrite_skill_name(staged_skill / 'SKILL.md', new_name)

    return staged

Bases: WrapperRegistry

A registry that exposes skills under names from a mapping.

Skills the map does not mention keep their original name. As with :class:~pydantic_ai_skills.registries.prefixed.PrefixedRegistry, renaming stages the package under its new directory name and rewrites the frontmatter name so harness finds the two in agreement.

Attributes:

Name Type Description
name_map dict[str, str]

Mapping of {new_name: original_name}.
Example 
registry.renamed({'anthropic-pdf': 'pdf'})
Source code in pydantic_ai_skills/registries/renamed.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class RenamedRegistry(WrapperRegistry):
    """A registry that exposes skills under names from a mapping.

    Skills the map does not mention keep their original name. As with
    :class:`~pydantic_ai_skills.registries.prefixed.PrefixedRegistry`, renaming stages the
    package under its new directory name and rewrites the frontmatter `name` so harness
    finds the two in agreement.

    Attributes:
        name_map: Mapping of ``{new_name: original_name}``.

    Example:
        ```python
        registry.renamed({'anthropic-pdf': 'pdf'})
        ```
    """

    name_map: dict[str, str] = field(default_factory=dict)

    def sync(self) -> Path:
        """Stage a library with the mapped skills renamed.

        Raises:
            ValueError: When a new name is one harness would reject, when the map names an
                original skill this registry does not hold, or when two skills would end
                up sharing a name.
        """
        source = self.wrapped.sync()
        staged = staging_directory(self.target_dir)

        renames = {original: new for new, original in self.name_map.items()}
        available = {
            info.name: info for child in sorted(source.iterdir()) if child.is_dir() if (info := read_skill_info(child))
        }

        unknown = sorted(set(renames) - set(available))
        if unknown:
            noun = 'skill' if len(unknown) == 1 else 'skills'
            available_text = ', '.join(sorted(available)) or '(none)'
            raise ValueError(f'Unknown {noun} in name_map: {", ".join(unknown)}. Available skills: {available_text}.')

        staged_names: dict[str, str] = {}
        for original, info in available.items():
            new_name = renames.get(original, original)
            if new_name != original:
                new_name = validate_skill_name(new_name, context=f'Renaming {original!r}')
            if previous := staged_names.get(new_name):
                raise ValueError(f'Renaming would give {previous!r} and {original!r} the same name {new_name!r}.')
            staged_names[new_name] = original

            staged_skill = copy_skill_directory(info.directory, staged, new_name)
            if new_name != original:
                rewrite_skill_name(staged_skill / 'SKILL.md', new_name)

        return staged

sync 

sync() -> Path

Stage a library with the mapped skills renamed.

Raises:

Type Description
ValueError

When a new name is one harness would reject, when the map names an original skill this registry does not hold, or when two skills would end up sharing a name.
Source code in pydantic_ai_skills/registries/renamed.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def sync(self) -> Path:
    """Stage a library with the mapped skills renamed.

    Raises:
        ValueError: When a new name is one harness would reject, when the map names an
            original skill this registry does not hold, or when two skills would end
            up sharing a name.
    """
    source = self.wrapped.sync()
    staged = staging_directory(self.target_dir)

    renames = {original: new for new, original in self.name_map.items()}
    available = {
        info.name: info for child in sorted(source.iterdir()) if child.is_dir() if (info := read_skill_info(child))
    }

    unknown = sorted(set(renames) - set(available))
    if unknown:
        noun = 'skill' if len(unknown) == 1 else 'skills'
        available_text = ', '.join(sorted(available)) or '(none)'
        raise ValueError(f'Unknown {noun} in name_map: {", ".join(unknown)}. Available skills: {available_text}.')

    staged_names: dict[str, str] = {}
    for original, info in available.items():
        new_name = renames.get(original, original)
        if new_name != original:
            new_name = validate_skill_name(new_name, context=f'Renaming {original!r}')
        if previous := staged_names.get(new_name):
            raise ValueError(f'Renaming would give {previous!r} and {original!r} the same name {new_name!r}.')
        staged_names[new_name] = original

        staged_skill = copy_skill_directory(info.directory, staged, new_name)
        if new_name != original:
            rewrite_skill_name(staged_skill / 'SKILL.md', new_name)

    return staged

Bases: SkillRegistry

A registry that merges several registries into one library.

Every child is synced and its packages staged into a single directory. Earlier registries win on a duplicate skill name, and the shadowed one is reported with a UserWarning — merging silently would hand harness a library whose contents depend on directory iteration order.

Passing the merged library to SkillsCapability is equivalent to passing each child's own library, except that this resolves the collisions itself rather than letting harness reject the duplicate.

Attributes:

Name Type Description
registries Sequence[SkillRegistry]

The registries to merge, in precedence order.
target_dir str | Path | None

Where to stage the merged library. When None, a process-lifetime temporary directory is used.
Example 
from pydantic_ai_skills.registries import CombinedRegistry

combined = CombinedRegistry(registries=[internal_registry, public_registry])
Source code in pydantic_ai_skills/registries/combined.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class CombinedRegistry(SkillRegistry):
    """A registry that merges several registries into one library.

    Every child is synced and its packages staged into a single directory. Earlier
    registries win on a duplicate skill name, and the shadowed one is reported with a
    `UserWarning` — merging silently would hand harness a library whose contents depend on
    directory iteration order.

    Passing the merged library to `SkillsCapability` is equivalent to passing each child's
    own library, except that this resolves the collisions itself rather than letting
    harness reject the duplicate.

    Attributes:
        registries: The registries to merge, in precedence order.
        target_dir: Where to stage the merged library. When None, a process-lifetime
            temporary directory is used.

    Example:
        ```python
        from pydantic_ai_skills.registries import CombinedRegistry

        combined = CombinedRegistry(registries=[internal_registry, public_registry])
        ```
    """

    registries: Sequence[SkillRegistry]
    target_dir: str | Path | None = field(default=None, kw_only=True)

    def sync(self) -> Path:
        """Sync every child registry and stage their skills into one library."""
        staged = staging_directory(self.target_dir)
        claimed: dict[str, SkillRegistry] = {}

        for registry in self.registries:
            source = registry.sync()
            for child in sorted(source.iterdir()):
                if not child.is_dir():
                    continue
                info = read_skill_info(child)
                if info is None:
                    continue
                if owner := claimed.get(info.name):
                    warnings.warn(
                        f"Skill '{info.name}' is provided by more than one registry; keeping the one from "
                        f'{owner!r} and skipping {registry!r}. Use `.prefixed()` or `.renamed()` to expose both.',
                        UserWarning,
                        stacklevel=2,
                    )
                    continue
                claimed[info.name] = registry
                copy_skill_directory(child, staged, info.name)

        return staged

sync 

sync() -> Path

Sync every child registry and stage their skills into one library.

Source code in pydantic_ai_skills/registries/combined.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def sync(self) -> Path:
    """Sync every child registry and stage their skills into one library."""
    staged = staging_directory(self.target_dir)
    claimed: dict[str, SkillRegistry] = {}

    for registry in self.registries:
        source = registry.sync()
        for child in sorted(source.iterdir()):
            if not child.is_dir():
                continue
            info = read_skill_info(child)
            if info is None:
                continue
            if owner := claimed.get(info.name):
                warnings.warn(
                    f"Skill '{info.name}' is provided by more than one registry; keeping the one from "
                    f'{owner!r} and skipping {registry!r}. Use `.prefixed()` or `.renamed()` to expose both.',
                    UserWarning,
                    stacklevel=2,
                )
                continue
            claimed[info.name] = registry
            copy_skill_directory(child, staged, info.name)

    return staged