airbyte.agents.workspaces

Airbyte Agents workspaces.

⚠️ Experimental Interface

The Airbyte Agents Python interfaces are experimental. Class names, method signatures, and result models may change or be removed without notice between minor versions of PyAirbyte. Pin an exact PyAirbyte version if you depend on them.

View Source

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

class AgentWorkspace: View Source

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

A workspace on the Airbyte Agents platform.

Airbyte Cloud credentials authenticate against the Agents API, so this class reads the same AIRBYTE_CLOUD_* environment variables as airbyte.cloud.CloudWorkspace.

from airbyte import agents

workspace = agents.AgentWorkspace.from_env()
for connector in workspace.list_connectors():
    print(connector.name)

 50    def __init__(
 51        self,
 52        *,
 53        workspace_id: str | None = None,
 54        organization_id: str | None = None,
 55        name: str | None = None,
 56        client_id: str | SecretString | None = None,
 57        client_secret: str | SecretString | None = None,
 58        bearer_token: str | SecretString | None = None,
 59        public_api_root: str | None = None,
 60        config_api_root: str | None = None,
 61    ) -> None:
 62        """Initialize an `AgentWorkspace`.
 63
 64        Credentials fall back to the `AIRBYTE_CLOUD_*` environment variables when they are
 65        not passed explicitly. API roots default to the `AIRBYTE_CLOUD_API_URL` /
 66        `AIRBYTE_CLOUD_CONFIG_API_URL` environment variables (public Airbyte Cloud when unset);
 67        custom roots require `AIRBYTE_AGENTS_API_URL`.
 68        """
 69        credentials = _AirbyteCredentials.from_auth(
 70            workspace_id=workspace_id,
 71            organization_id=organization_id,
 72            client_id=client_id,
 73            client_secret=client_secret,
 74            bearer_token=bearer_token,
 75            public_api_root=public_api_root,
 76            config_api_root=config_api_root,
 77            # Mirrors `CloudWorkspace.__init__`: any explicit credential disables env
 78            # fallback, since an env bearer token plus explicit client creds is rejected
 79            # as mutually exclusive auth.
 80            env_vars=not (client_id or client_secret or bearer_token),
 81        )
 82        if not credentials.workspace_id:
 83            raise PyAirbyteInputError(
 84                message="Workspace ID is required.",
 85                guidance=(
 86                    "Provide `workspace_id`, or set the `AIRBYTE_CLOUD_WORKSPACE_ID` "
 87                    "environment variable."
 88                ),
 89            )
 90
 91        self._credentials = credentials
 92
 93        self.workspace_id: str = credentials.workspace_id
 94        """The workspace ID."""
 95
 96        self.organization_id: str | None = credentials.organization_id
 97        """The organization ID, sent to the Agents API when it is known."""
 98
 99        self.name: str | None = name
100        """The workspace name, when known. Use `get_info()` to fetch it from the API."""

Initialize an AgentWorkspace.

Credentials fall back to the AIRBYTE_CLOUD_* environment variables when they are not passed explicitly. API roots default to the AIRBYTE_CLOUD_API_URL / AIRBYTE_CLOUD_CONFIG_API_URL environment variables (public Airbyte Cloud when unset); custom roots require AIRBYTE_AGENTS_API_URL.

workspace_id: str

The workspace ID.

organization_id: str | None

The organization ID, sent to the Agents API when it is known.

name: str | None

The workspace name, when known. Use get_info() to fetch it from the API.

@classmethod

def from_env( cls, workspace_id: str | None = None, *, organization_id: str | None = None) -> AgentWorkspace: View Source

102    @classmethod
103    def from_env(
104        cls,
105        workspace_id: str | None = None,
106        *,
107        organization_id: str | None = None,
108    ) -> AgentWorkspace:
109        """Create an `AgentWorkspace` from the `AIRBYTE_CLOUD_*` environment variables.
110
111        The variables used are `AIRBYTE_CLOUD_BEARER_TOKEN` or the
112        `AIRBYTE_CLOUD_CLIENT_ID` and `AIRBYTE_CLOUD_CLIENT_SECRET` pair, along with
113        `AIRBYTE_CLOUD_WORKSPACE_ID` and `AIRBYTE_CLOUD_ORGANIZATION_ID`.
114        """
115        return cls(workspace_id=workspace_id, organization_id=organization_id)

Create an AgentWorkspace from the AIRBYTE_CLOUD_* environment variables.

The variables used are AIRBYTE_CLOUD_BEARER_TOKEN or the AIRBYTE_CLOUD_CLIENT_ID and AIRBYTE_CLOUD_CLIENT_SECRET pair, along with AIRBYTE_CLOUD_WORKSPACE_ID and AIRBYTE_CLOUD_ORGANIZATION_ID.

