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:
            SkillNotFoundError: 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
SkillNotFoundError

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:
        SkillNotFoundError: 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
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
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 SkillRegistryError(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 SkillRegistryError(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 SkillRegistryError(
                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:
            SkillNotFoundError: When no skill with ``skill_name`` exists.
        """
        for skill in self.get_skills():
            if skill.name == skill_name:
                return skill
        raise SkillNotFoundError(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:
            SkillNotFoundError: When ``skill_name`` is not found in the registry.
            SkillRegistryError: 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 SkillNotFoundError(f"Skill '{skill_name}' not found in repository {self._clean_repo_url!r}.")

        dest_root = Path(target_dir).expanduser().resolve()
        dest_root.mkdir(parents=True, exist_ok=True)
        dest_skill_dir = dest_root / skill_name

        # Path traversal check on destination
        if not dest_skill_dir.resolve().is_relative_to(dest_root):
            raise SkillRegistryError(f"Destination path '{dest_skill_dir}' escapes target directory '{dest_root}'.")

        # Validate no source symlinks escape the skill directory
        src_resolved = src_skill_dir.resolve()
        for src_file in src_resolved.rglob('*'):
            if src_file.is_symlink() or src_file.is_file():
                try:
                    src_file.resolve().relative_to(src_resolved)
                except ValueError as exc:
                    raise SkillRegistryError(
                        f"Source path '{src_file}' escapes skill directory (path traversal detected)."
                    ) from exc

        # Copy the skill directory
        if dest_skill_dir.exists():
            shutil.rmtree(dest_skill_dir)
        shutil.copytree(src_resolved, dest_skill_dir)

        return dest_skill_dir

    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:
            SkillNotFoundError: When ``skill_name`` is not found after the pull.
            SkillRegistryError: 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 
491
492
493
494
495
496
497
498
499
500
501
502
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 
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
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
SkillNotFoundError

When no skill with skill_name exists.
Source code in pydantic_ai_skills/registries/git.py 
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
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:
        SkillNotFoundError: When no skill with ``skill_name`` exists.
    """
    for skill in self.get_skills():
        if skill.name == skill_name:
            return skill
    raise SkillNotFoundError(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
SkillNotFoundError

When skill_name is not found in the registry.
SkillRegistryError

On git or filesystem errors.
Source code in pydantic_ai_skills/registries/git.py 
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
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:
        SkillNotFoundError: When ``skill_name`` is not found in the registry.
        SkillRegistryError: 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 SkillNotFoundError(f"Skill '{skill_name}' not found in repository {self._clean_repo_url!r}.")

    dest_root = Path(target_dir).expanduser().resolve()
    dest_root.mkdir(parents=True, exist_ok=True)
    dest_skill_dir = dest_root / skill_name

    # Path traversal check on destination
    if not dest_skill_dir.resolve().is_relative_to(dest_root):
        raise SkillRegistryError(f"Destination path '{dest_skill_dir}' escapes target directory '{dest_root}'.")

    # Validate no source symlinks escape the skill directory
    src_resolved = src_skill_dir.resolve()
    for src_file in src_resolved.rglob('*'):
        if src_file.is_symlink() or src_file.is_file():
            try:
                src_file.resolve().relative_to(src_resolved)
            except ValueError as exc:
                raise SkillRegistryError(
                    f"Source path '{src_file}' escapes skill directory (path traversal detected)."
                ) from exc

    # Copy the skill directory
    if dest_skill_dir.exists():
        shutil.rmtree(dest_skill_dir)
    shutil.copytree(src_resolved, dest_skill_dir)

    return dest_skill_dir

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
SkillNotFoundError

When skill_name is not found after the pull.
SkillRegistryError

On git or network errors.
Source code in pydantic_ai_skills/registries/git.py 
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
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:
        SkillNotFoundError: When ``skill_name`` is not found after the pull.
        SkillRegistryError: 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)

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 
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
@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 SkillNotFoundError(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 
37
38
39
40
41
42
43
44
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 
46
47
48
49
50
51
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 SkillNotFoundError(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 
53
54
55
56
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 
58
59
60
61
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 
63
64
65
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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@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:
            SkillNotFoundError: When the name doesn't start with the
                expected prefix.
        """
        if not skill_name.startswith(self.prefix):
            raise SkillNotFoundError(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 
48
49
50
51
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
SkillNotFoundError

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

    Raises:
        SkillNotFoundError: When the name doesn't start with the
            expected prefix.
    """
    if not skill_name.startswith(self.prefix):
        raise SkillNotFoundError(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 
66
67
68
69
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 
71
72
73
74
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 
76
77
78
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 
 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
@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

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

        Raises:
            SkillNotFoundError: When no registry contains the skill.
        """
        for reg in self.registries:
            try:
                return await reg.get(skill_name)
            except SkillNotFoundError:
                continue
        raise SkillNotFoundError(
            f"Skill '{skill_name}' not found in any of the {len(self.registries)} combined registries."
        )

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

        Raises:
            SkillNotFoundError: When no registry contains the skill.
        """
        for reg in self.registries:
            try:
                await reg.get(skill_name)
                return await reg.install(skill_name, target_dir)
            except SkillNotFoundError:
                continue
        raise SkillNotFoundError(f"Skill '{skill_name}' not found in any combined registry for install.")

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

        Raises:
            SkillNotFoundError: When no registry contains the skill.
        """
        for reg in self.registries:
            try:
                await reg.get(skill_name)
                return await reg.update(skill_name, target_dir)
            except SkillNotFoundError:
                continue
        raise SkillNotFoundError(f"Skill '{skill_name}' not found in any combined registry for update.")

    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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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
SkillNotFoundError

When no registry contains the skill.
Source code in pydantic_ai_skills/registries/combined.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
async def get(self, skill_name: str) -> Skill:
    """Try each registry in order and return the first match.

    Raises:
        SkillNotFoundError: When no registry contains the skill.
    """
    for reg in self.registries:
        try:
            return await reg.get(skill_name)
        except SkillNotFoundError:
            continue
    raise SkillNotFoundError(
        f"Skill '{skill_name}' not found in any of the {len(self.registries)} combined registries."
    )

install async 

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

Install from the first registry that contains the skill.

Raises:

Type Description
SkillNotFoundError

When no registry contains the skill.
Source code in pydantic_ai_skills/registries/combined.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
async def install(self, skill_name: str, target_dir: str | Path) -> Path:
    """Install from the first registry that contains the skill.

    Raises:
        SkillNotFoundError: When no registry contains the skill.
    """
    for reg in self.registries:
        try:
            await reg.get(skill_name)
            return await reg.install(skill_name, target_dir)
        except SkillNotFoundError:
            continue
    raise SkillNotFoundError(f"Skill '{skill_name}' not found in any combined registry for install.")

update async 

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

Update from the first registry that contains the skill.

Raises:

Type Description
SkillNotFoundError

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

    Raises:
        SkillNotFoundError: When no registry contains the skill.
    """
    for reg in self.registries:
        try:
            await reg.get(skill_name)
            return await reg.update(skill_name, target_dir)
        except SkillNotFoundError:
            continue
    raise SkillNotFoundError(f"Skill '{skill_name}' not found in any combined registry for update.")

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 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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