feat(mcp): add cloud describe mcp tools (#879)

Author: aaronsteers

airbyte/mcp/cloud_ops.py

@@ -64,6 +64,55 @@ class CloudConnectionResult(BaseModel):
     """ID of the destination used by this connection."""
 
 
+class CloudSourceDetails(BaseModel):
+    """Detailed information about a deployed source connector in Airbyte Cloud."""
+
+    source_id: str
+    """The source ID."""
+    source_name: str
+    """Display name of the source."""
+    source_url: str
+    """Web URL for managing this source in Airbyte Cloud."""
+    connector_definition_id: str
+    """The connector definition ID (e.g., the ID for 'source-postgres')."""
+
+
+class CloudDestinationDetails(BaseModel):
+    """Detailed information about a deployed destination connector in Airbyte Cloud."""
+
+    destination_id: str
+    """The destination ID."""
+    destination_name: str
+    """Display name of the destination."""
+    destination_url: str
+    """Web URL for managing this destination in Airbyte Cloud."""
+    connector_definition_id: str
+    """The connector definition ID (e.g., the ID for 'destination-snowflake')."""
+
+
+class CloudConnectionDetails(BaseModel):
+    """Detailed information about a deployed connection in Airbyte Cloud."""
+
+    connection_id: str
+    """The connection ID."""
+    connection_name: str
+    """Display name of the connection."""
+    connection_url: str
+    """Web URL for managing this connection in Airbyte Cloud."""
+    source_id: str
+    """ID of the source used by this connection."""
+    source_name: str
+    """Display name of the source."""
+    destination_id: str
+    """ID of the destination used by this connection."""
+    destination_name: str
+    """Display name of the destination."""
+    selected_streams: list[str]
+    """List of stream names selected for syncing."""
+    table_prefix: str | None
+    """Table prefix applied when syncing to the destination."""
+
+
 def _get_cloud_workspace(workspace_id: str | None = None) -> CloudWorkspace:
     """Get an authenticated CloudWorkspace.
 
@@ -601,6 +650,128 @@ def list_deployed_cloud_destination_connectors(
     ]
 
 
+@mcp_tool(
+    domain="cloud",
+    read_only=True,
+    idempotent=True,
+    open_world=True,
+)
+def describe_cloud_source(
+    source_id: Annotated[
+        str,
+        Field(description="The ID of the source to describe."),
+    ],
+    *,
+    workspace_id: Annotated[
+        str | None,
+        Field(
+            description="Workspace ID. Defaults to AIRBYTE_CLOUD_WORKSPACE_ID env var.",
+            default=None,
+        ),
+    ],
+) -> CloudSourceDetails:
+    """Get detailed information about a specific deployed source connector.
+
+    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
+    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
+    Airbyte Cloud API.
+    """
+    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
+    source = workspace.get_source(source_id=source_id)
+
+    # Access name property to ensure _connector_info is populated
+    source_name = cast(str, source.name)
+
+    return CloudSourceDetails(
+        source_id=source.source_id,
+        source_name=source_name,
+        source_url=source.connector_url,
+        connector_definition_id=source._connector_info.definition_id,  # noqa: SLF001  # type: ignore[union-attr]
+    )
+
+
+@mcp_tool(
+    domain="cloud",
+    read_only=True,
+    idempotent=True,
+    open_world=True,
+)
+def describe_cloud_destination(
+    destination_id: Annotated[
+        str,
+        Field(description="The ID of the destination to describe."),
+    ],
+    *,
+    workspace_id: Annotated[
+        str | None,
+        Field(
+            description="Workspace ID. Defaults to AIRBYTE_CLOUD_WORKSPACE_ID env var.",
+            default=None,
+        ),
+    ],
+) -> CloudDestinationDetails:
+    """Get detailed information about a specific deployed destination connector.
+
+    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
+    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
+    Airbyte Cloud API.
+    """
+    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
+    destination = workspace.get_destination(destination_id=destination_id)
+
+    # Access name property to ensure _connector_info is populated
+    destination_name = cast(str, destination.name)
+
+    return CloudDestinationDetails(
+        destination_id=destination.destination_id,
+        destination_name=destination_name,
+        destination_url=destination.connector_url,
+        connector_definition_id=destination._connector_info.definition_id,  # noqa: SLF001  # type: ignore[union-attr]
+    )
+
+
+@mcp_tool(
+    domain="cloud",
+    read_only=True,
+    idempotent=True,
+    open_world=True,
+)
+def describe_cloud_connection(
+    connection_id: Annotated[
+        str,
+        Field(description="The ID of the connection to describe."),
+    ],
+    *,
+    workspace_id: Annotated[
+        str | None,
+        Field(
+            description="Workspace ID. Defaults to AIRBYTE_CLOUD_WORKSPACE_ID env var.",
+            default=None,
+        ),
+    ],
+) -> CloudConnectionDetails:
+    """Get detailed information about a specific deployed connection.
+
+    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
+    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
+    Airbyte Cloud API.
+    """
+    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
+    connection = workspace.get_connection(connection_id=connection_id)
+
+    return CloudConnectionDetails(
+        connection_id=connection.connection_id,
+        connection_name=cast(str, connection.name),
+        connection_url=cast(str, connection.connection_url),
+        source_id=connection.source_id,
+        source_name=cast(str, connection.source.name),
+        destination_id=connection.destination_id,
+        destination_name=cast(str, connection.destination.name),
+        selected_streams=connection.stream_names,
+        table_prefix=connection.table_prefix,
+    )
+
+
 @mcp_tool(
     domain="cloud",
     read_only=True,