Skip to content

Registries API Reference

Bases: ABC

Abstract base for skill registries.

A skill registry is a source of skills that can be searched, retrieved, installed and updated. Concrete implementations may back registries with a Git repository, a REST API, a local directory, etc.

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 
 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
class SkillRegistry(ABC):
    """Abstract base for skill registries.

    A skill registry is a source of skills that can be searched, retrieved,
    installed and updated. Concrete implementations may back registries with
    a Git repository, a REST API, a local directory, etc.

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

    @abstractmethod
    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search for skills by keyword.

        Args:
            query: Keyword matched case-insensitively against ``name`` and
                ``description``.
            limit: Maximum number of results.

        Returns:
            List of matching :class:`~pydantic_ai_skills.Skill` objects.
        """

    @abstractmethod
    async def get(self, skill_name: str) -> Skill:
        """Return a single skill by name.

        Args:
            skill_name: Exact skill name from ``SKILL.md`` frontmatter.

        Returns:
            The matching :class:`~pydantic_ai_skills.Skill`.

        Raises:
            KeyError: When no skill with ``skill_name`` exists.
        """

    @abstractmethod
    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Copy a skill into ``target_dir``.

        Args:
            skill_name: Name of the skill to install.
            target_dir: Destination directory; a ``skill_name`` subdirectory
                is created inside it.

        Returns:
            Path to the installed skill directory.
        """

    @abstractmethod
    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Update an already-installed skill to the latest version.

        Args:
            skill_name: Name of the skill to update.
            target_dir: Directory where the skill was previously installed.

        Returns:
            Path to the updated skill directory.
        """

    @abstractmethod
    def get_skills(self) -> list[Skill]:
        """Return all skills available in this registry.

        Concrete implementations must return pre-loaded skill objects.
        This is called synchronously by :class:`~pydantic_ai_skills.SkillsToolset`
        during initialization.

        Returns:
            List of :class:`~pydantic_ai_skills.Skill` objects.
        """

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

        Args:
            predicate: A callable that accepts a :class:`~pydantic_ai_skills.Skill`
                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.

        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)

search abstractmethod async 

search(query: str, limit: int = 10) -> list[Skill]

Search for skills by keyword.

Parameters:

Name Type Description Default
query str

Keyword matched case-insensitively against name and description.

 required
limit int

Maximum number of results.

 10

Returns:

Type Description
list[Skill]

List of matching :class:~pydantic_ai_skills.Skill objects.
Source code in pydantic_ai_skills/registries/_base.py 
44
45
46
47
48
49
50
51
52
53
54
55
@abstractmethod
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search for skills by keyword.

    Args:
        query: Keyword matched case-insensitively against ``name`` and
            ``description``.
        limit: Maximum number of results.

    Returns:
        List of matching :class:`~pydantic_ai_skills.Skill` objects.
    """

get abstractmethod async 

get(skill_name: str) -> Skill

Return a single skill by name.

Parameters:

Name Type Description Default
skill_name str

Exact skill name from SKILL.md frontmatter.

 required

Returns:

Type Description
Skill

The matching :class:~pydantic_ai_skills.Skill.

Raises:

Type Description
KeyError

When no skill with skill_name exists.
Source code in pydantic_ai_skills/registries/_base.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
@abstractmethod
async def get(self, skill_name: str) -> Skill:
    """Return a single skill by name.

    Args:
        skill_name: Exact skill name from ``SKILL.md`` frontmatter.

    Returns:
        The matching :class:`~pydantic_ai_skills.Skill`.

    Raises:
        KeyError: When no skill with ``skill_name`` exists.
    """

install abstractmethod async 

install(skill_name: str, target_dir: str | Path) -> Path

Copy a skill into target_dir.

Parameters:

Name Type Description Default
skill_name str

Name of the skill to install.

 required
target_dir str | Path

Destination directory; a skill_name subdirectory is created inside it.

 required

Returns:

Type Description
Path

Path to the installed skill directory.
Source code in pydantic_ai_skills/registries/_base.py 
71
72
73
74
75
76
77
78
79
80
81
82
@abstractmethod
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Copy a skill into ``target_dir``.

    Args:
        skill_name: Name of the skill to install.
        target_dir: Destination directory; a ``skill_name`` subdirectory
            is created inside it.

    Returns:
        Path to the installed skill directory.
    """

update abstractmethod async 

update(skill_name: str, target_dir: str | Path) -> Path

Update an already-installed skill to the latest version.

Parameters:

Name Type Description Default
skill_name str

Name of the skill to update.

 required
target_dir str | Path

Directory where the skill was previously installed.

 required

Returns:

Type Description
Path

Path to the updated skill directory.
Source code in pydantic_ai_skills/registries/_base.py 
84
85
86
87
88
89
90
91
92
93
94
@abstractmethod
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Update an already-installed skill to the latest version.

    Args:
        skill_name: Name of the skill to update.
        target_dir: Directory where the skill was previously installed.

    Returns:
        Path to the updated skill directory.
    """

get_skills abstractmethod 

get_skills() -> list[Skill]

Return all skills available in this registry.

Concrete implementations must return pre-loaded skill objects. This is called synchronously by :class:~pydantic_ai_skills.SkillsToolset during initialization.

Returns:

Type Description
list[Skill]

List of :class:~pydantic_ai_skills.Skill objects.
Source code in pydantic_ai_skills/registries/_base.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
@abstractmethod
def get_skills(self) -> list[Skill]:
    """Return all skills available in this registry.

    Concrete implementations must return pre-loaded skill objects.
    This is called synchronously by :class:`~pydantic_ai_skills.SkillsToolset`
    during initialization.

    Returns:
        List of :class:`~pydantic_ai_skills.Skill` objects.
    """

filtered 

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

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

Parameters:

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

A callable that accepts a :class:~pydantic_ai_skills.Skill 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 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def filtered(self, predicate: Callable[[Skill], bool]) -> FilteredRegistry:
    """Return a view of this registry limited to skills matching ``predicate``.

    Args:
        predicate: A callable that accepts a :class:`~pydantic_ai_skills.Skill`
            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.

 required

Returns:

Name Type Description
A PrefixedRegistry
PrefixedRegistry

view backed by the same underlying source.
Source code in pydantic_ai_skills/registries/_base.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
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.

    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 
137
138
139
140
141
142
143
144
145
146
147
148
149
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)

Bases: SkillRegistry

Skills registry backed by a Git repository cloned with GitPython.

Clones the target repository on the first call to install or search/get, then performs a git pull on subsequent calls (or a full re-clone if the local copy is corrupted/missing).

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.).

