feat(config): add standalone and O11y deployment authentication

pyproject.toml

@@ -42,6 +42,7 @@ openai = { version = ">=2.8.0,<3.0.0", optional = true }
 openai-agents = { version = ">=0.4.0,<1.0.0", optional = true }
 litellm = { version = ">=1.83.14,<2.0.0", optional = true, python = ">=3.11,<3.14" }
 galileo-core = "^4.3.0"
+httpx = ">=0.27.0,<0.29.0"
 starlette = { version = ">=0.27.0", optional = true }
 backoff = "^2.2.1"
 crewai = { version = ">=0.152.0,<2.0.0", optional = true, python = ">=3.11,<3.14" }

src/splunk_ao/__init__.py

@@ -47,10 +47,13 @@
 from splunk_ao.schema.metrics import SplunkAOMetrics
 from splunk_ao.shared.base import SyncState
 from splunk_ao.shared.exceptions import (
+    AmbiguousConfigurationError,
     APIError,
     ConfigurationError,
+    MissingConfigurationError,
     ResourceConflictError,
     ResourceNotFoundError,
+    SplunkAOConfigError,
     SplunkAOFutureError,
     ValidationError,
 )
@@ -65,6 +68,7 @@
     "AgentControlTarget",
     "AgentControlTargetUnresolvedError",
     "AgentSpan",
+    "AmbiguousConfigurationError",
     "AnthropicProvider",
     "AuthenticationError",
     "AzureProvider",
@@ -93,6 +97,7 @@
     "MessageRole",
     "Metric",
     "MetricSpec",
+    "MissingConfigurationError",
     "Model",
     "NotFoundError",
     "OpenAIProvider",
@@ -108,6 +113,7 @@
     "Span",
     "SplunkAOAPIError",
     "SplunkAOAgentControlBridge",
+    "SplunkAOConfigError",
     "SplunkAODecorator",
     "SplunkAOFutureError",
     "SplunkAOLogger",

src/splunk_ao/config.py

@@ -1,16 +1,53 @@
 # mypy: disable-error-code=syntax
 # We need to ignore syntax errors until https://github.com/python/mypy/issues/17535 is resolved.
 import os
+from collections.abc import Iterator
 from typing import Any, ClassVar, Optional
 
+from httpx import Response
+from pydantic import SecretStr, ValidationInfo, field_validator, model_validator
 from pydantic_core import Url
 
+from galileo_core.constants.request_method import RequestMethod
+from galileo_core.helpers.api_client import ApiClient
 from galileo_core.schemas.base_config import GalileoConfig
 from splunk_ao.constants import DEFAULT_CONSOLE_URL
-from splunk_ao.shared.exceptions import ConfigurationError
+from splunk_ao.deployment import DeploymentMode, O11yConfig
+from splunk_ao.deployment import resolve_deployment as _resolve_deployment
+from splunk_ao.shared.exceptions import ConfigurationError, MissingConfigurationError
+
+
+class O11yApiClient(ApiClient):
+    """API client for Splunk Observability Cloud AO endpoints."""
+
+    sf_token: SecretStr
+    path_prefix: str = "/v2/ao"
+
+    @property
+    def auth_header(self) -> dict[str, str]:
+        return {"X-SF-Token": self.sf_token.get_secret_value()}
+
+    def _prefixed(self, path: str) -> str:
+        normalized_path = f"/{path.lstrip('/')}"
+        prefix = self.path_prefix.strip("/")
+        if not prefix:
+            return normalized_path
+
+        normalized_prefix = f"/{prefix}"
+        if normalized_path == normalized_prefix or normalized_path.startswith(f"{normalized_prefix}/"):
+            return normalized_path
+        return f"{normalized_prefix}{normalized_path}"
+
+    async def arequest(self, method: RequestMethod, path: str, *args: Any, **kwargs: Any) -> Any:
+        return await super().arequest(method, self._prefixed(path), *args, **kwargs)
+
+    def stream_request(self, method: RequestMethod, path: str, *args: Any, **kwargs: Any) -> Iterator[Response]:
+        return super().stream_request(method, self._prefixed(path), *args, **kwargs)
 
 
 class SplunkAOConfig(GalileoConfig):
+    """Configure authentication and endpoints for standalone and O11y deployments."""
+
     # Config file for this project.
     config_filename: str = "galileo-python-config.json"
     console_url: Url = DEFAULT_CONSOLE_URL
