airbyte.agents.workspaces

  1# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
  2"""Airbyte Agents workspaces.
  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 import skills as _skills
 17from airbyte.agents.connectors import AgentConnector, _resolve_connector_lookup
 18from airbyte.agents.models import (
 19    AgentConnectorInfo,
 20    AgentSkillDocs,
 21    AgentWorkspaceInfo,
 22)
 23from airbyte.agents.skills import AgentSkill
 24from airbyte.cloud._credentials import _AirbyteCredentials
 25from airbyte.cloud.workspaces import CloudWorkspace
 26from airbyte.exceptions import AirbyteError, PyAirbyteInputError
 27
 28
 29if TYPE_CHECKING:
 30    from airbyte.secrets.base import SecretString
 31
 32
 33class AgentWorkspace:
 34    """A workspace on the Airbyte Agents platform.
 35
 36    Airbyte Cloud credentials authenticate against the Agents API, so this class reads the
 37    same `AIRBYTE_CLOUD_*` environment variables as `airbyte.cloud.CloudWorkspace`.
 38
 39    ```python
 40    from airbyte import agents
 41
 42    workspace = agents.AgentWorkspace.from_env()
 43    for connector in workspace.list_connectors():
 44        print(connector.name)
 45    ```
 46    """
 47
 48    def __init__(
 49        self,
 50        *,
 51        workspace_id: str | None = None,
 52        organization_id: str | None = None,
 53        name: str | None = None,
 54        client_id: str | SecretString | None = None,
 55        client_secret: str | SecretString | None = None,
 56        bearer_token: str | SecretString | None = None,
 57        public_api_root: str | None = None,
 58        config_api_root: str | None = None,
 59    ) -> None:
 60        """Initialize an `AgentWorkspace`.
 61
 62        Credentials fall back to the `AIRBYTE_CLOUD_*` environment variables when they are
 63        not passed explicitly. API roots default to the `AIRBYTE_CLOUD_API_URL` /
 64        `AIRBYTE_CLOUD_CONFIG_API_URL` environment variables (public Airbyte Cloud when unset);
 65        custom roots require `AIRBYTE_AGENTS_API_URL`.
 66        """
 67        credentials = _AirbyteCredentials.from_auth(
 68            workspace_id=workspace_id,
 69            organization_id=organization_id,
 70            client_id=client_id,
 71            client_secret=client_secret,
 72            bearer_token=bearer_token,
 73            public_api_root=public_api_root,
 74            config_api_root=config_api_root,
 75            # Mirrors `CloudWorkspace.__init__`: any explicit credential disables env
 76            # fallback, since an env bearer token plus explicit client creds is rejected
 77            # as mutually exclusive auth.
 78            env_vars=not (client_id or client_secret or bearer_token),
 79        )
 80        if not credentials.workspace_id:
 81            raise PyAirbyteInputError(
 82                message="Workspace ID is required.",
 83                guidance=(
 84                    "Provide `workspace_id`, or set the `AIRBYTE_CLOUD_WORKSPACE_ID` "
 85                    "environment variable."
 86                ),
 87            )
 88
 89        self._credentials = credentials
 90
 91        self.workspace_id: str = credentials.workspace_id
 92        """The workspace ID."""
 93
 94        self.organization_id: str | None = credentials.organization_id
 95        """The organization ID, sent to the Agents API when it is known."""
 96
 97        self.name: str | None = name
 98        """The workspace name, when known. Use `get_info()` to fetch it from the API."""
 99
100    @classmethod
101    def from_env(
102        cls,
103        workspace_id: str | None = None,
104        *,
105        organization_id: str | None = None,
106    ) -> AgentWorkspace:
107        """Create an `AgentWorkspace` from the `AIRBYTE_CLOUD_*` environment variables.
108
109        The variables used are `AIRBYTE_CLOUD_BEARER_TOKEN` or the
110        `AIRBYTE_CLOUD_CLIENT_ID` and `AIRBYTE_CLOUD_CLIENT_SECRET` pair, along with
111        `AIRBYTE_CLOUD_WORKSPACE_ID` and `AIRBYTE_CLOUD_ORGANIZATION_ID`.
112        """
113        return cls(workspace_id=workspace_id, organization_id=organization_id)
114
115    def get_info(self) -> AgentWorkspaceInfo:
116        """Fetch this workspace from the Agents API.
117
118        A successful call is authoritative proof that the workspace is reachable through
119        the Agents API with these credentials.
120        """
121        return AgentWorkspaceInfo.model_validate(
122            _api_util.get_agent_workspace(
123                workspace_id=self.workspace_id,
124                credentials=self._credentials,
125                organization_id=self.organization_id,
126            )
127        )
128
129    def list_connectors(self) -> list[AgentConnector]:
130        """List the connectors configured in this workspace."""
131        return [
132            AgentConnector(
133                connector_id=info.id,
134                name=info.name,
135                credentials=self._credentials,
136                workspace_id=self.workspace_id,
137            )
138            for info in (
139                AgentConnectorInfo.model_validate(record)
140                for record in _api_util.list_agent_connectors(
141                    workspace_id=self.workspace_id,
142                    credentials=self._credentials,
143                    organization_id=self.organization_id,
144                )
145            )
146        ]
147
148    def list_skills(self) -> list[AgentSkill]:
149        """List all skills available to this workspace, following pagination."""
150        return [
151            AgentSkill(
152                skill_id=info.id,
153                credentials=self._credentials,
154                workspace_id=self.workspace_id,
155                info=info,
156            )
157            for info in _skills.iter_skills(
158                credentials=self._credentials,
159                workspace_id=self.workspace_id,
160            )
161        ]
162
163    def get_skill(self, skill_id: str) -> AgentSkill:
164        """Get a skill by ID, without calling the Agents API."""
165        return AgentSkill(
166            skill_id,
167            credentials=self._credentials,
168            workspace_id=self.workspace_id,
169        )
170
171    def read_skill_docs(
172        self,
173        skill_id: str,
174        *,
175        section: str | None = None,
176    ) -> AgentSkillDocs:
177        """Read a skill's docs, optionally scoped to a single section.
178
179        Omit `section` for metadata, guidance, and the outline of available sections, or
180        pass an exact section `id` from the outline to read that section. Connector usage
181        docs use the `docs_skill_id` reported by `AgentConnector.inspect()`.
182        """
183        return self.get_skill(skill_id).read_docs(section=section)
184
185    def get_connector(
186        self,
187        id_or_name: str | None = None,
188        /,
189        *,
190        id: str | None = None,  # noqa: A002  # Shadows `id` deliberately, as a short alias.
191        connector_id: str | None = None,
192        name: str | None = None,
193    ) -> AgentConnector:
194        """Get a connector in this workspace, by ID or by name.
195
196        Pass a single positional value to look the connector up by either its ID or its
197        name, or name the argument to be explicit: `id` and `connector_id` are synonyms,
198        so pass whichever reads better.
199
200        Lookup by an explicit ID does not call the Agents API. Every other form lists the
201        workspace's connectors and matches on ID first, then on an exact name, then on a
202        unique substring, so `name="GitHub"` finds a connector named
203        `GitHub - <workspace_id>`. Name matching is case-insensitive.
204        """
205        lookup = _resolve_connector_lookup(
206            id_or_name,
207            id=id,
208            connector_id=connector_id,
209            name=name,
210        )
211
212        if lookup.connector_id and not lookup.name:
213            return AgentConnector(
214                connector_id=lookup.connector_id,
215                credentials=self._credentials,
216                workspace_id=self.workspace_id,
217            )
218
219        connectors = self.list_connectors()
220        if lookup.connector_id:
221            id_matches = [
222                connector
223                for connector in connectors
224                if connector.connector_id == lookup.connector_id
225            ]
226            if id_matches:
227                return id_matches[0]
228
229        name_lower = (lookup.name or "").lower()
230        matches = [
231            connector
232            for connector in connectors
233            if connector.name and connector.name.lower() == name_lower
234        ] or [
235            connector
236            for connector in connectors
237            if connector.name and name_lower in connector.name.lower()
238        ]
239        if not matches:
240            raise AirbyteError(
241                message="No connector found with the given ID or name.",
242                guidance="Use `list_connectors()` to see the available connectors.",
243                context={"lookup": lookup.name, "workspace_id": self.workspace_id},
244            )
245        if len(matches) > 1:
246            raise AirbyteError(
247                message="Multiple connectors matched the given name.",
248                guidance="Pass `connector_id`, or a name that matches only one connector.",
249                context={
250                    "name": lookup.name,
251                    "matched_names": [connector.name for connector in matches],
252                },
253            )
254        return matches[0]
255
256    def as_cloud_workspace(self) -> CloudWorkspace:
257        """Return this workspace as an `airbyte.cloud.CloudWorkspace`.
258
259        Every Agents workspace is also a Cloud workspace, so this conversion always
260        succeeds without calling either API.
261        """
262        return CloudWorkspace(
263            workspace_id=self.workspace_id,
264            client_id=self._credentials.client_id,
265            client_secret=self._credentials.client_secret,
266            bearer_token=self._credentials.bearer_token,
267            api_root=self._credentials.public_api_root,
268            config_api_root=self._credentials.config_api_root,
269        )
270
271    @classmethod
272    def from_cloud_workspace(
273        cls,
274        cloud_workspace: CloudWorkspace,
275        *,
276        organization_id: str | None = None,
277        verify: bool = True,
278    ) -> AgentWorkspace:
279        """Return a Cloud workspace as an `AgentWorkspace`.
280
281        Cloud workspace IDs are also Agents workspace IDs, but not every Cloud workspace is
282        reachable through the Agents API: the organization needs an Airbyte Agents
283        subscription. By default this is verified by fetching the workspace from the Agents
284        API, which raises `AirbyteError` when it is not eligible. Pass `verify=False` to
285        skip that call.
286
287        Raises `AirbyteAgentsUnavailableError` when the Cloud workspace uses non-public Cloud API
288        roots unless `AIRBYTE_AGENTS_API_URL` is set; the Cloud API roots are carried over so the
289        token exchange stays on the same deployment.
290        """
291        _api_util.check_public_cloud_api_roots(
292            cloud_workspace._credentials,  # noqa: SLF001  # Same-domain conversion.
293        )
294        workspace = cls(
295            workspace_id=cloud_workspace.workspace_id,
296            organization_id=organization_id,
297            client_id=cloud_workspace.client_id,
298            client_secret=cloud_workspace.client_secret,
299            bearer_token=cloud_workspace.bearer_token,
300            public_api_root=cloud_workspace._credentials.public_api_root,  # noqa: SLF001
301            config_api_root=cloud_workspace._credentials.config_api_root,  # noqa: SLF001
302        )
303        if verify:
304            workspace.get_info()
305        return workspace