search() and get() return :class:~pydantic_ai_skills.Skill objects parsed from SKILL.md frontmatter + body. Registry-specific metadata (source_url, version, repo) is stored in skill.metadata.

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. The cloned tree persists across install / update calls but 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
validate bool

Whether to run validate_skill_metadata() on every discovered SKILL.md after installation. Mirrors the homonymous flag on :class:~pydantic_ai_skills.SkillsDirectory. Defaults to True.

 True
auto_install bool

When True (default), search and get trigger a clone/pull automatically so the local copy is always up to date. Set to False to require explicit install / update calls, which is preferable in offline or air-gapped environments.

 True

Examples:

Basic usage — clone and register all skills:

from pydantic_ai_skills import SkillsToolset
from pydantic_ai_skills.registries.git import GitSkillsRegistry

toolset = SkillsToolset(
    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 skill: "pdf" in skill.name.lower())

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:

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 
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
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
class GitSkillsRegistry(SkillRegistry):
    """Skills registry backed by a Git repository cloned with GitPython.

    Clones the target repository on the first call to ``install`` or
    ``search``/``get``, then performs a ``git pull`` on subsequent calls
    (or a full re-clone if the local copy is corrupted/missing).

    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.).

    ``search()`` and ``get()`` return :class:`~pydantic_ai_skills.Skill` objects
    parsed from ``SKILL.md`` frontmatter + body. Registry-specific metadata
    (``source_url``, ``version``, ``repo``) is stored in ``skill.metadata``.

    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. The cloned
            tree persists across ``install`` / ``update`` calls but 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``.
        validate: Whether to run ``validate_skill_metadata()`` on every
            discovered ``SKILL.md`` after installation. Mirrors the homonymous
            flag on :class:`~pydantic_ai_skills.SkillsDirectory`. Defaults to ``True``.
        auto_install: When ``True`` (default), ``search`` and ``get`` trigger a
            clone/pull automatically so the local copy is always up to date.
            Set to ``False`` to require explicit ``install`` / ``update`` calls,
            which is preferable in offline or air-gapped environments.

    Examples:
        Basic usage — clone and register all skills:

        ```python
        from pydantic_ai_skills import SkillsToolset
        from pydantic_ai_skills.registries.git import GitSkillsRegistry

        toolset = SkillsToolset(
            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 skill: "pdf" in skill.name.lower())
        ```

        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:

        ```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,
        validate: bool = True,
        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._validate = validate
        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 metadata
        self._clean_repo_url = _sanitize_url(repo_url)

        # Eagerly clone/pull and cache discovered skills
        self._cached_skills: list[Skill] = []
        if self._auto_install:
            self._ensure_cloned()
            self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]

    # ------------------------------------------------------------------
    # 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 _get_commit_sha(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

    def _load_skills(self) -> list[Skill]:
        """Discover all skills from the cloned repository path."""
        skills_root = self._skills_root()
        if not skills_root.exists():
            return []
        return discover_skills(path=skills_root, validate=self._validate, max_depth=2)

    def _enrich_metadata(self, skill: Skill, *, version: str | None = None) -> Skill:
        """Inject registry-specific keys into ``skill.metadata``."""
        from dataclasses import replace

        extra: dict[str, Any] = {
            'source_url': _build_source_url(
                self._clean_repo_url,
                self._path,
                skill.name,
                self._clone_options.branch,
            ),
            'registry': type(self).__name__,
            'repo': self._clean_repo_url,
            'version': version or self._get_commit_sha(),
        }
        existing = dict(skill.metadata) if skill.metadata else {}
        existing.update(extra)
        return replace(skill, metadata=existing)

    def _refresh(self) -> None:
        """Pull latest changes and rebuild the skills cache."""
        self._ensure_cloned()
        self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]

    def _ensure_skills_loaded(self) -> None:
        """Populate the skills cache if empty, respecting ``auto_install``."""
        if self._cached_skills:
            return
        if self._auto_install:
            self._ensure_cloned()
        self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]

    # ------------------------------------------------------------------
    # Synchronous skill access for SkillsToolset integration
    # ------------------------------------------------------------------

    def get_skills(self) -> list[Skill]:
        """Return all skills discovered from the cloned repository.

        If ``auto_install=True`` (default), the repository was cloned during
        ``__init__`` and skills are returned from cache. Otherwise, loads
        from whatever exists on disk without triggering a clone/pull.

        Returns:
            List of enriched :class:`~pydantic_ai_skills.Skill` objects.
        """
        self._ensure_skills_loaded()
        return list(self._cached_skills)

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

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search available skills by keyword.

        Matches ``query`` (case-insensitively) against each skill's ``name`` and
        ``description``. Uses the cached skill list populated during ``__init__``.

        Args:
            query: Keyword to search for.
            limit: Maximum number of results.

        Returns:
            List of :class:`~pydantic_ai_skills.Skill` objects. Each skill's
            ``metadata`` dict contains ``"source_url"`` for traceability.
        """
        q = query.lower()
        results: list[Skill] = []
        for skill in self.get_skills():
            if q in skill.name.lower() or q in (skill.description or '').lower():
                results.append(skill)
                if len(results) >= limit:
                    break
        return results

    async def get(self, skill_name: str) -> Skill:
        """Return the full skill by name.

        Args:
            skill_name: Exact skill name (with optional prefix).

        Returns:
            A fully-parsed :class:`~pydantic_ai_skills.Skill` with ``metadata``
            containing ``"source_url"``.

        Raises:
            KeyError: When no skill with ``skill_name`` exists.
        """
        for skill in self.get_skills():
            if skill.name == skill_name:
                return skill
        raise KeyError(f"Skill '{skill_name}' not found in registry {self._clean_repo_url!r}.")

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Copy a skill from the cloned repository into ``target_dir``.

        Clones the repository first if the local cache doesn't exist. Validation
        is handled by ``discover_skills()`` during cache population, so skills
        in the cache are already validated.

        Args:
            skill_name: Name of the skill to install.
            target_dir: Destination directory; a ``skill_name`` subdirectory
                is created inside it.

        Returns:
            Path to the installed skill directory (``target_dir/skill_name``).

        Raises:
            KeyError: When ``skill_name`` is not found in the registry.
            ValueError: When the destination or source path escapes its expected
                directory (path traversal protection).
            RuntimeError: On git or filesystem errors.
        """
        # Ensure we have skills loaded (already parsed and validated by discover_skills)
        self._ensure_skills_loaded()

        # Find skill in cache — its uri points to the source directory
        src_skill_dir: Path | None = None
        for skill in self._cached_skills:
            if skill.name == skill_name and skill.uri:
                src_skill_dir = Path(skill.uri)
                break

        if src_skill_dir is None:
            raise KeyError(f"Skill '{skill_name}' not found in repository {self._clean_repo_url!r}.")

        return copy_skill_directory(src_skill_dir, target_dir, skill_name)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Pull the latest changes and re-copy the skill to ``target_dir``.

        Performs a ``git pull`` on the cached clone before re-installing.
        Falls back to a fresh ``install`` if the skill is not yet installed.

        Args:
            skill_name: Name of the skill to update.
            target_dir: Directory where the skill was previously installed.

        Returns:
            Path to the updated skill directory.

        Raises:
            KeyError: When ``skill_name`` is not found after the pull.
            RuntimeError: On git or network errors.
        """
        dest = Path(target_dir).expanduser().resolve() / skill_name
        if not dest.exists():
            return await self.install(skill_name, target_dir)

        # Pull latest and refresh cache before reinstalling
        self._ensure_cloned()
        self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]
        return await self.install(skill_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Return all skills discovered from the cloned repository.

If auto_install=True (default), the repository was cloned during __init__ and skills are returned from cache. Otherwise, loads from whatever exists on disk without triggering a clone/pull.

Returns:

Type Description
list[Skill]

List of enriched :class:~pydantic_ai_skills.Skill objects.
Source code in pydantic_ai_skills/registries/git.py 
489
490
491
492
493
494
495
496
497
498
499
500
def get_skills(self) -> list[Skill]:
    """Return all skills discovered from the cloned repository.

    If ``auto_install=True`` (default), the repository was cloned during
    ``__init__`` and skills are returned from cache. Otherwise, loads
    from whatever exists on disk without triggering a clone/pull.

    Returns:
        List of enriched :class:`~pydantic_ai_skills.Skill` objects.
    """
    self._ensure_skills_loaded()
    return list(self._cached_skills)

search async 

search(query: str, limit: int = 10) -> list[Skill]

Search available skills by keyword.

Matches query (case-insensitively) against each skill's name and description. Uses the cached skill list populated during __init__.

Parameters:

Name Type Description Default
query str

Keyword to search for.

 required
limit int

Maximum number of results.

 10

Returns:

Type Description
list[Skill]

List of :class:~pydantic_ai_skills.Skill objects. Each skill's
list[Skill]

metadata dict contains "source_url" for traceability.
Source code in pydantic_ai_skills/registries/git.py 
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search available skills by keyword.

    Matches ``query`` (case-insensitively) against each skill's ``name`` and
    ``description``. Uses the cached skill list populated during ``__init__``.

    Args:
        query: Keyword to search for.
        limit: Maximum number of results.

    Returns:
        List of :class:`~pydantic_ai_skills.Skill` objects. Each skill's
        ``metadata`` dict contains ``"source_url"`` for traceability.
    """
    q = query.lower()
    results: list[Skill] = []
    for skill in self.get_skills():
        if q in skill.name.lower() or q in (skill.description or '').lower():
            results.append(skill)
            if len(results) >= limit:
                break
    return results

get async 

get(skill_name: str) -> Skill

Return the full skill by name.

Parameters:

Name Type Description Default
skill_name str

Exact skill name (with optional prefix).

 required

Returns:

Type Description
Skill

A fully-parsed :class:~pydantic_ai_skills.Skill with metadata
Skill

containing "source_url".

Raises:

Type Description
KeyError

When no skill with skill_name exists.
Source code in pydantic_ai_skills/registries/git.py 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
async def get(self, skill_name: str) -> Skill:
    """Return the full skill by name.

    Args:
        skill_name: Exact skill name (with optional prefix).

    Returns:
        A fully-parsed :class:`~pydantic_ai_skills.Skill` with ``metadata``
        containing ``"source_url"``.

    Raises:
        KeyError: When no skill with ``skill_name`` exists.
    """
    for skill in self.get_skills():
        if skill.name == skill_name:
            return skill
    raise KeyError(f"Skill '{skill_name}' not found in registry {self._clean_repo_url!r}.")

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Copy a skill from the cloned repository into target_dir.

Clones the repository first if the local cache doesn't exist. Validation is handled by discover_skills() during cache population, so skills in the cache are already validated.

Parameters:

Name Type Description Default
skill_name str

Name of the skill to install.

 required
target_dir str | Path

Destination directory; a skill_name subdirectory is created inside it.

 required

Returns:

Type Description
Path

Path to the installed skill directory (target_dir/skill_name).

Raises:

Type Description
KeyError

When skill_name is not found in the registry.
ValueError

When the destination or source path escapes its expected directory (path traversal protection).
RuntimeError

On git or filesystem errors.
Source code in pydantic_ai_skills/registries/git.py 
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Copy a skill from the cloned repository into ``target_dir``.

    Clones the repository first if the local cache doesn't exist. Validation
    is handled by ``discover_skills()`` during cache population, so skills
    in the cache are already validated.

    Args:
        skill_name: Name of the skill to install.
        target_dir: Destination directory; a ``skill_name`` subdirectory
            is created inside it.

    Returns:
        Path to the installed skill directory (``target_dir/skill_name``).

    Raises:
        KeyError: When ``skill_name`` is not found in the registry.
        ValueError: When the destination or source path escapes its expected
            directory (path traversal protection).
        RuntimeError: On git or filesystem errors.
    """
    # Ensure we have skills loaded (already parsed and validated by discover_skills)
    self._ensure_skills_loaded()

    # Find skill in cache — its uri points to the source directory
    src_skill_dir: Path | None = None
    for skill in self._cached_skills:
        if skill.name == skill_name and skill.uri:
            src_skill_dir = Path(skill.uri)
            break

    if src_skill_dir is None:
        raise KeyError(f"Skill '{skill_name}' not found in repository {self._clean_repo_url!r}.")

    return copy_skill_directory(src_skill_dir, target_dir, skill_name)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Pull the latest changes and re-copy the skill to target_dir.

Performs a git pull on the cached clone before re-installing. Falls back to a fresh install if the skill is not yet installed.

Parameters:

Name Type Description Default
skill_name str

Name of the skill to update.

 required
target_dir str | Path

Directory where the skill was previously installed.

 required

Returns:

Type Description
Path

Path to the updated skill directory.

Raises:

Type Description
KeyError

When skill_name is not found after the pull.
RuntimeError

On git or network errors.
Source code in pydantic_ai_skills/registries/git.py 
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Pull the latest changes and re-copy the skill to ``target_dir``.

    Performs a ``git pull`` on the cached clone before re-installing.
    Falls back to a fresh ``install`` if the skill is not yet installed.

    Args:
        skill_name: Name of the skill to update.
        target_dir: Directory where the skill was previously installed.

    Returns:
        Path to the updated skill directory.

    Raises:
        KeyError: When ``skill_name`` is not found after the pull.
        RuntimeError: On git or network errors.
    """
    dest = Path(target_dir).expanduser().resolve() / skill_name
    if not dest.exists():
        return await self.install(skill_name, target_dir)

    # Pull latest and refresh cache before reinstalling
    self._ensure_cloned()
    self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]
    return await self.install(skill_name, target_dir)

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 
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
@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.

Lists and downloads every object under bucket/prefix into a local cache directory on the first call to search/get/install (or eagerly during __init__ when auto_install=True), then parses the synced tree with :func:~pydantic_ai_skills.discover_skills. Each sync mirrors the remote prefix — the cached subtree is cleared first, so skills removed from the bucket no longer appear locally. Re-syncs on update.

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.

search() and get() return :class:~pydantic_ai_skills.Skill objects parsed from SKILL.md frontmatter + body. Registry-specific metadata (source_url, version, bucket, prefix) is stored in skill.metadata.

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. The synced tree persists across install / update calls but 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
validate bool

Whether to run metadata validation on every discovered SKILL.md after syncing. Mirrors the homonymous flag on :class:~pydantic_ai_skills.SkillsDirectory. Defaults to True.

 True
auto_install bool

When True (default), search and get trigger a sync automatically so the local copy is always up to date. Set to False to skip syncing on construction and on search / get; the registry then reads only what already exists in target_dir. Note that an explicit update() call always contacts S3.

 True

Examples:

Amazon S3 with the ambient credential chain:

from pydantic_ai_skills import SkillsToolset
from pydantic_ai_skills.registries.s3 import S3SkillsRegistry

toolset = SkillsToolset(
    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 
 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
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
class S3SkillsRegistry(SkillRegistry):
    """Skills registry backed by an S3 bucket, downloaded with boto3.

    Lists and downloads every object under ``bucket/prefix`` into a local cache
    directory on the first call to ``search``/``get``/``install`` (or eagerly
    during ``__init__`` when ``auto_install=True``), then parses the synced tree
    with :func:`~pydantic_ai_skills.discover_skills`. Each sync mirrors the
    remote prefix — the cached subtree is cleared first, so skills removed from
    the bucket no longer appear locally. Re-syncs on ``update``.

    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.

    ``search()`` and ``get()`` return :class:`~pydantic_ai_skills.Skill` objects
    parsed from ``SKILL.md`` frontmatter + body. Registry-specific metadata
    (``source_url``, ``version``, ``bucket``, ``prefix``) is stored in
    ``skill.metadata``.

    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. The synced tree
            persists across ``install`` / ``update`` calls but 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]``).
        validate: Whether to run metadata validation on every discovered
            ``SKILL.md`` after syncing. Mirrors the homonymous flag on
            :class:`~pydantic_ai_skills.SkillsDirectory`. Defaults to ``True``.
        auto_install: When ``True`` (default), ``search`` and ``get`` trigger a
            sync automatically so the local copy is always up to date. Set to
            ``False`` to skip syncing on construction and on ``search`` / ``get``;
            the registry then reads only what already exists in ``target_dir``.
            Note that an explicit ``update()`` call always contacts S3.

    Examples:
        Amazon S3 with the ambient credential chain:

        ```python
        from pydantic_ai_skills import SkillsToolset
        from pydantic_ai_skills.registries.s3 import S3SkillsRegistry

        toolset = SkillsToolset(
            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,
        validate: bool = True,
        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._validate = validate
        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()

        self._cached_skills: list[Skill] = []
        if self._auto_install:
            self._sync()
            self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]

    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 _load_skills(self) -> list[Skill]:
        """Discover all skills from the synced cache path."""
        skills_root = self._skills_root()
        if not skills_root.exists():
            return []
        return discover_skills(path=skills_root, validate=self._validate, max_depth=2)

    def _skill_version(self, skill: Skill) -> str | None:
        """Return the latest LastModified across the skill's objects, ISO-formatted.

        Uses the cached object listing from the most recent :meth:`_sync`, so it
        performs no additional S3 calls.
        """
        if not skill.uri or not self._object_modified:
            return None
        try:
            skill_dir = Path(skill.uri).resolve().relative_to(self._target_dir.resolve())
        except ValueError:
            return None
        key_prefix = f'{skill_dir.as_posix()}/'
        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

    def _enrich_metadata(self, skill: Skill) -> Skill:
        """Inject registry-specific keys into ``skill.metadata``."""
        skill_path = f'{self._prefix}/{skill.name}'.strip('/')
        extra: dict[str, Any] = {
            'source_url': f's3://{self._bucket}/{skill_path}',
            'registry': type(self).__name__,
            'bucket': self._bucket,
            'prefix': self._prefix,
            'version': self._skill_version(skill),
        }
        existing = dict(skill.metadata) if skill.metadata else {}
        existing.update(extra)
        return replace(skill, metadata=existing)

    def _ensure_skills_loaded(self) -> None:
        """Populate the skills cache if empty, respecting ``auto_install``."""
        if self._cached_skills:
            return
        if self._auto_install:
            self._sync()
        self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]

    # ------------------------------------------------------------------
    # Synchronous skill access for SkillsToolset integration
    # ------------------------------------------------------------------

    def get_skills(self) -> list[Skill]:
        """Return all skills discovered from the bucket.

        Returns:
            List of enriched :class:`~pydantic_ai_skills.Skill` objects.
        """
        self._ensure_skills_loaded()
        return list(self._cached_skills)

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

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search available skills by keyword.

        Matches ``query`` (case-insensitively) against each skill's ``name`` and
        ``description``.

        Args:
            query: Keyword to search for.
            limit: Maximum number of results.

        Returns:
            List of matching :class:`~pydantic_ai_skills.Skill` objects.
        """
        q = query.lower()
        results: list[Skill] = []
        for skill in self.get_skills():
            if q in skill.name.lower() or q in (skill.description or '').lower():
                results.append(skill)
                if len(results) >= limit:
                    break
        return results

    async def get(self, skill_name: str) -> Skill:
        """Return the full skill by name.

        Args:
            skill_name: Exact skill name.

        Returns:
            A fully-parsed :class:`~pydantic_ai_skills.Skill`.

        Raises:
            KeyError: When no skill with ``skill_name`` exists.
        """
        for skill in self.get_skills():
            if skill.name == skill_name:
                return skill
        raise KeyError(f"Skill '{skill_name}' not found in bucket '{self._bucket}'.")

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Copy a skill from the synced cache into ``target_dir``.

        Args:
            skill_name: Name of the skill to install.
            target_dir: Destination directory; a ``skill_name`` subdirectory is
                created inside it.

        Returns:
            Path to the installed skill directory (``target_dir/skill_name``).

        Raises:
            KeyError: When ``skill_name`` is not found in the registry.
            ValueError: When the destination or source path escapes its expected
                directory (path traversal protection).
        """
        self._ensure_skills_loaded()

        src_skill_dir: Path | None = None
        for skill in self._cached_skills:
            if skill.name == skill_name and skill.uri:
                src_skill_dir = Path(skill.uri)
                break

        if src_skill_dir is None:
            raise KeyError(f"Skill '{skill_name}' not found in bucket '{self._bucket}'.")

        return copy_skill_directory(src_skill_dir, target_dir, skill_name)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Re-sync from S3 and re-copy the skill to ``target_dir``.

        Performs a fresh sync of the cache before re-installing. Falls back to a
        plain ``install`` if the skill is not yet installed.

        Args:
            skill_name: Name of the skill to update.
            target_dir: Directory where the skill was previously installed.

        Returns:
            Path to the updated skill directory.

        Raises:
            KeyError: When ``skill_name`` is not found after the sync.
        """
        dest = Path(target_dir).expanduser().resolve() / skill_name
        if not dest.exists():
            return await self.install(skill_name, target_dir)

        self._sync()
        self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]
        return await self.install(skill_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Return all skills discovered from the bucket.

Returns:

Type Description
list[Skill]

List of enriched :class:~pydantic_ai_skills.Skill objects.
Source code in pydantic_ai_skills/registries/s3.py 
260
261
262
263
264
265
266
267
def get_skills(self) -> list[Skill]:
    """Return all skills discovered from the bucket.

    Returns:
        List of enriched :class:`~pydantic_ai_skills.Skill` objects.
    """
    self._ensure_skills_loaded()
    return list(self._cached_skills)

search async 

search(query: str, limit: int = 10) -> list[Skill]

Search available skills by keyword.

Matches query (case-insensitively) against each skill's name and description.

Parameters:

Name Type Description Default
query str

Keyword to search for.

 required
limit int

Maximum number of results.

 10

Returns:

Type Description
list[Skill]

List of matching :class:~pydantic_ai_skills.Skill objects.
Source code in pydantic_ai_skills/registries/s3.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search available skills by keyword.

    Matches ``query`` (case-insensitively) against each skill's ``name`` and
    ``description``.

    Args:
        query: Keyword to search for.
        limit: Maximum number of results.

    Returns:
        List of matching :class:`~pydantic_ai_skills.Skill` objects.
    """
    q = query.lower()
    results: list[Skill] = []
    for skill in self.get_skills():
        if q in skill.name.lower() or q in (skill.description or '').lower():
            results.append(skill)
            if len(results) >= limit:
                break
    return results

get async 

get(skill_name: str) -> Skill

Return the full skill by name.

Parameters:

Name Type Description Default
skill_name str

Exact skill name.

 required

Returns:

Type Description
Skill

A fully-parsed :class:~pydantic_ai_skills.Skill.

Raises:

Type Description
KeyError

When no skill with skill_name exists.
Source code in pydantic_ai_skills/registries/s3.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
async def get(self, skill_name: str) -> Skill:
    """Return the full skill by name.

    Args:
        skill_name: Exact skill name.

    Returns:
        A fully-parsed :class:`~pydantic_ai_skills.Skill`.

    Raises:
        KeyError: When no skill with ``skill_name`` exists.
    """
    for skill in self.get_skills():
        if skill.name == skill_name:
            return skill
    raise KeyError(f"Skill '{skill_name}' not found in bucket '{self._bucket}'.")

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Copy a skill from the synced cache into target_dir.

Parameters:

Name Type Description Default
skill_name str

Name of the skill to install.

 required
target_dir str | Path

Destination directory; a skill_name subdirectory is created inside it.

 required

Returns:

Type Description
Path

Path to the installed skill directory (target_dir/skill_name).

Raises:

Type Description
KeyError

When skill_name is not found in the registry.
ValueError

When the destination or source path escapes its expected directory (path traversal protection).
Source code in pydantic_ai_skills/registries/s3.py 
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
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Copy a skill from the synced cache into ``target_dir``.

    Args:
        skill_name: Name of the skill to install.
        target_dir: Destination directory; a ``skill_name`` subdirectory is
            created inside it.

    Returns:
        Path to the installed skill directory (``target_dir/skill_name``).

    Raises:
        KeyError: When ``skill_name`` is not found in the registry.
        ValueError: When the destination or source path escapes its expected
            directory (path traversal protection).
    """
    self._ensure_skills_loaded()

    src_skill_dir: Path | None = None
    for skill in self._cached_skills:
        if skill.name == skill_name and skill.uri:
            src_skill_dir = Path(skill.uri)
            break

    if src_skill_dir is None:
        raise KeyError(f"Skill '{skill_name}' not found in bucket '{self._bucket}'.")

    return copy_skill_directory(src_skill_dir, target_dir, skill_name)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Re-sync from S3 and re-copy the skill to target_dir.

Performs a fresh sync of the cache before re-installing. Falls back to a plain install if the skill is not yet installed.

Parameters:

Name Type Description Default
skill_name str

Name of the skill to update.

 required
target_dir str | Path

Directory where the skill was previously installed.

 required

Returns:

Type Description
Path

Path to the updated skill directory.

Raises:

Type Description
KeyError

When skill_name is not found after the sync.
Source code in pydantic_ai_skills/registries/s3.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Re-sync from S3 and re-copy the skill to ``target_dir``.

    Performs a fresh sync of the cache before re-installing. Falls back to a
    plain ``install`` if the skill is not yet installed.

    Args:
        skill_name: Name of the skill to update.
        target_dir: Directory where the skill was previously installed.

    Returns:
        Path to the updated skill directory.

    Raises:
        KeyError: When ``skill_name`` is not found after the sync.
    """
    dest = Path(target_dir).expanduser().resolve() / skill_name
    if not dest.exists():
        return await self.install(skill_name, target_dir)

    self._sync()
    self._cached_skills = [self._enrich_metadata(s) for s in self._load_skills()]
    return await self.install(skill_name, target_dir)

Composition Wrappers

Bases: SkillRegistry

A registry that wraps another registry and delegates to it.

All abstract methods are forwarded to wrapped. Subclasses override only the methods they need to modify.

Example 
class MyCustomRegistry(WrapperRegistry):
    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        results = await self.wrapped.search(query, limit)
        # custom post-processing
        return results
Source code in pydantic_ai_skills/registries/wrapper.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
@dataclass
class WrapperRegistry(SkillRegistry):
    """A registry that wraps another registry and delegates to it.

    All abstract methods are forwarded to ``wrapped``. Subclasses
    override only the methods they need to modify.

    Example:
        ```python
        class MyCustomRegistry(WrapperRegistry):
            async def search(self, query: str, limit: int = 10) -> list[Skill]:
                results = await self.wrapped.search(query, limit)
                # custom post-processing
                return results
        ```
    """

    wrapped: SkillRegistry

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Delegate search to the wrapped registry."""
        return await self.wrapped.search(query, limit)

    async def get(self, skill_name: str) -> Skill:
        """Delegate get to the wrapped registry."""
        return await self.wrapped.get(skill_name)

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Delegate install to the wrapped registry."""
        return await self.wrapped.install(skill_name, target_dir)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Delegate update to the wrapped registry."""
        return await self.wrapped.update(skill_name, target_dir)

    def get_skills(self) -> list[Skill]:
        """Delegate get_skills to the wrapped registry."""
        return self.wrapped.get_skills()

search async 

search(query: str, limit: int = 10) -> list[Skill]

Delegate search to the wrapped registry.

Source code in pydantic_ai_skills/registries/wrapper.py 
38
39
40
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Delegate search to the wrapped registry."""
    return await self.wrapped.search(query, limit)

get async 

get(skill_name: str) -> Skill

Delegate get to the wrapped registry.

Source code in pydantic_ai_skills/registries/wrapper.py 
42
43
44
async def get(self, skill_name: str) -> Skill:
    """Delegate get to the wrapped registry."""
    return await self.wrapped.get(skill_name)

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Delegate install to the wrapped registry.

Source code in pydantic_ai_skills/registries/wrapper.py 
46
47
48
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Delegate install to the wrapped registry."""
    return await self.wrapped.install(skill_name, target_dir)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Delegate update to the wrapped registry.

Source code in pydantic_ai_skills/registries/wrapper.py 
50
51
52
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Delegate update to the wrapped registry."""
    return await self.wrapped.update(skill_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Delegate get_skills to the wrapped registry.

Source code in pydantic_ai_skills/registries/wrapper.py 
54
55
56
def get_skills(self) -> list[Skill]:
    """Delegate get_skills to the wrapped registry."""
    return self.wrapped.get_skills()

Bases: WrapperRegistry

A registry that filters skills using a predicate function.

Only skills for which predicate(skill) returns True are visible through this view. The underlying registry is never modified.

Example 
pdf_only = registry.filtered(lambda s: 'pdf' in s.name)
results = await pdf_only.search('document')
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass
class FilteredRegistry(WrapperRegistry):
    """A registry that filters skills using a predicate function.

    Only skills for which ``predicate(skill)`` returns ``True`` are
    visible through this view. The underlying registry is never modified.

    Example:
        ```python
        pdf_only = registry.filtered(lambda s: 'pdf' in s.name)
        results = await pdf_only.search('document')
        ```
    """

    predicate: Callable[[Skill], bool]

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search the wrapped registry and filter results by predicate.

        Fetches extra results from the inner registry to compensate for
        filtering, then trims to ``limit``.
        """
        results = await self.wrapped.search(query, limit=limit * 5)
        return [s for s in results if self.predicate(s)][:limit]

    async def get(self, skill_name: str) -> Skill:
        """Get a skill by name, raising if it doesn't pass the predicate."""
        skill = await self.wrapped.get(skill_name)
        if not self.predicate(skill):
            raise KeyError(f"Skill '{skill_name}' not found in filtered registry.")
        return skill

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Install a skill after validating it passes the predicate."""
        await self.get(skill_name)  # validates predicate
        return await self.wrapped.install(skill_name, target_dir)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Update a skill after validating it passes the predicate."""
        await self.get(skill_name)  # validates predicate
        return await self.wrapped.update(skill_name, target_dir)

    def get_skills(self) -> list[Skill]:
        """Return only skills that pass the predicate."""
        return [s for s in self.wrapped.get_skills() if self.predicate(s)]

search async 

search(query: str, limit: int = 10) -> list[Skill]

Search the wrapped registry and filter results by predicate.

Fetches extra results from the inner registry to compensate for filtering, then trims to limit.

Source code in pydantic_ai_skills/registries/filtered.py 
36
37
38
39
40
41
42
43
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search the wrapped registry and filter results by predicate.

    Fetches extra results from the inner registry to compensate for
    filtering, then trims to ``limit``.
    """
    results = await self.wrapped.search(query, limit=limit * 5)
    return [s for s in results if self.predicate(s)][:limit]

get async 

get(skill_name: str) -> Skill

Get a skill by name, raising if it doesn't pass the predicate.

Source code in pydantic_ai_skills/registries/filtered.py 
45
46
47
48
49
50
async def get(self, skill_name: str) -> Skill:
    """Get a skill by name, raising if it doesn't pass the predicate."""
    skill = await self.wrapped.get(skill_name)
    if not self.predicate(skill):
        raise KeyError(f"Skill '{skill_name}' not found in filtered registry.")
    return skill

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Install a skill after validating it passes the predicate.

Source code in pydantic_ai_skills/registries/filtered.py 
52
53
54
55
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Install a skill after validating it passes the predicate."""
    await self.get(skill_name)  # validates predicate
    return await self.wrapped.install(skill_name, target_dir)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Update a skill after validating it passes the predicate.

Source code in pydantic_ai_skills/registries/filtered.py 
57
58
59
60
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Update a skill after validating it passes the predicate."""
    await self.get(skill_name)  # validates predicate
    return await self.wrapped.update(skill_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Return only skills that pass the predicate.

Source code in pydantic_ai_skills/registries/filtered.py 
62
63
64
def get_skills(self) -> list[Skill]:
    """Return only skills that pass the predicate."""
    return [s for s in self.wrapped.get_skills() if self.predicate(s)]

Bases: WrapperRegistry

A registry that prepends a prefix to every skill name.

The prefix is added to names returned by search and get, and stripped before delegating install and update to the wrapped registry.

Example 
prefixed = registry.prefixed('anthropic-')
# Skill 'pdf' is now accessible as 'anthropic-pdf'
skill = await prefixed.get('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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@dataclass
class PrefixedRegistry(WrapperRegistry):
    """A registry that prepends a prefix to every skill name.

    The prefix is added to names returned by ``search`` and ``get``,
    and stripped before delegating ``install`` and ``update`` to the
    wrapped registry.

    Example:
        ```python
        prefixed = registry.prefixed('anthropic-')
        # Skill 'pdf' is now accessible as 'anthropic-pdf'
        skill = await prefixed.get('anthropic-pdf')
        ```
    """

    prefix: str

    def _add_prefix(self, skill: Skill) -> Skill:
        """Return a copy of the skill with the prefix prepended to its name."""
        return replace(skill, name=f'{self.prefix}{skill.name}')

    def _strip_prefix(self, name: str) -> str:
        """Remove the prefix from a skill name if present."""
        if name.startswith(self.prefix):
            return name[len(self.prefix) :]
        return name

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search the wrapped registry and prefix all result names."""
        results = await self.wrapped.search(query, limit)
        return [self._add_prefix(s) for s in results]

    async def get(self, skill_name: str) -> Skill:
        """Get a skill by its prefixed name.

        Raises:
            KeyError: When the name doesn't start with the expected prefix.
        """
        if not skill_name.startswith(self.prefix):
            raise KeyError(f"Skill '{skill_name}' not found — expected prefix '{self.prefix}'.")
        inner_name = self._strip_prefix(skill_name)
        skill = await self.wrapped.get(inner_name)
        return self._add_prefix(skill)

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Install a skill, stripping the prefix before delegating."""
        inner_name = self._strip_prefix(skill_name)
        return await self.wrapped.install(inner_name, target_dir)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Update a skill, stripping the prefix before delegating."""
        inner_name = self._strip_prefix(skill_name)
        return await self.wrapped.update(inner_name, target_dir)

    def get_skills(self) -> list[Skill]:
        """Return all skills with the prefix prepended to their names."""
        return [self._add_prefix(s) for s in self.wrapped.get_skills()]

search async 

search(query: str, limit: int = 10) -> list[Skill]

Search the wrapped registry and prefix all result names.

Source code in pydantic_ai_skills/registries/prefixed.py 
47
48
49
50
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search the wrapped registry and prefix all result names."""
    results = await self.wrapped.search(query, limit)
    return [self._add_prefix(s) for s in results]

get async 

get(skill_name: str) -> Skill

Get a skill by its prefixed name.

Raises:

Type Description
KeyError

When the name doesn't start with the expected prefix.
Source code in pydantic_ai_skills/registries/prefixed.py 
52
53
54
55
56
57
58
59
60
61
62
async def get(self, skill_name: str) -> Skill:
    """Get a skill by its prefixed name.

    Raises:
        KeyError: When the name doesn't start with the expected prefix.
    """
    if not skill_name.startswith(self.prefix):
        raise KeyError(f"Skill '{skill_name}' not found — expected prefix '{self.prefix}'.")
    inner_name = self._strip_prefix(skill_name)
    skill = await self.wrapped.get(inner_name)
    return self._add_prefix(skill)

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Install a skill, stripping the prefix before delegating.

Source code in pydantic_ai_skills/registries/prefixed.py 
64
65
66
67
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Install a skill, stripping the prefix before delegating."""
    inner_name = self._strip_prefix(skill_name)
    return await self.wrapped.install(inner_name, target_dir)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Update a skill, stripping the prefix before delegating.

Source code in pydantic_ai_skills/registries/prefixed.py 
69
70
71
72
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Update a skill, stripping the prefix before delegating."""
    inner_name = self._strip_prefix(skill_name)
    return await self.wrapped.update(inner_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Return all skills with the prefix prepended to their names.

Source code in pydantic_ai_skills/registries/prefixed.py 
74
75
76
def get_skills(self) -> list[Skill]:
    """Return all skills with the prefix prepended to their names."""
    return [self._add_prefix(s) for s in self.wrapped.get_skills()]

Bases: WrapperRegistry

A registry that renames skills using a name map.

name_map maps new names to original names: {'new-name': 'original-name'}. Skills not present in the map keep their original names.

Example 
renamed = registry.renamed({'doc-tool': 'pdf', 'sheet-tool': 'xlsx'})
skill = await renamed.get('doc-tool')    # fetches 'pdf'
skill = await renamed.get('xlsx')         # still works (unmapped)
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
75
76
@dataclass
class RenamedRegistry(WrapperRegistry):
    """A registry that renames skills using a name map.

    ``name_map`` maps **new names to original names**:
    ``{'new-name': 'original-name'}``.  Skills not present in the
    map keep their original names.

    Example:
        ```python
        renamed = registry.renamed({'doc-tool': 'pdf', 'sheet-tool': 'xlsx'})
        skill = await renamed.get('doc-tool')    # fetches 'pdf'
        skill = await renamed.get('xlsx')         # still works (unmapped)
        ```
    """

    name_map: dict[str, str]

    @property
    def _reverse_map(self) -> dict[str, str]:
        """Map from original name → new name."""
        return {v: k for k, v in self.name_map.items()}

    def _to_new_name(self, skill: Skill) -> Skill:
        """Apply the rename mapping to a skill, if applicable."""
        new_name = self._reverse_map.get(skill.name)
        if new_name:
            return replace(skill, name=new_name)
        return skill

    def _to_original_name(self, name: str) -> str:
        """Resolve a possibly-renamed name back to the original."""
        return self.name_map.get(name, name)

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search the wrapped registry and apply renames to results."""
        results = await self.wrapped.search(query, limit)
        return [self._to_new_name(s) for s in results]

    async def get(self, skill_name: str) -> Skill:
        """Get a skill by its (possibly renamed) name."""
        original_name = self._to_original_name(skill_name)
        skill = await self.wrapped.get(original_name)
        return self._to_new_name(skill)

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Install a skill, resolving the renamed name first."""
        original_name = self._to_original_name(skill_name)
        return await self.wrapped.install(original_name, target_dir)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Update a skill, resolving the renamed name first."""
        original_name = self._to_original_name(skill_name)
        return await self.wrapped.update(original_name, target_dir)

    def get_skills(self) -> list[Skill]:
        """Return all skills with renamed names applied."""
        return [self._to_new_name(s) for s in self.wrapped.get_skills()]

search async 

search(query: str, limit: int = 10) -> list[Skill]

Search the wrapped registry and apply renames to results.

Source code in pydantic_ai_skills/registries/renamed.py 
53
54
55
56
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search the wrapped registry and apply renames to results."""
    results = await self.wrapped.search(query, limit)
    return [self._to_new_name(s) for s in results]

get async 

get(skill_name: str) -> Skill

Get a skill by its (possibly renamed) name.

Source code in pydantic_ai_skills/registries/renamed.py 
58
59
60
61
62
async def get(self, skill_name: str) -> Skill:
    """Get a skill by its (possibly renamed) name."""
    original_name = self._to_original_name(skill_name)
    skill = await self.wrapped.get(original_name)
    return self._to_new_name(skill)

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Install a skill, resolving the renamed name first.

Source code in pydantic_ai_skills/registries/renamed.py 
64
65
66
67
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Install a skill, resolving the renamed name first."""
    original_name = self._to_original_name(skill_name)
    return await self.wrapped.install(original_name, target_dir)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Update a skill, resolving the renamed name first.

Source code in pydantic_ai_skills/registries/renamed.py 
69
70
71
72
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Update a skill, resolving the renamed name first."""
    original_name = self._to_original_name(skill_name)
    return await self.wrapped.update(original_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Return all skills with renamed names applied.

Source code in pydantic_ai_skills/registries/renamed.py 
74
75
76
def get_skills(self) -> list[Skill]:
    """Return all skills with renamed names applied."""
    return [self._to_new_name(s) for s in self.wrapped.get_skills()]

Bases: SkillRegistry

A registry that aggregates multiple registries into one.

Searches are fanned out to every child registry in parallel. get, install, and update try each registry in order and return the first successful result.

Example 
from pydantic_ai_skills.registries import CombinedRegistry

combined = CombinedRegistry(registries=[github_registry, gitlab_registry])
results = await combined.search('pdf')
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
 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
@dataclass
class CombinedRegistry(SkillRegistry):
    """A registry that aggregates multiple registries into one.

    Searches are fanned out to every child registry in parallel.
    ``get``, ``install``, and ``update`` try each registry in order
    and return the first successful result.

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

        combined = CombinedRegistry(registries=[github_registry, gitlab_registry])
        results = await combined.search('pdf')
        ```
    """

    registries: Sequence[SkillRegistry]

    async def search(self, query: str, limit: int = 10) -> list[Skill]:
        """Search all child registries in parallel and merge results.

        Results are deduplicated by skill name (first occurrence wins).
        """
        all_results = await asyncio.gather(*(reg.search(query, limit) for reg in self.registries))
        seen: set[str] = set()
        merged: list[Skill] = []
        for results in all_results:
            for skill in results:
                if skill.name not in seen:
                    seen.add(skill.name)
                    merged.append(skill)
                if len(merged) >= limit:
                    return merged
        return merged

    def _find_owner(self, skill_name: str) -> SkillRegistry | None:
        """Return the first child registry that holds ``skill_name``, or None.

        Uses the synchronous ``get_skills()`` cache so we don't swallow
        unrelated ``KeyError``s from a child registry's internal logic.
        """
        for reg in self.registries:
            for skill in reg.get_skills():
                if skill.name == skill_name:
                    return reg
        return None

    async def get(self, skill_name: str) -> Skill:
        """Try each registry in order and return the first match.

        Raises:
            KeyError: When no registry contains the skill.
        """
        owner = self._find_owner(skill_name)
        if owner is None:
            raise KeyError(f"Skill '{skill_name}' not found in any of the {len(self.registries)} combined registries.")
        return await owner.get(skill_name)

    async def install(self, skill_name: str, target_dir: str | Path) -> Path:
        """Install from the first registry that contains the skill.

        Raises:
            KeyError: When no registry contains the skill.
        """
        owner = self._find_owner(skill_name)
        if owner is None:
            raise KeyError(f"Skill '{skill_name}' not found in any combined registry for install.")
        return await owner.install(skill_name, target_dir)

    async def update(self, skill_name: str, target_dir: str | Path) -> Path:
        """Update from the first registry that contains the skill.

        Raises:
            KeyError: When no registry contains the skill.
        """
        owner = self._find_owner(skill_name)
        if owner is None:
            raise KeyError(f"Skill '{skill_name}' not found in any combined registry for update.")
        return await owner.update(skill_name, target_dir)

    def get_skills(self) -> list[Skill]:
        """Return skills from all child registries, deduplicated by name.

        First occurrence wins when multiple registries provide skills with
        the same name.
        """
        seen: set[str] = set()
        merged: list[Skill] = []
        for reg in self.registries:
            for skill in reg.get_skills():
                if skill.name not in seen:
                    seen.add(skill.name)
                    merged.append(skill)
        return merged

search async 

search(query: str, limit: int = 10) -> list[Skill]

Search all child registries in parallel and merge results.

Results are deduplicated by skill name (first occurrence wins).

Source code in pydantic_ai_skills/registries/combined.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
async def search(self, query: str, limit: int = 10) -> list[Skill]:
    """Search all child registries in parallel and merge results.

    Results are deduplicated by skill name (first occurrence wins).
    """
    all_results = await asyncio.gather(*(reg.search(query, limit) for reg in self.registries))
    seen: set[str] = set()
    merged: list[Skill] = []
    for results in all_results:
        for skill in results:
            if skill.name not in seen:
                seen.add(skill.name)
                merged.append(skill)
            if len(merged) >= limit:
                return merged
    return merged

get async 

get(skill_name: str) -> Skill

Try each registry in order and return the first match.

Raises:

Type Description
KeyError

When no registry contains the skill.
Source code in pydantic_ai_skills/registries/combined.py 
69
70
71
72
73
74
75
76
77
78
async def get(self, skill_name: str) -> Skill:
    """Try each registry in order and return the first match.

    Raises:
        KeyError: When no registry contains the skill.
    """
    owner = self._find_owner(skill_name)
    if owner is None:
        raise KeyError(f"Skill '{skill_name}' not found in any of the {len(self.registries)} combined registries.")
    return await owner.get(skill_name)

install async 

install(skill_name: str, target_dir: str | Path) -> Path

Install from the first registry that contains the skill.

Raises:

Type Description
KeyError

When no registry contains the skill.
Source code in pydantic_ai_skills/registries/combined.py 
80
81
82
83
84
85
86
87
88
89
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Install from the first registry that contains the skill.

    Raises:
        KeyError: When no registry contains the skill.
    """
    owner = self._find_owner(skill_name)
    if owner is None:
        raise KeyError(f"Skill '{skill_name}' not found in any combined registry for install.")
    return await owner.install(skill_name, target_dir)

update async 

update(skill_name: str, target_dir: str | Path) -> Path

Update from the first registry that contains the skill.

Raises:

Type Description
KeyError

When no registry contains the skill.
Source code in pydantic_ai_skills/registries/combined.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def update(self, skill_name: str, target_dir: str | Path) -> Path:
    """Update from the first registry that contains the skill.

    Raises:
        KeyError: When no registry contains the skill.
    """
    owner = self._find_owner(skill_name)
    if owner is None:
        raise KeyError(f"Skill '{skill_name}' not found in any combined registry for update.")
    return await owner.update(skill_name, target_dir)

get_skills 

get_skills() -> list[Skill]

Return skills from all child registries, deduplicated by name.

First occurrence wins when multiple registries provide skills with the same name.

Source code in pydantic_ai_skills/registries/combined.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def get_skills(self) -> list[Skill]:
    """Return skills from all child registries, deduplicated by name.

    First occurrence wins when multiple registries provide skills with
    the same name.
    """
    seen: set[str] = set()
    merged: list[Skill] = []
    for reg in self.registries:
        for skill in reg.get_skills():
            if skill.name not in seen:
                seen.add(skill.name)
                merged.append(skill)
    return merged