airbyte.agents.skills

  1# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
  2"""Airbyte Agents skills: reusable documentation served by the Agents API.
  3
  4> ## ⚠️ Experimental Interface
  5>
  6> **The Airbyte Agents Python interfaces are experimental.** Class names, method signatures,
  7> and result models may change or be removed without notice between minor versions of
  8> PyAirbyte. Pin an exact PyAirbyte version if you depend on them.
  9"""
 10
 11from __future__ import annotations
 12
 13from typing import TYPE_CHECKING
 14
 15from airbyte._direct_connectors import api_util as _api_util
 16from airbyte._direct_connectors import skills as _skills
 17from airbyte._direct_connectors.models import AgentSkillDocs, AgentSkillInfo, AgentSkillList
 18
 19
 20if TYPE_CHECKING:
 21    from collections.abc import Callable, Iterator
 22
 23    from airbyte.cloud._credentials import _AirbyteCredentials
 24
 25
 26class AgentSkill:
 27    """A skill on the Airbyte Agents platform.
 28
 29    Get one from `AgentWorkspace.get_skill()` or `AgentWorkspace.list_skills()` rather than
 30    constructing it directly.
 31    """
 32
 33    def __init__(
 34        self,
 35        skill_id: str,
 36        *,
 37        credentials: _AirbyteCredentials,
 38        workspace_id: str | None = None,
 39        info: AgentSkillInfo | None = None,
 40    ) -> None:
 41        """Initialize an `AgentSkill`. Prefer `AgentWorkspace.get_skill()`."""
 42        self.skill_id = skill_id
 43        """The skill ID."""
 44
 45        self._credentials = credentials
 46        self._workspace_id = workspace_id
 47        self._info = info
 48
 49    @property
 50    def info(self) -> AgentSkillInfo:
 51        """The skill's metadata, fetched from the Agents API if not already known."""
 52        if self._info is None:
 53            self._info = self.read_docs().metadata
 54        return self._info
 55
 56    @property
 57    def title(self) -> str | None:
 58        """The human-readable skill title."""
 59        return self.info.title
 60
 61    @property
 62    def kind(self) -> str | None:
 63        """The skill category, for example `static` or `connector_source`."""
 64        return self.info.kind
 65
 66    def read_docs(self, *, section: str | None = None) -> AgentSkillDocs:
 67        """Read this skill's docs, optionally scoped to a single section.
 68
 69        Omit `section` for metadata, guidance, and the outline of available sections, or
 70        pass an exact section `id` from the outline to read that section.
 71        """
 72        docs = AgentSkillDocs.model_validate(
 73            _api_util.read_agent_skill_docs(
 74                skill_id=self.skill_id,
 75                credentials=self._credentials,
 76                organization_id=self._credentials.organization_id,
 77                workspace_id=self._workspace_id,
 78                section=section,
 79            )
 80        )
 81        if self._info is None:
 82            self._info = docs.metadata
 83        return docs
 84
 85
 86def list_skills(
 87    *,
 88    credentials: _AirbyteCredentials,
 89    workspace_id: str | None = None,
 90    limit: int | None = None,
 91    cursor: str | None = None,
 92) -> AgentSkillList:
 93    """List the skills available to a workspace or organization.
 94
 95    Pass `limit` to cap the page size and the `next_cursor` of a previous result as
 96    `cursor` to fetch the next page.
 97    """
 98    return AgentSkillList.model_validate(
 99        _api_util.list_agent_skills(
100            credentials=credentials,
101            organization_id=credentials.organization_id,
102            workspace_id=workspace_id,
103            limit=limit,
104            cursor=cursor,
105        )
106    )
107
108
109def _iter_skill_pages(
110    fetch_page: Callable[[str | None], AgentSkillList],
111) -> Iterator[AgentSkillInfo]:
112    """Yield skills across pages, following `next_cursor` until it is `None`.
113
114    Stops early if the server returns a blank cursor or one already seen, rather than
115    requesting the same page forever.
116    """
117    return _skills.iter_skill_pages(fetch_page)
118
119
120def iter_skills(
121    *,
122    credentials: _AirbyteCredentials,
123    workspace_id: str | None = None,
124) -> Iterator[AgentSkillInfo]:
125    """Yield all available skills, following the API's pagination cursor.
126
127    This is the pagination-free way to list skills: each page is fetched lazily as the
128    caller iterates, so no cursor bookkeeping is needed.
129    """
130    return _skills.iter_skill_infos(
131        credentials=credentials,
132        workspace_id=workspace_id,
133    )