def get_info(self) -> airbyte._direct_connectors.models.AgentWorkspaceInfo: View Source

117    def get_info(self) -> AgentWorkspaceInfo:
118        """Fetch this workspace from the Agents API.
119
120        A successful call is authoritative proof that the workspace is reachable through
121        the Agents API with these credentials.
122        """
123        return AgentWorkspaceInfo.model_validate(
124            _api_util.get_agent_workspace(
125                workspace_id=self.workspace_id,
126                credentials=self._credentials,
127                organization_id=self.organization_id,
128            )
129        )

Fetch this workspace from the Agents API.

A successful call is authoritative proof that the workspace is reachable through the Agents API with these credentials.

def list_connectors(self) -> list[airbyte.agents.AgentConnector]: View Source

131    def list_connectors(self) -> list[AgentConnector]:
132        """List the connectors configured in this workspace."""
133        return [
134            AgentConnector(
135                connector_id=info.id,
136                name=info.name,
137                credentials=self._credentials,
138                workspace_id=self.workspace_id,
139            )
140            for info in (
141                AgentConnectorInfo.model_validate(record)
142                for record in _api_util.list_agent_connectors(
143                    workspace_id=self.workspace_id,
144                    credentials=self._credentials,
145                    organization_id=self.organization_id,
146                )
147            )
148        ]

List the connectors configured in this workspace.

def list_skills(self) -> list[airbyte.agents.AgentSkill]: View Source

150    def list_skills(self) -> list[AgentSkill]:
151        """List all skills available to this workspace, following pagination."""
152        return [
153            AgentSkill(
154                skill_id=info.id,
155                credentials=self._credentials,
156                workspace_id=self.workspace_id,
157                info=info,
158            )
159            for info in _skills.iter_skills(
160                credentials=self._credentials,
161                workspace_id=self.workspace_id,
162            )
163        ]

List all skills available to this workspace, following pagination.

def get_skill(self, skill_id: str) -> airbyte.agents.AgentSkill: View Source

165    def get_skill(self, skill_id: str) -> AgentSkill:
166        """Get a skill by ID, without calling the Agents API."""
167        return AgentSkill(
168            skill_id,
169            credentials=self._credentials,
170            workspace_id=self.workspace_id,
171        )

Get a skill by ID, without calling the Agents API.

def read_skill_docs( self, skill_id: str, *, section: str | None = None) -> airbyte._direct_connectors.models.AgentSkillDocs: View Source

173    def read_skill_docs(
174        self,
175        skill_id: str,
176        *,
177        section: str | None = None,
178    ) -> AgentSkillDocs:
179        """Read a skill's docs, optionally scoped to a single section.
180
181        Omit `section` for metadata, guidance, and the outline of available sections, or
182        pass an exact section `id` from the outline to read that section. Connector usage
183        docs use the `docs_skill_id` reported by `AgentConnector.inspect()`.
184        """
185        return self.get_skill(skill_id).read_docs(section=section)

Read a skill's docs, optionally scoped to a single section.

Omit section for metadata, guidance, and the outline of available sections, or pass an exact section id from the outline to read that section. Connector usage docs use the docs_skill_id reported by AgentConnector.inspect().

def get_connector( self, id_or_name: str | None = None, /, *, id: str | None = None, connector_id: str | None = None, name: str | None = None) -> airbyte.agents.AgentConnector: View Source

187    def get_connector(
188        self,
189        id_or_name: str | None = None,
190        /,
191        *,
192        id: str | None = None,  # noqa: A002  # Shadows `id` deliberately, as a short alias.
193        connector_id: str | None = None,
194        name: str | None = None,
195    ) -> AgentConnector:
196        """Get a connector in this workspace, by ID or by name.
197
198        Pass a single positional value to look the connector up by either its ID or its
199        name, or name the argument to be explicit: `id` and `connector_id` are synonyms,
200        so pass whichever reads better.
201
202        Lookup by an explicit ID does not call the Agents API. Every other form lists the
203        workspace's connectors and matches on ID first, then on an exact name, then on a
204        unique substring, so `name="GitHub"` finds a connector named
205        `GitHub - <workspace_id>`. Name matching is case-insensitive.
206        """
207        lookup = _resolve_connector_lookup(
208            id_or_name,
209            id=id,
210            connector_id=connector_id,
211            name=name,
212        )
213
214        if lookup.connector_id and not lookup.name:
215            return AgentConnector(
216                connector_id=lookup.connector_id,
217                credentials=self._credentials,
218                workspace_id=self.workspace_id,
219            )
220
221        connectors = self.list_connectors()
222        if lookup.connector_id:
223            id_matches = [
224                connector
225                for connector in connectors
226                if connector.connector_id == lookup.connector_id
227            ]
228            if id_matches:
229                return id_matches[0]
230
231        name_lower = (lookup.name or "").lower()
232        matches = [
233            connector
234            for connector in connectors
235            if connector.name and connector.name.lower() == name_lower
236        ] or [
237            connector
238            for connector in connectors
239            if connector.name and name_lower in connector.name.lower()
240        ]
241        if not matches:
242            raise AirbyteError(
243                message="No connector found with the given ID or name.",
244                guidance="Use `list_connectors()` to see the available connectors.",
245                context={"lookup": lookup.name, "workspace_id": self.workspace_id},
246            )
247        if len(matches) > 1:
248            raise AirbyteError(
249                message="Multiple connectors matched the given name.",
250                guidance="Pass `connector_id`, or a name that matches only one connector.",
251                context={
252                    "name": lookup.name,
253                    "matched_names": [connector.name for connector in matches],
254                },
255            )
256        return matches[0]

