datarobot_genai.drtools.core.clients.atlassian

atlassian

Atlassian API client utilities for OAuth Bearer and API token Basic auth.

AtlassianAuthMode

Bases: StrEnum

How Atlassian REST calls are authenticated.

Source code in datarobot_genai/drtools/core/clients/atlassian.py

class AtlassianAuthMode(StrEnum):
    """How Atlassian REST calls are authenticated."""

    OAUTH_BEARER = "oauth_bearer"
    API_TOKEN_BASIC = "api_token_basic"

AtlassianAuth dataclass

Resolved Atlassian credentials for Jira/Confluence clients.

Source code in datarobot_genai/drtools/core/clients/atlassian.py

@dataclass(frozen=True)
class AtlassianAuth:
    """Resolved Atlassian credentials for Jira/Confluence clients."""

    mode: AtlassianAuthMode
    token: str
    email: str = ""
    site_url: str = ""
    cloud_id: str = ""

    @classmethod
    def oauth_bearer(cls, token: str, *, cloud_id: str = "") -> "AtlassianAuth":
        return cls(mode=AtlassianAuthMode.OAUTH_BEARER, token=token, cloud_id=cloud_id)

    @classmethod
    def api_token_basic(
        cls,
        *,
        email: str,
        api_token: str,
        site_url: str,
        cloud_id: str = "",
    ) -> "AtlassianAuth":
        return cls(
            mode=AtlassianAuthMode.API_TOKEN_BASIC,
            token=api_token,
            email=email,
            site_url=site_url,
            cloud_id=cloud_id,
        )

    def authorization_header(self) -> str:
        if self.mode == AtlassianAuthMode.API_TOKEN_BASIC:
            raw = f"{self.email}:{self.token}".encode()
            encoded = base64.b64encode(raw).decode("ascii")
            return f"Basic {encoded}"
        return f"Bearer {self.token}"

    def http_headers(self) -> dict[str, str]:
        return {
            "Authorization": self.authorization_header(),
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

normalize_atlassian_site_url

normalize_atlassian_site_url(site_url: str) -> str

Normalize a Jira/Confluence Cloud site URL (https://acme.atlassian.net).

Source code in datarobot_genai/drtools/core/clients/atlassian.py

def normalize_atlassian_site_url(site_url: str) -> str:
    """Normalize a Jira/Confluence Cloud site URL (``https://acme.atlassian.net``)."""
    url = site_url.strip()
    if not url:
        return ""
    if not url.startswith(("http://", "https://")):
        url = f"https://{url}"
    return url.rstrip("/")

resolve_config_atlassian_auth

resolve_config_atlassian_auth() -> AtlassianAuth | ToolError

Build Atlassian auth from config/env when auth_resolution_strategy=config.

Source code in datarobot_genai/drtools/core/clients/atlassian.py

def resolve_config_atlassian_auth() -> AtlassianAuth | ToolError:
    """Build Atlassian auth from config/env when ``auth_resolution_strategy=config``."""
    creds = get_credentials()
    api_token = creds.atlassian_api_token.strip()
    if not api_token:
        return ToolError(
            "No Atlassian API token configured. Set ATLASSIAN_API_TOKEN.",
            kind=ToolErrorKind.AUTHENTICATION,
        )

    email = creds.atlassian_email.strip()
    if email:
        site_url = normalize_atlassian_site_url(creds.atlassian_site_url)
        if not site_url:
            return ToolError(
                "ATLASSIAN_SITE_URL is required when ATLASSIAN_EMAIL is set "
                "(API token Basic auth). Example: https://your-domain.atlassian.net",
                kind=ToolErrorKind.AUTHENTICATION,
            )
        return AtlassianAuth.api_token_basic(
            email=email,
            api_token=api_token,
            site_url=site_url,
        )

    return AtlassianAuth.oauth_bearer(api_token)

get_jira_access_token async

get_jira_access_token() -> AtlassianAuth | ToolError

Resolve Jira Cloud credentials (OAuth Bearer or API token Basic in config mode).

HTTP strategy: DataRobot OBO for provider jira, then x-datarobot-jira-access-token header fallback.

Config strategy: ATLASSIAN_API_TOKEN with optional ATLASSIAN_EMAIL / ATLASSIAN_SITE_URL for API token Basic auth; token alone is treated as a static OAuth access token (Bearer).

Source code in datarobot_genai/drtools/core/clients/atlassian.py

async def get_jira_access_token() -> AtlassianAuth | ToolError:
    """
    Resolve Jira Cloud credentials (OAuth Bearer or API token Basic in config mode).

    HTTP strategy: DataRobot OBO for provider ``jira``, then
    ``x-datarobot-jira-access-token`` header fallback.

    Config strategy: ``ATLASSIAN_API_TOKEN`` with optional ``ATLASSIAN_EMAIL`` /
    ``ATLASSIAN_SITE_URL`` for API token Basic auth; token alone is treated as a
    static OAuth access token (Bearer).
    """
    creds = get_credentials()
    if creds.auth_resolution_strategy == AuthResolutionStrategy.CONFIG:
        return resolve_config_atlassian_auth()

    token = await get_oauth_access_token_with_header_fallback(
        "jira",
        display_name="Jira",
        access_token_header_segment="jira",
    )
    if isinstance(token, ToolError):
        return token
    return AtlassianAuth.oauth_bearer(token)

get_confluence_access_token async

get_confluence_access_token() -> AtlassianAuth | ToolError

Resolve Confluence Cloud credentials (OAuth Bearer or API token Basic in config mode).

HTTP strategy: DataRobot OBO for provider confluence, then x-datarobot-confluence-access-token header fallback.

Config strategy: same as :func:get_jira_access_token (shared Atlassian site credentials).

Source code in datarobot_genai/drtools/core/clients/atlassian.py

async def get_confluence_access_token() -> AtlassianAuth | ToolError:
    """
    Resolve Confluence Cloud credentials (OAuth Bearer or API token Basic in config mode).

    HTTP strategy: DataRobot OBO for provider ``confluence``, then
    ``x-datarobot-confluence-access-token`` header fallback.

    Config strategy: same as :func:`get_jira_access_token` (shared Atlassian site credentials).
    """
    creds = get_credentials()
    if creds.auth_resolution_strategy == AuthResolutionStrategy.CONFIG:
        return resolve_config_atlassian_auth()

    token = await get_oauth_access_token_with_header_fallback(
        "confluence",
        display_name="Confluence",
        access_token_header_segment="confluence",
    )
    if isinstance(token, ToolError):
        return token
    return AtlassianAuth.oauth_bearer(token)

get_atlassian_cloud_id_from_site async

get_atlassian_cloud_id_from_site(client: AsyncClient, site_url: str) -> str

Resolve cloud ID for API token auth via {site}/_edge/tenant_info.

Parameters:

Name	Type	Description	Default
client	AsyncClient	HTTP client with Basic auth headers configured	required
site_url	str	Normalized site URL (e.g. https://acme.atlassian.net)	required

Returns

Cloud ID string for the Atlassian Cloud site

Raises

ValueError: If cloud ID cannot be retrieved

Source code in datarobot_genai/drtools/core/clients/atlassian.py

async def get_atlassian_cloud_id_from_site(
    client: httpx.AsyncClient,
    site_url: str,
) -> str:
    """
    Resolve cloud ID for API token auth via ``{site}/_edge/tenant_info``.

    Args:
        client: HTTP client with Basic auth headers configured
        site_url: Normalized site URL (e.g. ``https://acme.atlassian.net``)

    Returns
    -------
        Cloud ID string for the Atlassian Cloud site

    Raises
    ------
        ValueError: If cloud ID cannot be retrieved
    """
    url = f"{site_url.rstrip('/')}{TENANT_INFO_PATH}"

    try:
        response = await client.get(url)
        response.raise_for_status()
        data = response.json()
        cloud_id = data.get("cloudId") if isinstance(data, dict) else None
        if not cloud_id:
            raise ValueError("No cloudId in tenant_info response")
        logger.debug("Resolved Atlassian cloud ID from tenant_info: %s", cloud_id)
        return str(cloud_id)
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 401:
            raise ValueError(
                "Authentication failed for Atlassian API token. "
                "Check ATLASSIAN_EMAIL, ATLASSIAN_API_TOKEN, and ATLASSIAN_SITE_URL."
            ) from e
        logger.error("HTTP error getting cloud ID from tenant_info: %s", e.response.status_code)
        raise ValueError(
            f"Failed to get Atlassian cloud ID from tenant_info: HTTP {e.response.status_code}"
        ) from e
    except httpx.RequestError as e:
        logger.error("Request error getting cloud ID from tenant_info: %s", e)
        raise ValueError("Failed to get Atlassian cloud ID: Network error") from e

get_atlassian_cloud_id async

get_atlassian_cloud_id(client: AsyncClient, service_type: AtlassianServiceType | None = None, *, auth: AtlassianAuth | None = None) -> str

Get the cloud ID for the authenticated Atlassian instance.

OAuth Bearer: https://api.atlassian.com/oauth/token/accessible-resources

API token Basic: {site}/_edge/tenant_info

According to Atlassian documentation, API calls use: https://api.atlassian.com/ex/{service}/{cloudId}/...

Parameters:

Name	Type	Description	Default
client	AsyncClient	HTTP client with authentication headers configured	required
service_type	AtlassianServiceType | None	Optional service type to filter OAuth resources (e.g. jira)	None
auth	AtlassianAuth | None	Optional resolved auth; when mode is API token Basic, uses tenant_info	None

Returns

Cloud ID string for the Atlassian instance

Raises

ValueError: If cloud ID cannot be retrieved

Source code in datarobot_genai/drtools/core/clients/atlassian.py

async def get_atlassian_cloud_id(
    client: httpx.AsyncClient,
    service_type: AtlassianServiceType | None = None,
    *,
    auth: AtlassianAuth | None = None,
) -> str:
    """
    Get the cloud ID for the authenticated Atlassian instance.

    OAuth Bearer: ``https://api.atlassian.com/oauth/token/accessible-resources``

    API token Basic: ``{site}/_edge/tenant_info``

    According to Atlassian documentation, API calls use:
    ``https://api.atlassian.com/ex/{service}/{cloudId}/...``

    Args:
        client: HTTP client with authentication headers configured
        service_type: Optional service type to filter OAuth resources (e.g. ``jira``)
        auth: Optional resolved auth; when mode is API token Basic, uses ``tenant_info``

    Returns
    -------
        Cloud ID string for the Atlassian instance

    Raises
    ------
        ValueError: If cloud ID cannot be retrieved
    """
    if auth and auth.cloud_id:
        return auth.cloud_id

    if auth and auth.mode == AtlassianAuthMode.API_TOKEN_BASIC:
        if not auth.site_url:
            raise ValueError("Atlassian site URL is required for API token cloud ID lookup")
        return await get_atlassian_cloud_id_from_site(client, auth.site_url)

    return await _get_oauth_cloud_id(client, service_type)