@@ -21,8 +58,60 @@ def reset(self) -> None:
         SplunkAOConfig._instance = None
         super().reset()
 
+    @classmethod
+    def resolve_deployment(cls) -> DeploymentMode:
+        """Infer the deployment mode from configured environment variables."""
+        return _resolve_deployment()
+
+    @classmethod
+    def _is_o11y_env(cls) -> bool:
+        try:
+            return cls.resolve_deployment() == DeploymentMode.O11Y
+        except MissingConfigurationError:
+            return False
+
+    @field_validator("api_url", mode="before")
+    @classmethod
+    def set_api_url(cls, api_url: str | Url | None, info: ValidationInfo) -> Url:
+        """Derive the O11y API URL from its realm and preserve standalone validation."""
+        if cls._is_o11y_env():
+            return Url(O11yConfig.from_env().require_api_url())
+        return super().set_api_url(api_url, info)
+
+    @model_validator(mode="after")
+    def set_jwt_token(self) -> "SplunkAOConfig":
+        """Skip standalone JWT exchange when O11y uses direct SF-token authentication."""
+        if self._is_o11y_env():
+            self.jwt_token = None
+            self.refresh_token = None
+            return self
+        super().set_jwt_token()
+        return self
+
+    @model_validator(mode="after")
+    def set_validated_api_client(self) -> "SplunkAOConfig":
+        """Use the SF-token aware API client for O11y deployments."""
+        if self._is_o11y_env():
+            o11y = O11yConfig.from_env()
+            self.validated_api_client = O11yApiClient(
+                host=o11y.api_root, sf_token=o11y.crud_token, jwt_token=SecretStr(""), ssl_context=self.ssl_context
+            )
+            return self
+        super().set_validated_api_client()
+        return self
+
+    def _uses_o11y_api_client(self) -> bool:
+        return isinstance(self.validated_api_client, O11yApiClient)
+
+    def refresh_jwt_token(self) -> None:
+        """Skip JWT refresh when authenticating directly with an O11y SF token."""
+        if self._uses_o11y_api_client():
+            return
+        super().refresh_jwt_token()
+
     @classmethod
     def get(cls, **kwargs: Any) -> "SplunkAOConfig":
+        """Initialize the shared config with deployment-aware auth validation."""
         if cls._instance is None:
             cls._bridge_env_vars()
             error_message = cls._check_auth_config(kwargs)
@@ -32,8 +121,8 @@ def get(cls, **kwargs: Any) -> "SplunkAOConfig":
         assert cls._instance is not None, "Failed to initialize SplunkAOConfig"
         return cls._instance
 
-    @staticmethod
-    def _bridge_env_vars() -> None:
+    @classmethod
+    def _bridge_env_vars(cls) -> None:
         """Bridge SPLUNK_AO_* env vars into GALILEO_* for galileo-core compatibility.
 
         galileo-core still reads GALILEO_* env vars. Until galileo-core is updated,
@@ -42,7 +131,10 @@ def _bridge_env_vars() -> None:
 
         Only bridges values that are not already set — explicit GALILEO_* overrides win.
         """
