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