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.agents import _api_util
 16from airbyte.agents.models import AgentSkillDocs, AgentSkillInfo, AgentSkillList
 17
 18
 19if TYPE_CHECKING:
 20    from collections.abc import Callable, Iterator
 21
 22    from airbyte.cloud._credentials import _AirbyteCredentials
 23
 24
 25class AgentSkill:
 26    """A skill on the Airbyte Agents platform.
 27
 28    Get one from `AgentWorkspace.get_skill()` or `AgentWorkspace.list_skills()` rather than
 29    constructing it directly.
 30    """
 31
 32    def __init__(
 33        self,
 34        skill_id: str,
 35        *,
 36        credentials: _AirbyteCredentials,
 37        workspace_id: str | None = None,
 38        info: AgentSkillInfo | None = None,
 39    ) -> None:
 40        """Initialize an `AgentSkill`. Prefer `AgentWorkspace.get_skill()`."""
 41        self.skill_id = skill_id
 42        """The skill ID."""
 43
 44        self._credentials = credentials
 45        self._workspace_id = workspace_id
 46        self._info = info
 47
 48    @property
 49    def info(self) -> AgentSkillInfo:
 50        """The skill's metadata, fetched from the Agents API if not already known."""
 51        if self._info is None:
 52            self._info = self.read_docs().metadata
 53        return self._info
 54
 55    @property
 56    def title(self) -> str | None:
 57        """The human-readable skill title."""
 58        return self.info.title
 59
 60    @property
 61    def kind(self) -> str | None:
 62        """The skill category, for example `static` or `connector_source`."""
 63        return self.info.kind
 64
 65    def read_docs(self, *, section: str | None = None) -> AgentSkillDocs:
 66        """Read this skill's docs, optionally scoped to a single section.
 67
 68        Omit `section` for metadata, guidance, and the outline of available sections, or
 69        pass an exact section `id` from the outline to read that section.
 70        """
 71        docs = AgentSkillDocs.model_validate(
 72            _api_util.read_agent_skill_docs(
 73                skill_id=self.skill_id,
 74                credentials=self._credentials,
 75                organization_id=self._credentials.organization_id,
 76                workspace_id=self._workspace_id,
 77                section=section,
 78            )
 79        )
 80        if self._info is None:
 81            self._info = docs.metadata
 82        return docs
 83
 84
 85def list_skills(
 86    *,
 87    credentials: _AirbyteCredentials,
 88    workspace_id: str | None = None,
 89    limit: int | None = None,
 90    cursor: str | None = None,
 91) -> AgentSkillList:
 92    """List the skills available to a workspace or organization.
 93
 94    Pass `limit` to cap the page size and the `next_cursor` of a previous result as
 95    `cursor` to fetch the next page.
 96    """
 97    return AgentSkillList.model_validate(
 98        _api_util.list_agent_skills(
 99            credentials=credentials,
100            organization_id=credentials.organization_id,
101            workspace_id=workspace_id,
102            limit=limit,
103            cursor=cursor,
104        )
105    )
106
107
108def _iter_skill_pages(
109    fetch_page: Callable[[str | None], AgentSkillList],
110) -> Iterator[AgentSkillInfo]:
111    """Yield skills across pages, following `next_cursor` until it is `None`.
112
113    Stops early if the server returns a blank cursor or one already seen, rather than
114    requesting the same page forever.
115    """
116    cursor: str | None = None
117    seen_cursors: set[str] = set()
118    while True:
119        page = fetch_page(cursor)
120        yield from page.data
121        cursor = page.next_cursor
122        if cursor is None or not cursor.strip() or cursor in seen_cursors:
123            return
124        seen_cursors.add(cursor)
125
126
127def iter_skills(
128    *,
129    credentials: _AirbyteCredentials,
130    workspace_id: str | None = None,
131) -> Iterator[AgentSkillInfo]:
132    """Yield all available skills, following the API's pagination cursor.
133
134    This is the pagination-free way to list skills: each page is fetched lazily as the
135    caller iterates, so no cursor bookkeeping is needed.
136    """
137    return _iter_skill_pages(
138        lambda cursor: list_skills(
139            credentials=credentials,
140            workspace_id=workspace_id,
141            cursor=cursor,
142        )
143    )