-        _BRIDGE = [
+        if cls._is_o11y_env() and "GALILEO_CONSOLE_URL" not in os.environ:
+            os.environ["GALILEO_CONSOLE_URL"] = O11yConfig.from_env().require_console_url()
+
+        bridge = [
             ("SPLUNK_AO_API_KEY", "GALILEO_API_KEY"),
             ("SPLUNK_AO_API_URL", "GALILEO_API_URL"),
             ("SPLUNK_AO_CONSOLE_URL", "GALILEO_CONSOLE_URL"),
@@ -57,7 +149,7 @@ def _bridge_env_vars() -> None:
             ("SPLUNK_AO_PASSWORD", "GALILEO_PASSWORD"),
             ("SPLUNK_AO_MODE", "GALILEO_MODE"),
         ]
-        for new_key, old_key in _BRIDGE:
+        for new_key, old_key in bridge:
             if new_key in os.environ and old_key not in os.environ:
                 os.environ[old_key] = os.environ[new_key]
 
@@ -70,13 +162,15 @@ def _check_auth_config(kwargs: dict) -> str | None:
         message identifying what's missing.
 
         Auth methods supported by the underlying config model:
+          - SF tokens (o11y): SPLUNK_AO_SF_TOKEN and optional SPLUNK_AO_SF_API_TOKEN env vars
           - API key (standalone): api_key kwarg or SPLUNK_AO_API_KEY env
           - Pre-exchanged JWT (standalone): jwt_token or SPLUNK_AO_JWT_TOKEN
           - SSO (paired): sso_id_token + sso_provider, both kwargs and env vars
           - Username/password (paired): username + password, both kwargs and env vars
 
-        Kwargs and env vars are interchangeable — e.g. sso_id_token can come
-        from a kwarg while sso_provider comes from the environment.
+        For standalone auth, kwargs and env vars are interchangeable — e.g.
+        sso_id_token can come from a kwarg while sso_provider comes from the
+        environment. O11y tokens are environment-only.
         """
 
         def _val(kwarg_name: str, env_name: str) -> str | None:
@@ -85,6 +179,9 @@ def _val(kwarg_name: str, env_name: str) -> str | None:
                 return str(value)
             return os.environ.get(env_name)
 
+        if os.environ.get("SPLUNK_AO_SF_TOKEN") or os.environ.get("SPLUNK_AO_SF_API_TOKEN"):
+            return None
+
         # Standalone methods — either alone is sufficient.
         if _val("api_key", "SPLUNK_AO_API_KEY"):
             return None
@@ -128,8 +225,8 @@ def _val(kwarg_name: str, env_name: str) -> str | None:
 
         # Nothing configured anywhere.
         return (
-            "No Splunk AO authentication detected. Set one of: "
-            "SPLUNK_AO_API_KEY; SPLUNK_AO_SSO_ID_TOKEN with SPLUNK_AO_SSO_PROVIDER; "
+            "No Splunk AO authentication detected. Set one of: SPLUNK_AO_REALM with "
+            "SPLUNK_AO_SF_TOKEN; SPLUNK_AO_API_KEY; SPLUNK_AO_SSO_ID_TOKEN with SPLUNK_AO_SSO_PROVIDER; "
             "or SPLUNK_AO_USERNAME with SPLUNK_AO_PASSWORD. "
             "Alternatively, pass the equivalent kwargs to SplunkAOConfig.get(). "
             "See https://docs.splunk.com for setup instructions."

src/splunk_ao/deployment.py

@@ -0,0 +1,136 @@
+"""Deployment detection and mode-specific configuration."""
+
+import os
+from dataclasses import dataclass
+from enum import StrEnum
+
+from pydantic import SecretStr
+
+from splunk_ao.shared.exceptions import AmbiguousConfigurationError, MissingConfigurationError
+
+_O11Y_ENV_VARS = ("SPLUNK_AO_REALM", "SPLUNK_AO_SF_TOKEN", "SPLUNK_AO_SF_API_TOKEN")
+_STANDALONE_ENV_VARS = ("SPLUNK_AO_API_KEY", "SPLUNK_AO_CONSOLE_URL", "SPLUNK_AO_API_URL")
+
+
+class DeploymentMode(StrEnum):
+    """Supported Splunk AO deployment modes."""
+
+    O11Y = "o11y"
+    STANDALONE = "standalone"
+
+
+def _env(name: str) -> str | None:
+    return os.environ.get(name) or None
+
+
+def resolve_deployment() -> DeploymentMode:
+    """Infer the deployment mode from the configured environment variables."""
+    present_o11y = [name for name in _O11Y_ENV_VARS if _env(name)]
+    present_standalone = [name for name in _STANDALONE_ENV_VARS if _env(name)]
+
+    if present_o11y and present_standalone:
+        raise AmbiguousConfigurationError(
+            "Both o11y and standalone configuration detected. "
+            f"O11y variables set: {', '.join(present_o11y)}. "
+            f"Standalone variables set: {', '.join(present_standalone)}."
+        )
+    if present_o11y:
+        return DeploymentMode.O11Y
+    if present_standalone:
+        return DeploymentMode.STANDALONE
+
+    raise MissingConfigurationError(
+        "No Splunk AO deployment configuration detected. Set an o11y variable "
+        f"({', '.join(_O11Y_ENV_VARS)}) or a standalone variable "
+        f"({', '.join(_STANDALONE_ENV_VARS)})."
+    )
+
+
+@dataclass
+class O11yConfig:
+    """Configuration for a Splunk Observability Cloud deployment."""
+
+    realm: str
+    sf_token: SecretStr
+    sf_api_token: SecretStr | None = None
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.sf_token, SecretStr):
+            self.sf_token = SecretStr(self.sf_token)
+        if self.sf_api_token is not None and not isinstance(self.sf_api_token, SecretStr):
+            self.sf_api_token = SecretStr(self.sf_api_token)
+
+    @classmethod
+    def from_env(cls) -> "O11yConfig":
+        """Load and validate o11y configuration from the environment."""
+        realm = _env("SPLUNK_AO_REALM")
+        sf_token = _env("SPLUNK_AO_SF_TOKEN")
+
+        if realm is None or sf_token is None:
+            missing = [
+                name for name, value in (("SPLUNK_AO_REALM", realm), ("SPLUNK_AO_SF_TOKEN", sf_token)) if value is None
+            ]
+            raise MissingConfigurationError(f"O11y deployment requires {' and '.join(missing)} to be set.")
+
+        sf_api_token = _env("SPLUNK_AO_SF_API_TOKEN")
+        return cls(
+            realm=realm, sf_token=SecretStr(sf_token), sf_api_token=SecretStr(sf_api_token) if sf_api_token else None
+        )
+
+    @property
+    def otlp_endpoint(self) -> str:
+        """Return the realm-derived OTLP trace ingest endpoint."""
+        return f"https://ingest.{self.realm}.observability.splunkcloud.com/v2/trace/otlp"
+
+    @property
+    def crud_token(self) -> SecretStr:
+        """Return the API token when set, otherwise the ingest token."""
+        return self.sf_api_token if self.sf_api_token is not None else self.sf_token
+
+    @property
+    def api_root(self) -> str:
+        """Return the realm-derived AO API origin."""
+        return f"https://api.{self.realm}.observability.splunkcloud.com"
+
+    def require_api_url(self) -> str:
+        """Return the realm-derived AO CRUD API URL."""
+        return f"{self.api_root}/v2/ao"
+
+    def require_console_url(self) -> str:
+        """Return the realm-derived AO console URL."""
+        return f"https://app.{self.realm}.observability.splunkcloud.com/#/ao"
+
+
+@dataclass
+class StandaloneConfig:
+    """Configuration for a standalone Splunk AO deployment."""
+
+    api_key: SecretStr
+    console_url: str
+    api_url: str | None = None
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.api_key, SecretStr):
+            self.api_key = SecretStr(self.api_key)
+
+    @classmethod
+    def from_env(cls) -> "StandaloneConfig":
+        """Load and validate standalone configuration from the environment."""
+        api_key = _env("SPLUNK_AO_API_KEY")
+        console_url = _env("SPLUNK_AO_CONSOLE_URL")
+
+        if api_key is None or console_url is None:
+            missing = [
+                name
+                for name, value in (("SPLUNK_AO_API_KEY", api_key), ("SPLUNK_AO_CONSOLE_URL", console_url))
+                if value is None
+            ]
+            raise MissingConfigurationError(f"Standalone deployment requires {' and '.join(missing)} to be set.")
+
+        return cls(api_key=SecretStr(api_key), console_url=console_url, api_url=_env("SPLUNK_AO_API_URL"))
+
+    @property
+    def otlp_endpoint(self) -> str:
+        """Return the explicit or console-derived OTLP trace endpoint."""
+        base = self.api_url or self.console_url.replace("://console.", "://api.", 1).replace("://app.", "://api.", 1)
+        return f"{base.rstrip('/')}/otel/traces"

src/splunk_ao/shared/exceptions.py

@@ -30,6 +30,18 @@ class ValidationError(SplunkAOFutureError):
     """
 
 
+class SplunkAOConfigError(ConfigurationError):
+    """Base exception for deployment configuration errors."""
+
+
+class AmbiguousConfigurationError(SplunkAOConfigError):
+    """Raised when o11y and standalone configuration are both present."""
+
+
+class MissingConfigurationError(SplunkAOConfigError):
+    """Raised when required deployment configuration is absent."""
+
+
 class ResourceNotFoundError(NotFoundError, SplunkAOFutureError):
     """
     Backward-compatible alias for NotFoundError.