airbyte.agents.models

  1# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
  2"""Response models for the Airbyte 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
 10All models allow extra fields, because the Agents API returns rich connector-specific
 11payloads that PyAirbyte deliberately does not attempt to model exhaustively.
 12"""
 13
 14from __future__ import annotations
 15
 16from typing import Any
 17
 18from pydantic import BaseModel, ConfigDict, Field
 19
 20from airbyte.exceptions import PyAirbyteInputError
 21
 22
 23class AgentWorkspaceInfo(BaseModel):
 24    """Summary information about a workspace, as returned by the Agents API."""
 25
 26    model_config = ConfigDict(extra="allow")
 27
 28    id: str
 29    """The workspace ID."""
 30
 31    name: str | None = None
 32    """The workspace name."""
 33
 34    organization_id: str | None = None
 35    """The ID of the organization the workspace belongs to."""
 36
 37    status: str | None = None
 38    """The workspace status, for example `active`."""
 39
 40
 41class AgentConnectorInfo(BaseModel):
 42    """Summary information about a connector, as returned by the Agents API."""
 43
 44    model_config = ConfigDict(extra="allow")
 45
 46    id: str
 47    """The connector ID."""
 48
 49    name: str | None = None
 50    """The connector name, for example `GitHub - <workspace_id>`."""
 51
 52
 53class AgentContextStoreEntity(BaseModel):
 54    """An entity that a connector supports caching in the Airbyte Context Store."""
 55
 56    model_config = ConfigDict(extra="allow")
 57
 58    entity: str
 59    """The entity name, for example `issues`."""
 60
 61    suggested: bool | None = None
 62    """Whether Airbyte suggests caching this entity."""
 63
 64
 65class AgentContextStoreReadiness(BaseModel):
 66    """Context Store readiness information for a connector."""
 67
 68    model_config = ConfigDict(extra="allow")
 69
 70    supported_context_store_entities: list[AgentContextStoreEntity] = Field(default_factory=list)
 71    """The entities this connector can cache in the Context Store."""
 72
 73    configured_cache_entities: list[dict[str, Any]] = Field(default_factory=list)
 74    """The entities currently configured for caching, with their sync status."""
 75
 76
 77class AgentConnectorDetails(BaseModel):
 78    """Connector metadata returned by the Agents API `inspect` endpoint."""
 79
 80    model_config = ConfigDict(extra="allow")
 81
 82    connector_id: str
 83    """The connector ID."""
 84
 85    name: str | None = None
 86    """The connector name."""
 87
 88    workspace_id: str | None = None
 89    """The ID of the workspace the connector belongs to."""
 90
 91    organization_id: str | None = None
 92    """The ID of the organization the connector belongs to."""
 93
 94    source_definition_id: str | None = None
 95    """The ID of the underlying Airbyte source definition."""
 96
 97    source_definition_name: str | None = None
 98    """The name of the underlying Airbyte source definition, for example `GitHub`."""
 99
100    context_store_readiness: AgentContextStoreReadiness | None = None
101    """Context Store readiness information, when reported."""
102
103    warnings: list[Any] = Field(default_factory=list)
104    """Warnings reported by the Agents API, for example degraded capabilities."""
105
106    @property
107    def context_store_entities(self) -> list[str]:
108        """The entity names this connector can cache in the Context Store.
109
110        Note that this lists Context Store-supported entities specifically. The Agents API
111        does not publish an exhaustive list of executable entity and action pairs, so an
112        entity may be executable via `AgentConnector.execute()` without appearing here.
113        """
114        if self.context_store_readiness is None:
115            return []
116        return [
117            entity.entity
118            for entity in self.context_store_readiness.supported_context_store_entities
119        ]
120
121
122class AgentExecutionMetadata(BaseModel):
123    """Metadata describing how an Agents connector action was executed."""
124
125    model_config = ConfigDict(extra="allow")
126
127    connector_instance_id: str | None = None
128    """The connector instance that served the request."""
129
130    execution_time_ms: int | None = None
131    """The server-side execution time, in milliseconds."""
132
133
134class AgentConnectorMetadata(BaseModel):
135    """Connector-reported metadata about a single action's result, including pagination."""
136
137    model_config = ConfigDict(extra="allow")
138
139    has_next_page: bool | None = None
140    """Whether more entities are available after this page, when the connector reports it."""
141
142    end_cursor: str | None = None
143    """The cursor to pass as `cursor` to fetch the next page, when one is available."""
144
145
146class AgentExecuteResult(BaseModel):
147    """The result of executing a single action against an Airbyte Agents connector."""
148
149    model_config = ConfigDict(extra="allow")
150
151    status: str
152    """The execution status reported by the Agents API, for example `success`."""
153
154    result: Any = None
155    """The action's payload. Entity-returning actions put a list of entities here."""
156
157    connector_metadata: AgentConnectorMetadata = Field(default_factory=AgentConnectorMetadata)
158    """Connector-reported metadata about the result, including pagination cursors."""
159
160    execution_metadata: AgentExecutionMetadata = Field(default_factory=AgentExecutionMetadata)
161    """Metadata describing how the action was executed."""
162
163    warning: dict[str, Any] | None = None
164    """A warning reported alongside an otherwise successful result."""
165
166    @property
167    def entities(self) -> list[dict[str, Any]]:
168        """The result as a list of entities.
169
170        Raises `PyAirbyteInputError` if the action did not return a list of entities. Use
171        `result` for actions whose payload is not a list of entities.
172        """
173        if not isinstance(self.result, list):
174            raise PyAirbyteInputError(
175                message="This action did not return a list of entities.",
176                guidance="Use the `result` attribute to read non-entity result payloads.",
177                context={"result_type": type(self.result).__name__},
178            )
179
180        invalid_types = sorted(
181            {type(entity).__name__ for entity in self.result if not isinstance(entity, dict)}
182        )
183        if invalid_types:
184            raise PyAirbyteInputError(
185                message="This action returned a list that is not a list of entities.",
186                guidance="Use the `result` attribute to read non-entity result payloads.",
187                context={"unexpected_item_types": invalid_types},
188            )
189        return self.result
190
191    @property
192    def has_next_page(self) -> bool:
193        """Whether the connector reported more entities after this page."""
194        return bool(self.connector_metadata.has_next_page)
195
196    @property
197    def end_cursor(self) -> str | None:
198        """The cursor for the next page, or `None` when there is no next page."""
199        return self.connector_metadata.end_cursor