Get a connector in this workspace, by ID or by name.

Pass a single positional value to look the connector up by either its ID or its name, or name the argument to be explicit: id and connector_id are synonyms, so pass whichever reads better.

Lookup by an explicit ID does not call the Agents API. Every other form lists the workspace's connectors and matches on ID first, then on an exact name, then on a unique substring, so name="GitHub" finds a connector named GitHub - <workspace_id>. Name matching is case-insensitive.

def as_cloud_workspace(self) -> airbyte.cloud.CloudWorkspace: View Source

258    def as_cloud_workspace(self) -> CloudWorkspace:
259        """Return this workspace as an `airbyte.cloud.CloudWorkspace`.
260
261        Every Agents workspace is also a Cloud workspace, so this conversion always
262        succeeds without calling either API.
263        """
264        return CloudWorkspace(
265            workspace_id=self.workspace_id,
266            client_id=self._credentials.client_id,
267            client_secret=self._credentials.client_secret,
268            bearer_token=self._credentials.bearer_token,
269            api_root=self._credentials.public_api_root,
270            config_api_root=self._credentials.config_api_root,
271        )

Return this workspace as an airbyte.cloud.CloudWorkspace.

Every Agents workspace is also a Cloud workspace, so this conversion always succeeds without calling either API.

@classmethod

def from_cloud_workspace( cls, cloud_workspace: airbyte.cloud.CloudWorkspace, *, organization_id: str | None = None, verify: bool = True) -> AgentWorkspace: View Source

273    @classmethod
274    def from_cloud_workspace(
275        cls,
276        cloud_workspace: CloudWorkspace,
277        *,
278        organization_id: str | None = None,
279        verify: bool = True,
280    ) -> AgentWorkspace:
281        """Return a Cloud workspace as an `AgentWorkspace`.
282
283        Cloud workspace IDs are also Agents workspace IDs, but not every Cloud workspace is
284        reachable through the Agents API: the organization needs an Airbyte Agents
285        subscription. By default this is verified by fetching the workspace from the Agents
286        API, which raises `AirbyteError` when it is not eligible. Pass `verify=False` to
287        skip that call.
288
289        Raises `AirbyteAgentsUnavailableError` when the Cloud workspace uses non-public Cloud API
290        roots unless `AIRBYTE_AGENTS_API_URL` is set; the Cloud API roots are carried over so the
291        token exchange stays on the same deployment.
292        """
293        _api_util.check_public_cloud_api_roots(
294            cloud_workspace._credentials,  # noqa: SLF001  # Same-domain conversion.
295        )
296        workspace = cls(
297            workspace_id=cloud_workspace.workspace_id,
298            organization_id=organization_id,
299            client_id=cloud_workspace.client_id,
300            client_secret=cloud_workspace.client_secret,
301            bearer_token=cloud_workspace.bearer_token,
302            public_api_root=cloud_workspace._credentials.public_api_root,  # noqa: SLF001
303            config_api_root=cloud_workspace._credentials.config_api_root,  # noqa: SLF001
304        )
305        if verify:
306            workspace.get_info()
307        return workspace

Return a Cloud workspace as an AgentWorkspace.

Cloud workspace IDs are also Agents workspace IDs, but not every Cloud workspace is reachable through the Agents API: the organization needs an Airbyte Agents subscription. By default this is verified by fetching the workspace from the Agents API, which raises AirbyteError when it is not eligible. Pass verify=False to skip that call.

Raises AirbyteAgentsUnavailableError when the Cloud workspace uses non-public Cloud API roots unless AIRBYTE_AGENTS_API_URL is set; the Cloud API roots are carried over so the token exchange stays on the same deployment.