Configuration API Reference

Complete API reference for SDK configuration.

Global Configuration

STKAI module-attribute

STKAI: _STKAI = _STKAI()

Configuration Classes

STKAIConfig dataclass

Global configuration for the stkai SDK.

Aggregates all configuration sections: sdk, auth, rqc, agent, file_upload, and rate_limit. Access via the global STKAI.config property.

Attributes:

Name	Type	Description
sdk	SdkConfig	SDK metadata (version, cli_mode). Read-only.
auth	AuthConfig	Authentication configuration.
rqc	RqcConfig	RemoteQuickCommand configuration.
agent	AgentConfig	Agent configuration.
file_upload	FileUploadConfig	File upload configuration.
rate_limit	RateLimitConfig	Rate limiting configuration for HTTP clients.

Example

from stkai import STKAI STKAI.config.sdk.version '0.2.8' STKAI.config.sdk.cli_mode True STKAI.config.rqc.request_timeout 30 STKAI.config.auth.has_credentials() False STKAI.config.file_upload.base_url 'https://data-integration-api.stackspot.com' STKAI.config.rate_limit.enabled False

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class STKAIConfig:
    """
    Global configuration for the stkai SDK.

    Aggregates all configuration sections: sdk, auth, rqc, agent, file_upload,
    and rate_limit. Access via the global `STKAI.config` property.

    Attributes:
        sdk: SDK metadata (version, cli_mode). Read-only.
        auth: Authentication configuration.
        rqc: RemoteQuickCommand configuration.
        agent: Agent configuration.
        file_upload: File upload configuration.
        rate_limit: Rate limiting configuration for HTTP clients.

    Example:
        >>> from stkai import STKAI
        >>> STKAI.config.sdk.version
        '0.2.8'
        >>> STKAI.config.sdk.cli_mode
        True
        >>> STKAI.config.rqc.request_timeout
        30
        >>> STKAI.config.auth.has_credentials()
        False
        >>> STKAI.config.file_upload.base_url
        'https://data-integration-api.stackspot.com'
        >>> STKAI.config.rate_limit.enabled
        False
    """

    sdk: SdkConfig = field(default_factory=SdkConfig.detect)
    auth: AuthConfig = field(default_factory=AuthConfig)
    rqc: RqcConfig = field(default_factory=RqcConfig)
    agent: AgentConfig = field(default_factory=AgentConfig)
    file_upload: FileUploadConfig = field(default_factory=FileUploadConfig)
    rate_limit: RateLimitConfig = field(default_factory=RateLimitConfig)
    _tracker: STKAIConfigTracker = field(default_factory=STKAIConfigTracker, repr=False)

    @STKAIConfigTracker.track_changes("env")
    def with_env_vars(self) -> STKAIConfig:
        """
        Return a new config with environment variables applied on top.

        Reads STKAI_* environment variables and applies them over the
        current configuration values.

        Returns:
            New STKAIConfig instance with env vars applied.

        Example:
            >>> config = STKAIConfig().with_env_vars()
            >>> # Or apply on top of custom config
            >>> custom = STKAIConfig(rqc=RqcConfig(request_timeout=60))
            >>> final = custom.with_env_vars()
        """
        return STKAIConfig(
            sdk=self.sdk,
            auth=self.auth.with_env_vars(),
            rqc=self.rqc.with_env_vars(),
            agent=self.agent.with_env_vars(),
            file_upload=self.file_upload.with_env_vars(),
            rate_limit=self.rate_limit.with_env_vars(),
        )

    @STKAIConfigTracker.track_changes("CLI")
    def with_cli_defaults(self) -> STKAIConfig:
        """
        Return a new config with CLI-provided values applied.

        CLI values take precedence over env vars. When running in CLI mode,
        the CLI knows the correct endpoints for the current environment.

        Returns:
            New STKAIConfig instance with CLI values applied.

        Example:
            >>> # Apply CLI defaults on top of env vars
            >>> config = STKAIConfig().with_env_vars().with_cli_defaults()
        """
        from stkai._cli import StkCLI

        cli_rqc_overrides: dict[str, Any] = {}
        if codebuddy_base_url := StkCLI.get_codebuddy_base_url():
            cli_rqc_overrides["base_url"] = codebuddy_base_url

        cli_agent_overrides: dict[str, Any] = {}
        if inference_app_base_url := StkCLI.get_inference_app_base_url():
            cli_agent_overrides["base_url"] = inference_app_base_url

        cli_file_upload_overrides: dict[str, Any] = {}
        if data_integration_base_url := StkCLI.get_data_integration_base_url():
            cli_file_upload_overrides["base_url"] = data_integration_base_url

        return STKAIConfig(
            sdk=self.sdk,
            auth=self.auth,
            rqc=self.rqc.with_overrides(cli_rqc_overrides),
            agent=self.agent.with_overrides(cli_agent_overrides),
            file_upload=self.file_upload.with_overrides(cli_file_upload_overrides),
            rate_limit=self.rate_limit,
        )

    @STKAIConfigTracker.track_changes("user")
    def with_section_overrides(
        self,
        *,
        auth: dict[str, Any] | None = None,
        rqc: dict[str, Any] | None = None,
        agent: dict[str, Any] | None = None,
        file_upload: dict[str, Any] | None = None,
        rate_limit: dict[str, Any] | None = None,
    ) -> STKAIConfig:
        """
        Return a new config with overrides applied to nested sections.

        Each section dict is merged with the existing section config,
        only overriding the specified fields.

        Args:
            auth: Authentication config overrides.
            rqc: RemoteQuickCommand config overrides.
            agent: Agent config overrides.
            file_upload: File upload config overrides.
            rate_limit: Rate limiting config overrides.

        Returns:
            New STKAIConfig instance with overrides applied.

        Example:
            >>> config = STKAIConfig()
            >>> custom = config.with_section_overrides(
            ...     rqc={"request_timeout": 60},
            ...     agent={"request_timeout": 120},
            ... )
        """
        return STKAIConfig(
            sdk=self.sdk,
            auth=self.auth.with_overrides(auth or {}),
            rqc=self.rqc.with_overrides(rqc or {}),
            agent=self.agent.with_overrides(agent or {}),
            file_upload=self.file_upload.with_overrides(file_upload or {}),
            rate_limit=self.rate_limit.with_overrides(rate_limit or {}),
        )

    def explain_data(self) -> dict[str, list[ConfigEntry]]:
        """
        Return config data structured for explain output.

        Provides a structured representation of all config values and their
        sources, useful for debugging, testing, or custom formatting.

        Returns:
            Dict mapping section names to list of ConfigEntry objects.

        Example:
            >>> config = STKAIConfig().with_env_vars()
            >>> data = config.explain_data()
            >>> for entry in data["rqc"]:
            ...     print(f"{entry.name}: {entry.value} ({entry.source})")
            request_timeout: 30 (default)
            ...
        """
        result: dict[str, list[ConfigEntry]] = {}

        # SDK section (read-only, not tracked)
        result["sdk"] = [
            ConfigEntry(name=f.name, value=getattr(self.sdk, f.name), source="-")
            for f in fields(self.sdk)
        ]

        for section_name in ("auth", "rqc", "agent", "file_upload", "rate_limit"):
            section_config = getattr(self, section_name)
            section_sources = self._tracker.sources.get(section_name, {})
            result[section_name] = [
                ConfigEntry(
                    name=f.name,
                    value=getattr(section_config, f.name),
                    source=section_sources.get(f.name, "default"),
                )
                for f in fields(section_config)
            ]

        return result

Functions

with_env_vars

with_env_vars() -> STKAIConfig

Return a new config with environment variables applied on top.

Reads STKAI_* environment variables and applies them over the current configuration values.

Returns:

Type	Description
STKAIConfig	New STKAIConfig instance with env vars applied.

Example

config = STKAIConfig().with_env_vars()

Or apply on top of custom config

custom = STKAIConfig(rqc=RqcConfig(request_timeout=60)) final = custom.with_env_vars()

Source code in src/stkai/_config.py

@STKAIConfigTracker.track_changes("env")
def with_env_vars(self) -> STKAIConfig:
    """
    Return a new config with environment variables applied on top.

    Reads STKAI_* environment variables and applies them over the
    current configuration values.

    Returns:
        New STKAIConfig instance with env vars applied.

    Example:
        >>> config = STKAIConfig().with_env_vars()
        >>> # Or apply on top of custom config
        >>> custom = STKAIConfig(rqc=RqcConfig(request_timeout=60))
        >>> final = custom.with_env_vars()
    """
    return STKAIConfig(
        sdk=self.sdk,
        auth=self.auth.with_env_vars(),
        rqc=self.rqc.with_env_vars(),
        agent=self.agent.with_env_vars(),
        file_upload=self.file_upload.with_env_vars(),
        rate_limit=self.rate_limit.with_env_vars(),
    )

with_cli_defaults

with_cli_defaults() -> STKAIConfig

Return a new config with CLI-provided values applied.

CLI values take precedence over env vars. When running in CLI mode, the CLI knows the correct endpoints for the current environment.

Returns:

Type	Description
STKAIConfig	New STKAIConfig instance with CLI values applied.

Example

Apply CLI defaults on top of env vars

config = STKAIConfig().with_env_vars().with_cli_defaults()

Source code in src/stkai/_config.py

@STKAIConfigTracker.track_changes("CLI")
def with_cli_defaults(self) -> STKAIConfig:
    """
    Return a new config with CLI-provided values applied.

    CLI values take precedence over env vars. When running in CLI mode,
    the CLI knows the correct endpoints for the current environment.

    Returns:
        New STKAIConfig instance with CLI values applied.

    Example:
        >>> # Apply CLI defaults on top of env vars
        >>> config = STKAIConfig().with_env_vars().with_cli_defaults()
    """
    from stkai._cli import StkCLI

    cli_rqc_overrides: dict[str, Any] = {}
    if codebuddy_base_url := StkCLI.get_codebuddy_base_url():
        cli_rqc_overrides["base_url"] = codebuddy_base_url

    cli_agent_overrides: dict[str, Any] = {}
    if inference_app_base_url := StkCLI.get_inference_app_base_url():
        cli_agent_overrides["base_url"] = inference_app_base_url

    cli_file_upload_overrides: dict[str, Any] = {}
    if data_integration_base_url := StkCLI.get_data_integration_base_url():
        cli_file_upload_overrides["base_url"] = data_integration_base_url

    return STKAIConfig(
        sdk=self.sdk,
        auth=self.auth,
        rqc=self.rqc.with_overrides(cli_rqc_overrides),
        agent=self.agent.with_overrides(cli_agent_overrides),
        file_upload=self.file_upload.with_overrides(cli_file_upload_overrides),
        rate_limit=self.rate_limit,
    )

with_section_overrides

with_section_overrides(*, auth: dict[str, Any] | None = None, rqc: dict[str, Any] | None = None, agent: dict[str, Any] | None = None, file_upload: dict[str, Any] | None = None, rate_limit: dict[str, Any] | None = None) -> STKAIConfig

Return a new config with overrides applied to nested sections.

Each section dict is merged with the existing section config, only overriding the specified fields.

Parameters:

Name	Type	Description	Default
auth	dict[str, Any] | None	Authentication config overrides.	None
rqc	dict[str, Any] | None	RemoteQuickCommand config overrides.	None
agent	dict[str, Any] | None	Agent config overrides.	None
file_upload	dict[str, Any] | None	File upload config overrides.	None
rate_limit	dict[str, Any] | None	Rate limiting config overrides.	None

Returns:

Type	Description
STKAIConfig	New STKAIConfig instance with overrides applied.

Example

config = STKAIConfig() custom = config.with_section_overrides( ... rqc={"request_timeout": 60}, ... agent={"request_timeout": 120}, ... )

Source code in src/stkai/_config.py

@STKAIConfigTracker.track_changes("user")
def with_section_overrides(
    self,
    *,
    auth: dict[str, Any] | None = None,
    rqc: dict[str, Any] | None = None,
    agent: dict[str, Any] | None = None,
    file_upload: dict[str, Any] | None = None,
    rate_limit: dict[str, Any] | None = None,
) -> STKAIConfig:
    """
    Return a new config with overrides applied to nested sections.

    Each section dict is merged with the existing section config,
    only overriding the specified fields.

    Args:
        auth: Authentication config overrides.
        rqc: RemoteQuickCommand config overrides.
        agent: Agent config overrides.
        file_upload: File upload config overrides.
        rate_limit: Rate limiting config overrides.

    Returns:
        New STKAIConfig instance with overrides applied.

    Example:
        >>> config = STKAIConfig()
        >>> custom = config.with_section_overrides(
        ...     rqc={"request_timeout": 60},
        ...     agent={"request_timeout": 120},
        ... )
    """
    return STKAIConfig(
        sdk=self.sdk,
        auth=self.auth.with_overrides(auth or {}),
        rqc=self.rqc.with_overrides(rqc or {}),
        agent=self.agent.with_overrides(agent or {}),
        file_upload=self.file_upload.with_overrides(file_upload or {}),
        rate_limit=self.rate_limit.with_overrides(rate_limit or {}),
    )

explain_data

explain_data() -> dict[str, list[ConfigEntry]]

Return config data structured for explain output.

Provides a structured representation of all config values and their sources, useful for debugging, testing, or custom formatting.

Returns:

Type	Description
dict[str, list[ConfigEntry]]	Dict mapping section names to list of ConfigEntry objects.

Example

config = STKAIConfig().with_env_vars() data = config.explain_data() for entry in data["rqc"]: ... print(f"{entry.name}: {entry.value} ({entry.source})") request_timeout: 30 (default) ...

Source code in src/stkai/_config.py

def explain_data(self) -> dict[str, list[ConfigEntry]]:
    """
    Return config data structured for explain output.

    Provides a structured representation of all config values and their
    sources, useful for debugging, testing, or custom formatting.

    Returns:
        Dict mapping section names to list of ConfigEntry objects.

    Example:
        >>> config = STKAIConfig().with_env_vars()
        >>> data = config.explain_data()
        >>> for entry in data["rqc"]:
        ...     print(f"{entry.name}: {entry.value} ({entry.source})")
        request_timeout: 30 (default)
        ...
    """
    result: dict[str, list[ConfigEntry]] = {}

    # SDK section (read-only, not tracked)
    result["sdk"] = [
        ConfigEntry(name=f.name, value=getattr(self.sdk, f.name), source="-")
        for f in fields(self.sdk)
    ]

    for section_name in ("auth", "rqc", "agent", "file_upload", "rate_limit"):
        section_config = getattr(self, section_name)
        section_sources = self._tracker.sources.get(section_name, {})
        result[section_name] = [
            ConfigEntry(
                name=f.name,
                value=getattr(section_config, f.name),
                source=section_sources.get(f.name, "default"),
            )
            for f in fields(section_config)
        ]

    return result

SdkConfig dataclass

SDK metadata (read-only, not configurable).

Provides information about the SDK version and runtime environment. These values are automatically detected and cannot be overridden.

Attributes:

Name	Type	Description
version	str	The installed SDK version.
cli_mode	bool	Whether StackSpot CLI (oscli) is available.

Example

from stkai import STKAI STKAI.config.sdk.version '0.2.8' STKAI.config.sdk.cli_mode True

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class SdkConfig:
    """
    SDK metadata (read-only, not configurable).

    Provides information about the SDK version and runtime environment.
    These values are automatically detected and cannot be overridden.

    Attributes:
        version: The installed SDK version.
        cli_mode: Whether StackSpot CLI (oscli) is available.

    Example:
        >>> from stkai import STKAI
        >>> STKAI.config.sdk.version
        '0.2.8'
        >>> STKAI.config.sdk.cli_mode
        True
    """

    version: str
    cli_mode: bool

    @classmethod
    def detect(cls) -> SdkConfig:
        """
        Detect SDK metadata from the runtime environment.

        Returns:
            SdkConfig with version and cli_mode auto-detected.
        """
        from stkai import __version__
        from stkai._cli import StkCLI

        return cls(
            version=__version__,
            cli_mode=StkCLI.is_available(),
        )

Functions

detect classmethod

detect() -> SdkConfig

Detect SDK metadata from the runtime environment.

Returns:

Type	Description
SdkConfig	SdkConfig with version and cli_mode auto-detected.

Source code in src/stkai/_config.py

@classmethod
def detect(cls) -> SdkConfig:
    """
    Detect SDK metadata from the runtime environment.

    Returns:
        SdkConfig with version and cli_mode auto-detected.
    """
    from stkai import __version__
    from stkai._cli import StkCLI

    return cls(
        version=__version__,
        cli_mode=StkCLI.is_available(),
    )

AuthConfig dataclass

Bases: OverridableConfig

Authentication configuration for StackSpot AI.

Credentials are used for standalone authentication without StackSpot CLI. When using oscli-based HTTP clients, authentication is delegated to the CLI.

Attributes:

Name	Type	Description
client_id	str | None	StackSpot client ID for authentication. Env var: STKAI_AUTH_CLIENT_ID
client_secret	str | None	StackSpot client secret for authentication. Env var: STKAI_AUTH_CLIENT_SECRET
token_url	str	OAuth2 token endpoint URL for client credentials flow. Env var: STKAI_AUTH_TOKEN_URL

Example

from stkai import STKAI if STKAI.config.auth.has_credentials(): ... print("Credentials configured")

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class AuthConfig(OverridableConfig):
    """
    Authentication configuration for StackSpot AI.

    Credentials are used for standalone authentication without StackSpot CLI.
    When using oscli-based HTTP clients, authentication is delegated to the CLI.

    Attributes:
        client_id: StackSpot client ID for authentication.
            Env var: STKAI_AUTH_CLIENT_ID

        client_secret: StackSpot client secret for authentication.
            Env var: STKAI_AUTH_CLIENT_SECRET

        token_url: OAuth2 token endpoint URL for client credentials flow.
            Env var: STKAI_AUTH_TOKEN_URL

    Example:
        >>> from stkai import STKAI
        >>> if STKAI.config.auth.has_credentials():
        ...     print("Credentials configured")
    """

    client_id: str | None = field(default=None, metadata={"env": "STKAI_AUTH_CLIENT_ID"})
    client_secret: str | None = field(default=None, metadata={"env": "STKAI_AUTH_CLIENT_SECRET"})
    token_url: str = field(default="https://idm.stackspot.com/stackspot-dev/oidc/oauth/token", metadata={"env": "STKAI_AUTH_TOKEN_URL"})

    def has_credentials(self) -> bool:
        """Check if both client_id and client_secret are set."""
        return bool(self.client_id and self.client_secret)

    def validate(self) -> Self:
        """Validate auth configuration fields."""
        if self.client_id is not None and self.client_id == "":
            raise ConfigValidationError(
                "client_id", self.client_id,
                "Must not be empty string.", section="auth"
            )
        if self.client_secret is not None and self.client_secret == "":
            raise ConfigValidationError(
                "client_secret", self.client_secret,
                "Must not be empty string.", section="auth"
            )
        if self.token_url and not (self.token_url.startswith("http://") or self.token_url.startswith("https://")):
            raise ConfigValidationError(
                "token_url", self.token_url,
                "Must start with 'http://' or 'https://'.", section="auth"
            )
        return self

Functions

has_credentials

has_credentials() -> bool

Check if both client_id and client_secret are set.

Source code in src/stkai/_config.py

def has_credentials(self) -> bool:
    """Check if both client_id and client_secret are set."""
    return bool(self.client_id and self.client_secret)

validate

validate() -> Self

Validate auth configuration fields.

Source code in src/stkai/_config.py

def validate(self) -> Self:
    """Validate auth configuration fields."""
    if self.client_id is not None and self.client_id == "":
        raise ConfigValidationError(
            "client_id", self.client_id,
            "Must not be empty string.", section="auth"
        )
    if self.client_secret is not None and self.client_secret == "":
        raise ConfigValidationError(
            "client_secret", self.client_secret,
            "Must not be empty string.", section="auth"
        )
    if self.token_url and not (self.token_url.startswith("http://") or self.token_url.startswith("https://")):
        raise ConfigValidationError(
            "token_url", self.token_url,
            "Must start with 'http://' or 'https://'.", section="auth"
        )
    return self

RqcConfig dataclass

Bases: OverridableConfig

Configuration for RemoteQuickCommand clients.

These settings are used as defaults when creating RemoteQuickCommand instances without explicitly providing CreateExecutionOptions or GetResultOptions.

Attributes:

Name	Type	Description
request_timeout	int	HTTP request timeout in seconds for API calls. Env var: STKAI_RQC_REQUEST_TIMEOUT
retry_max_retries	int	Maximum retry attempts for failed create-execution calls. Use 0 to disable retries (single attempt only). Use 3 for 4 total attempts (1 original + 3 retries). Env var: STKAI_RQC_RETRY_MAX_RETRIES
retry_initial_delay	float	Initial delay in seconds for the first retry attempt. Subsequent retries use exponential backoff (delay doubles each attempt). Example: with 0.5s initial delay, retries wait 0.5s, 1s, 2s, 4s... Env var: STKAI_RQC_RETRY_INITIAL_DELAY
poll_interval	float	Seconds to wait between polling status checks. Env var: STKAI_RQC_POLL_INTERVAL
poll_max_duration	float	Maximum seconds to wait for execution completion before timing out. Env var: STKAI_RQC_POLL_MAX_DURATION
poll_overload_timeout	float	Maximum seconds to tolerate CREATED status before assuming server overload. Env var: STKAI_RQC_POLL_OVERLOAD_TIMEOUT
max_workers	int	Maximum number of concurrent threads for execute_many(). Env var: STKAI_RQC_MAX_WORKERS
base_url	str	Base URL for the RQC API. If None, uses StackSpot CLI. Env var: STKAI_RQC_BASE_URL

Example

from stkai import STKAI STKAI.config.rqc.request_timeout 30 STKAI.config.rqc.retry_max_retries 3

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class RqcConfig(OverridableConfig):
    """
    Configuration for RemoteQuickCommand clients.

    These settings are used as defaults when creating RemoteQuickCommand
    instances without explicitly providing CreateExecutionOptions or
    GetResultOptions.

    Attributes:
        request_timeout: HTTP request timeout in seconds for API calls.
            Env var: STKAI_RQC_REQUEST_TIMEOUT

        retry_max_retries: Maximum retry attempts for failed create-execution calls.
            Use 0 to disable retries (single attempt only).
            Use 3 for 4 total attempts (1 original + 3 retries).
            Env var: STKAI_RQC_RETRY_MAX_RETRIES

        retry_initial_delay: Initial delay in seconds for the first retry attempt.
            Subsequent retries use exponential backoff (delay doubles each attempt).
            Example: with 0.5s initial delay, retries wait 0.5s, 1s, 2s, 4s...
            Env var: STKAI_RQC_RETRY_INITIAL_DELAY

        poll_interval: Seconds to wait between polling status checks.
            Env var: STKAI_RQC_POLL_INTERVAL

        poll_max_duration: Maximum seconds to wait for execution completion
            before timing out.
            Env var: STKAI_RQC_POLL_MAX_DURATION

        poll_overload_timeout: Maximum seconds to tolerate CREATED status before
            assuming server overload.
            Env var: STKAI_RQC_POLL_OVERLOAD_TIMEOUT

        max_workers: Maximum number of concurrent threads for execute_many().
            Env var: STKAI_RQC_MAX_WORKERS

        base_url: Base URL for the RQC API. If None, uses StackSpot CLI.
            Env var: STKAI_RQC_BASE_URL

    Example:
        >>> from stkai import STKAI
        >>> STKAI.config.rqc.request_timeout
        30
        >>> STKAI.config.rqc.retry_max_retries
        3
    """

    request_timeout: int = field(default=30, metadata={"env": "STKAI_RQC_REQUEST_TIMEOUT"})
    retry_max_retries: int = field(default=3, metadata={"env": "STKAI_RQC_RETRY_MAX_RETRIES"})
    retry_initial_delay: float = field(default=0.5, metadata={"env": "STKAI_RQC_RETRY_INITIAL_DELAY"})
    poll_interval: float = field(default=10.0, metadata={"env": "STKAI_RQC_POLL_INTERVAL"})
    poll_max_duration: float = field(default=600.0, metadata={"env": "STKAI_RQC_POLL_MAX_DURATION"})
    poll_overload_timeout: float = field(default=60.0, metadata={"env": "STKAI_RQC_POLL_OVERLOAD_TIMEOUT"})
    max_workers: int = field(default=8, metadata={"env": "STKAI_RQC_MAX_WORKERS"})
    base_url: str = field(default="https://genai-code-buddy-api.stackspot.com", metadata={"env": "STKAI_RQC_BASE_URL"})

    def validate(self) -> Self:
        """Validate RQC configuration fields."""
        if self.request_timeout <= 0:
            raise ConfigValidationError(
                "request_timeout", self.request_timeout,
                "Must be greater than 0.", section="rqc"
            )
        if self.retry_max_retries < 0:
            raise ConfigValidationError(
                "retry_max_retries", self.retry_max_retries,
                "Must be >= 0.", section="rqc"
            )
        if self.retry_initial_delay <= 0:
            raise ConfigValidationError(
                "retry_initial_delay", self.retry_initial_delay,
                "Must be greater than 0.", section="rqc"
            )
        if self.poll_interval <= 0:
            raise ConfigValidationError(
                "poll_interval", self.poll_interval,
                "Must be greater than 0.", section="rqc"
            )
        if self.poll_max_duration <= 0:
            raise ConfigValidationError(
                "poll_max_duration", self.poll_max_duration,
                "Must be greater than 0.", section="rqc"
            )
        if self.poll_overload_timeout <= 0:
            raise ConfigValidationError(
                "poll_overload_timeout", self.poll_overload_timeout,
                "Must be greater than 0.", section="rqc"
            )
        if self.max_workers <= 0:
            raise ConfigValidationError(
                "max_workers", self.max_workers,
                "Must be greater than 0.", section="rqc"
            )
        if self.base_url and not (self.base_url.startswith("http://") or self.base_url.startswith("https://")):
            raise ConfigValidationError(
                "base_url", self.base_url,
                "Must start with 'http://' or 'https://'.", section="rqc"
            )
        return self

Functions

validate

validate() -> Self

Validate RQC configuration fields.

Source code in src/stkai/_config.py

def validate(self) -> Self:
    """Validate RQC configuration fields."""
    if self.request_timeout <= 0:
        raise ConfigValidationError(
            "request_timeout", self.request_timeout,
            "Must be greater than 0.", section="rqc"
        )
    if self.retry_max_retries < 0:
        raise ConfigValidationError(
            "retry_max_retries", self.retry_max_retries,
            "Must be >= 0.", section="rqc"
        )
    if self.retry_initial_delay <= 0:
        raise ConfigValidationError(
            "retry_initial_delay", self.retry_initial_delay,
            "Must be greater than 0.", section="rqc"
        )
    if self.poll_interval <= 0:
        raise ConfigValidationError(
            "poll_interval", self.poll_interval,
            "Must be greater than 0.", section="rqc"
        )
    if self.poll_max_duration <= 0:
        raise ConfigValidationError(
            "poll_max_duration", self.poll_max_duration,
            "Must be greater than 0.", section="rqc"
        )
    if self.poll_overload_timeout <= 0:
        raise ConfigValidationError(
            "poll_overload_timeout", self.poll_overload_timeout,
            "Must be greater than 0.", section="rqc"
        )
    if self.max_workers <= 0:
        raise ConfigValidationError(
            "max_workers", self.max_workers,
            "Must be greater than 0.", section="rqc"
        )
    if self.base_url and not (self.base_url.startswith("http://") or self.base_url.startswith("https://")):
        raise ConfigValidationError(
            "base_url", self.base_url,
            "Must start with 'http://' or 'https://'.", section="rqc"
        )
    return self

AgentConfig dataclass

Bases: OverridableConfig

Configuration for Agent clients.

These settings are used as defaults when creating Agent instances without explicitly providing AgentOptions.

Attributes:

Name	Type	Description
request_timeout	int	HTTP request timeout in seconds for chat requests. Env var: STKAI_AGENT_REQUEST_TIMEOUT
base_url	str	Base URL for the Agent API. Env var: STKAI_AGENT_BASE_URL
retry_max_retries	int	Maximum number of retry attempts for failed chat calls. Use 0 to disable retries (single attempt only). Use 3 for 4 total attempts (1 original + 3 retries). Env var: STKAI_AGENT_RETRY_MAX_RETRIES
retry_initial_delay	float	Initial delay in seconds for the first retry attempt. Subsequent retries use exponential backoff (delay doubles each attempt). Example: with 0.5s initial delay, retries wait 0.5s, 1s, 2s, 4s... Env var: STKAI_AGENT_RETRY_INITIAL_DELAY
max_workers	int	Maximum number of concurrent threads for chat_many(). Env var: STKAI_AGENT_MAX_WORKERS

Example

from stkai import STKAI STKAI.config.agent.request_timeout 60 STKAI.config.agent.base_url 'https://genai-inference-app.stackspot.com' STKAI.config.agent.retry_max_retries 0

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class AgentConfig(OverridableConfig):
    """
    Configuration for Agent clients.

    These settings are used as defaults when creating Agent instances
    without explicitly providing AgentOptions.

    Attributes:
        request_timeout: HTTP request timeout in seconds for chat requests.
            Env var: STKAI_AGENT_REQUEST_TIMEOUT

        base_url: Base URL for the Agent API.
            Env var: STKAI_AGENT_BASE_URL

        retry_max_retries: Maximum number of retry attempts for failed chat calls.
            Use 0 to disable retries (single attempt only).
            Use 3 for 4 total attempts (1 original + 3 retries).
            Env var: STKAI_AGENT_RETRY_MAX_RETRIES

        retry_initial_delay: Initial delay in seconds for the first retry attempt.
            Subsequent retries use exponential backoff (delay doubles each attempt).
            Example: with 0.5s initial delay, retries wait 0.5s, 1s, 2s, 4s...
            Env var: STKAI_AGENT_RETRY_INITIAL_DELAY

        max_workers: Maximum number of concurrent threads for chat_many().
            Env var: STKAI_AGENT_MAX_WORKERS

    Example:
        >>> from stkai import STKAI
        >>> STKAI.config.agent.request_timeout
        60
        >>> STKAI.config.agent.base_url
        'https://genai-inference-app.stackspot.com'
        >>> STKAI.config.agent.retry_max_retries
        0
    """

    request_timeout: int = field(default=60, metadata={"env": "STKAI_AGENT_REQUEST_TIMEOUT"})
    base_url: str = field(default="https://genai-inference-app.stackspot.com", metadata={"env": "STKAI_AGENT_BASE_URL"})
    retry_max_retries: int = field(default=3, metadata={"env": "STKAI_AGENT_RETRY_MAX_RETRIES"})
    retry_initial_delay: float = field(default=0.5, metadata={"env": "STKAI_AGENT_RETRY_INITIAL_DELAY"})
    max_workers: int = field(default=8, metadata={"env": "STKAI_AGENT_MAX_WORKERS"})

    def validate(self) -> Self:
        """Validate Agent configuration fields."""
        if self.request_timeout <= 0:
            raise ConfigValidationError(
                "request_timeout", self.request_timeout,
                "Must be greater than 0.", section="agent"
            )
        if self.base_url and not (self.base_url.startswith("http://") or self.base_url.startswith("https://")):
            raise ConfigValidationError(
                "base_url", self.base_url,
                "Must start with 'http://' or 'https://'.", section="agent"
            )
        if self.retry_max_retries < 0:
            raise ConfigValidationError(
                "retry_max_retries", self.retry_max_retries,
                "Must be >= 0.", section="agent"
            )
        if self.retry_initial_delay <= 0:
            raise ConfigValidationError(
                "retry_initial_delay", self.retry_initial_delay,
                "Must be greater than 0.", section="agent"
            )
        if self.max_workers <= 0:
            raise ConfigValidationError(
                "max_workers", self.max_workers,
                "Must be greater than 0.", section="agent"
            )
        return self

Functions

validate

validate() -> Self

Validate Agent configuration fields.

Source code in src/stkai/_config.py

def validate(self) -> Self:
    """Validate Agent configuration fields."""
    if self.request_timeout <= 0:
        raise ConfigValidationError(
            "request_timeout", self.request_timeout,
            "Must be greater than 0.", section="agent"
        )
    if self.base_url and not (self.base_url.startswith("http://") or self.base_url.startswith("https://")):
        raise ConfigValidationError(
            "base_url", self.base_url,
            "Must start with 'http://' or 'https://'.", section="agent"
        )
    if self.retry_max_retries < 0:
        raise ConfigValidationError(
            "retry_max_retries", self.retry_max_retries,
            "Must be >= 0.", section="agent"
        )
    if self.retry_initial_delay <= 0:
        raise ConfigValidationError(
            "retry_initial_delay", self.retry_initial_delay,
            "Must be greater than 0.", section="agent"
        )
    if self.max_workers <= 0:
        raise ConfigValidationError(
            "max_workers", self.max_workers,
            "Must be greater than 0.", section="agent"
        )
    return self

RateLimitConfig dataclass

Bases: OverridableConfig

Configuration for HTTP client rate limiting.

When enabled, the EnvironmentAwareHttpClient automatically wraps the underlying HTTP client with rate limiting based on the selected strategy.

Available strategies

• "token_bucket": Simple Token Bucket algorithm. Limits requests to max_requests per time_window. Blocks if limit exceeded.
• "adaptive": Adaptive rate limiting with AIMD (Additive Increase, Multiplicative Decrease). Automatically adjusts rate based on HTTP 429 responses from the server.

Note: HTTP 429 retry logic is handled by the Retrying class, not the rate limiter. The adaptive strategy only applies AIMD penalty on 429 responses.

Attributes:

Name	Type	Description
enabled	bool	Whether to enable rate limiting. Env var: STKAI_RATE_LIMIT_ENABLED
strategy	RateLimitStrategy	Rate limiting strategy to use. Env var: STKAI_RATE_LIMIT_STRATEGY
max_requests	int	Maximum requests allowed in the time window. Env var: STKAI_RATE_LIMIT_MAX_REQUESTS
time_window	float	Time window in seconds for the rate limit. Env var: STKAI_RATE_LIMIT_TIME_WINDOW
max_wait_time	float | None	Maximum seconds to wait for a token before raising TokenAcquisitionTimeoutError. None means wait indefinitely. Env var: STKAI_RATE_LIMIT_MAX_WAIT_TIME
min_rate_floor	float	(adaptive only) Minimum rate as fraction of max_requests. Prevents rate from dropping below this floor. Env var: STKAI_RATE_LIMIT_MIN_RATE_FLOOR
penalty_factor	float	(adaptive only) Rate reduction factor on 429 (0-1). Env var: STKAI_RATE_LIMIT_PENALTY_FACTOR
recovery_factor	float	(adaptive only) Rate increase factor on success (0-1). Env var: STKAI_RATE_LIMIT_RECOVERY_FACTOR

Example

from stkai import STKAI STKAI.configure( ... rate_limit={ ... "enabled": True, ... "strategy": "token_bucket", ... "max_requests": 10, ... "time_window": 60.0, ... } ... )

Or with adaptive strategy

STKAI.configure( ... rate_limit={ ... "enabled": True, ... "strategy": "adaptive", ... "max_requests": 100, ... "min_rate_floor": 0.1, ... } ... )

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class RateLimitConfig(OverridableConfig):
    """
    Configuration for HTTP client rate limiting.

    When enabled, the EnvironmentAwareHttpClient automatically wraps the
    underlying HTTP client with rate limiting based on the selected strategy.

    Available strategies:
        - "token_bucket": Simple Token Bucket algorithm. Limits requests to
            max_requests per time_window. Blocks if limit exceeded.
        - "adaptive": Adaptive rate limiting with AIMD (Additive Increase,
            Multiplicative Decrease). Automatically adjusts rate based on
            HTTP 429 responses from the server.

    Note: HTTP 429 retry logic is handled by the Retrying class, not the rate
    limiter. The adaptive strategy only applies AIMD penalty on 429 responses.

    Attributes:
        enabled: Whether to enable rate limiting.
            Env var: STKAI_RATE_LIMIT_ENABLED

        strategy: Rate limiting strategy to use.
            Env var: STKAI_RATE_LIMIT_STRATEGY

        max_requests: Maximum requests allowed in the time window.
            Env var: STKAI_RATE_LIMIT_MAX_REQUESTS

        time_window: Time window in seconds for the rate limit.
            Env var: STKAI_RATE_LIMIT_TIME_WINDOW

        max_wait_time: Maximum seconds to wait for a token before raising
            TokenAcquisitionTimeoutError. None means wait indefinitely.
            Env var: STKAI_RATE_LIMIT_MAX_WAIT_TIME

        min_rate_floor: (adaptive only) Minimum rate as fraction of max_requests.
            Prevents rate from dropping below this floor.
            Env var: STKAI_RATE_LIMIT_MIN_RATE_FLOOR

        penalty_factor: (adaptive only) Rate reduction factor on 429 (0-1).
            Env var: STKAI_RATE_LIMIT_PENALTY_FACTOR

        recovery_factor: (adaptive only) Rate increase factor on success (0-1).
            Env var: STKAI_RATE_LIMIT_RECOVERY_FACTOR

    Example:
        >>> from stkai import STKAI
        >>> STKAI.configure(
        ...     rate_limit={
        ...         "enabled": True,
        ...         "strategy": "token_bucket",
        ...         "max_requests": 10,
        ...         "time_window": 60.0,
        ...     }
        ... )

        >>> # Or with adaptive strategy
        >>> STKAI.configure(
        ...     rate_limit={
        ...         "enabled": True,
        ...         "strategy": "adaptive",
        ...         "max_requests": 100,
        ...         "min_rate_floor": 0.1,
        ...     }
        ... )
    """

    # Normal fields (processed automatically by with_env_vars)
    enabled: bool = field(default=False, metadata={"env": "STKAI_RATE_LIMIT_ENABLED"})
    strategy: RateLimitStrategy = field(default="token_bucket", metadata={"env": "STKAI_RATE_LIMIT_STRATEGY"})
    # Common parameters
    max_requests: int = field(default=100, metadata={"env": "STKAI_RATE_LIMIT_MAX_REQUESTS"})
    time_window: float = field(default=60.0, metadata={"env": "STKAI_RATE_LIMIT_TIME_WINDOW"})
    # Special field (processed manually - can be None for "unlimited")
    max_wait_time: float | None = field(
        default=45.0,
        metadata={"env": "STKAI_RATE_LIMIT_MAX_WAIT_TIME", "skip": True},
    )
    # Adaptive strategy parameters (ignored if strategy != "adaptive")
    min_rate_floor: float = field(default=0.1, metadata={"env": "STKAI_RATE_LIMIT_MIN_RATE_FLOOR"})
    penalty_factor: float = field(default=0.3, metadata={"env": "STKAI_RATE_LIMIT_PENALTY_FACTOR"})
    recovery_factor: float = field(default=0.05, metadata={"env": "STKAI_RATE_LIMIT_RECOVERY_FACTOR"})

    def with_overrides(
        self,
        overrides: dict[str, Any],
        allow_none_fields: set[str] | None = None,
    ) -> Self:
        """
        Return a new instance with specified fields overridden.

        Extends base implementation to support special values:
        - max_wait_time: Can be None, a float, or "unlimited"/"none"/"null"
          (strings are converted to None for unlimited waiting).

        Args:
            overrides: Dict of field names to new values.
            allow_none_fields: Additional fields that accept None (merged with max_wait_time).

        Returns:
            New instance with updated values.
        """
        if not overrides:
            return self

        # Convert "unlimited"/"none"/"null" strings to None for max_wait_time
        processed = dict(overrides)
        if "max_wait_time" in processed:
            value = processed["max_wait_time"]
            if isinstance(value, str) and value.lower() in ("none", "null", "unlimited"):
                processed["max_wait_time"] = None

        # Always allow None for max_wait_time, plus any additional fields
        merged_allow_none = {"max_wait_time"} | (allow_none_fields or set())
        return super().with_overrides(processed, allow_none_fields=merged_allow_none)

    def with_env_vars(self) -> Self:
        """Override to handle max_wait_time (can be None for 'unlimited')."""
        # Process normal fields via base class
        result = super().with_env_vars()

        # Process max_wait_time manually
        overrides: dict[str, Any] = {}
        if max_wait := os.environ.get("STKAI_RATE_LIMIT_MAX_WAIT_TIME"):
            if max_wait.lower() in ("none", "null", "unlimited"):
                overrides["max_wait_time"] = None
            else:
                overrides["max_wait_time"] = float(max_wait)

        return result.with_overrides(overrides, allow_none_fields={"max_wait_time"})

    def validate(self) -> Self:
        """Validate rate limit configuration fields."""
        valid_strategies = ("token_bucket", "adaptive")
        if self.strategy not in valid_strategies:
            raise ConfigValidationError(
                "strategy", self.strategy,
                f"Must be one of: {valid_strategies}.", section="rate_limit"
            )
        if self.max_requests <= 0:
            raise ConfigValidationError(
                "max_requests", self.max_requests,
                "Must be greater than 0.", section="rate_limit"
            )
        if self.time_window <= 0:
            raise ConfigValidationError(
                "time_window", self.time_window,
                "Must be greater than 0.", section="rate_limit"
            )
        if self.max_wait_time is not None and self.max_wait_time <= 0:
            raise ConfigValidationError(
                "max_wait_time", self.max_wait_time,
                "Must be greater than 0 (or None for unlimited).", section="rate_limit"
            )
        if self.min_rate_floor <= 0 or self.min_rate_floor > 1:
            raise ConfigValidationError(
                "min_rate_floor", self.min_rate_floor,
                "Must be greater than 0 and <= 1.", section="rate_limit"
            )
        if self.penalty_factor <= 0 or self.penalty_factor >= 1:
            raise ConfigValidationError(
                "penalty_factor", self.penalty_factor,
                "Must be greater than 0 and less than 1.", section="rate_limit"
            )
        if self.recovery_factor <= 0 or self.recovery_factor >= 1:
            raise ConfigValidationError(
                "recovery_factor", self.recovery_factor,
                "Must be greater than 0 and less than 1.", section="rate_limit"
            )
        return self

    # -------------------------------------------------------------------------
    # Presets
    # -------------------------------------------------------------------------

    @classmethod
    def conservative_preset(
        cls,
        max_requests: int = 20,
        time_window: float = 60.0,
    ) -> RateLimitConfig:
        """
        Conservative rate limiting preset.

        Prioritizes stability over throughput. Best for:
        - Critical batch jobs
        - CI/CD pipelines
        - Scenarios with many concurrent processes

        Behavior:
        - Waits up to 120s for tokens (patient, but not forever)
        - Aggressive penalty on 429 (halves rate)
        - Slow recovery (2% per success)
        - Can drop to 5% of max_requests under stress

        Args:
            max_requests: Maximum requests allowed in the time window.
                Calculate based on your quota and expected concurrent processes.
                Default assumes ~5 processes sharing a 100 req/min quota.
            time_window: Time window in seconds for the rate limit.

        Returns:
            RateLimitConfig with conservative settings.

        Example:
            >>> # Quota of 100 req/min, expect ~5 processes
            >>> config = RateLimitConfig.conservative_preset(max_requests=20)

            >>> # Quota of 200 req/min, expect ~4 processes
            >>> config = RateLimitConfig.conservative_preset(max_requests=50)
        """
        return cls(
            enabled=True,
            strategy="adaptive",
            max_requests=max_requests,
            time_window=time_window,
            max_wait_time=120.0,
            min_rate_floor=0.05,
            penalty_factor=0.5,
            recovery_factor=0.02,
        )

    @classmethod
    def balanced_preset(
        cls,
        max_requests: int = 40,
        time_window: float = 60.0,
    ) -> RateLimitConfig:
        """
        Balanced rate limiting preset (recommended).

        Sensible defaults for most use cases. Best for:
        - General batch processing
        - 2-5 concurrent processes
        - When unsure which preset to use

        Behavior:
        - Waits up to 45s for tokens
        - Moderate penalty on 429 (30% reduction)
        - Medium recovery (5% per success)
        - Can drop to 10% of max_requests under stress

        Args:
            max_requests: Maximum requests allowed in the time window.
                Calculate based on your quota and expected concurrent processes.
                Default assumes ~2-5 processes sharing a 100 req/min quota.
            time_window: Time window in seconds for the rate limit.

        Returns:
            RateLimitConfig with balanced settings.

        Example:
            >>> # Quota of 100 req/min, expect ~2 processes
            >>> config = RateLimitConfig.balanced_preset(max_requests=50)

            >>> # Use defaults (good for typical scenarios)
            >>> config = RateLimitConfig.balanced_preset()
        """
        return cls(
            enabled=True,
            strategy="adaptive",
            max_requests=max_requests,
            time_window=time_window,
            max_wait_time=45.0,
            min_rate_floor=0.1,
            penalty_factor=0.3,
            recovery_factor=0.05,
        )

    @classmethod
    def optimistic_preset(
        cls,
        max_requests: int = 80,
        time_window: float = 60.0,
    ) -> RateLimitConfig:
        """
        Optimistic rate limiting preset.

        Prioritizes throughput over stability. Best for:
        - Interactive/CLI usage
        - Single-process scenarios
        - When external retry logic exists

        Behavior:
        - Waits up to 20s for tokens
        - Light penalty on 429 (15% reduction)
        - Fast recovery (10% per success)
        - Never drops below 30% of max_requests

        Args:
            max_requests: Maximum requests allowed in the time window.
                Calculate based on your quota. Default assumes single process
                using ~80% of a 100 req/min quota.
            time_window: Time window in seconds for the rate limit.

        Returns:
            RateLimitConfig with optimistic settings.

        Example:
            >>> # Quota of 100 req/min, single process
            >>> config = RateLimitConfig.optimistic_preset(max_requests=80)

            >>> # Quota of 200 req/min, want maximum throughput
            >>> config = RateLimitConfig.optimistic_preset(max_requests=180)
        """
        return cls(
            enabled=True,
            strategy="adaptive",
            max_requests=max_requests,
            time_window=time_window,
            max_wait_time=20.0,
            min_rate_floor=0.3,
            penalty_factor=0.15,
            recovery_factor=0.1,
        )

Functions

with_overrides

with_overrides(overrides: dict[str, Any], allow_none_fields: set[str] | None = None) -> Self

Return a new instance with specified fields overridden.

Extends base implementation to support special values: - max_wait_time: Can be None, a float, or "unlimited"/"none"/"null" (strings are converted to None for unlimited waiting).

Parameters:

Name	Type	Description	Default
overrides	dict[str, Any]	Dict of field names to new values.	required
allow_none_fields	set[str] | None	Additional fields that accept None (merged with max_wait_time).	None

Returns:

Type	Description
Self	New instance with updated values.

Source code in src/stkai/_config.py

def with_overrides(
    self,
    overrides: dict[str, Any],
    allow_none_fields: set[str] | None = None,
) -> Self:
    """
    Return a new instance with specified fields overridden.

    Extends base implementation to support special values:
    - max_wait_time: Can be None, a float, or "unlimited"/"none"/"null"
      (strings are converted to None for unlimited waiting).

    Args:
        overrides: Dict of field names to new values.
        allow_none_fields: Additional fields that accept None (merged with max_wait_time).

    Returns:
        New instance with updated values.
    """
    if not overrides:
        return self

    # Convert "unlimited"/"none"/"null" strings to None for max_wait_time
    processed = dict(overrides)
    if "max_wait_time" in processed:
        value = processed["max_wait_time"]
        if isinstance(value, str) and value.lower() in ("none", "null", "unlimited"):
            processed["max_wait_time"] = None

    # Always allow None for max_wait_time, plus any additional fields
    merged_allow_none = {"max_wait_time"} | (allow_none_fields or set())
    return super().with_overrides(processed, allow_none_fields=merged_allow_none)

with_env_vars

with_env_vars() -> Self

Override to handle max_wait_time (can be None for 'unlimited').

Source code in src/stkai/_config.py

def with_env_vars(self) -> Self:
    """Override to handle max_wait_time (can be None for 'unlimited')."""
    # Process normal fields via base class
    result = super().with_env_vars()

    # Process max_wait_time manually
    overrides: dict[str, Any] = {}
    if max_wait := os.environ.get("STKAI_RATE_LIMIT_MAX_WAIT_TIME"):
        if max_wait.lower() in ("none", "null", "unlimited"):
            overrides["max_wait_time"] = None
        else:
            overrides["max_wait_time"] = float(max_wait)

    return result.with_overrides(overrides, allow_none_fields={"max_wait_time"})

validate

validate() -> Self

Validate rate limit configuration fields.

Source code in src/stkai/_config.py

def validate(self) -> Self:
    """Validate rate limit configuration fields."""
    valid_strategies = ("token_bucket", "adaptive")
    if self.strategy not in valid_strategies:
        raise ConfigValidationError(
            "strategy", self.strategy,
            f"Must be one of: {valid_strategies}.", section="rate_limit"
        )
    if self.max_requests <= 0:
        raise ConfigValidationError(
            "max_requests", self.max_requests,
            "Must be greater than 0.", section="rate_limit"
        )
    if self.time_window <= 0:
        raise ConfigValidationError(
            "time_window", self.time_window,
            "Must be greater than 0.", section="rate_limit"
        )
    if self.max_wait_time is not None and self.max_wait_time <= 0:
        raise ConfigValidationError(
            "max_wait_time", self.max_wait_time,
            "Must be greater than 0 (or None for unlimited).", section="rate_limit"
        )
    if self.min_rate_floor <= 0 or self.min_rate_floor > 1:
        raise ConfigValidationError(
            "min_rate_floor", self.min_rate_floor,
            "Must be greater than 0 and <= 1.", section="rate_limit"
        )
    if self.penalty_factor <= 0 or self.penalty_factor >= 1:
        raise ConfigValidationError(
            "penalty_factor", self.penalty_factor,
            "Must be greater than 0 and less than 1.", section="rate_limit"
        )
    if self.recovery_factor <= 0 or self.recovery_factor >= 1:
        raise ConfigValidationError(
            "recovery_factor", self.recovery_factor,
            "Must be greater than 0 and less than 1.", section="rate_limit"
        )
    return self

conservative_preset classmethod

conservative_preset(max_requests: int = 20, time_window: float = 60.0) -> RateLimitConfig

Conservative rate limiting preset.

Prioritizes stability over throughput. Best for: - Critical batch jobs - CI/CD pipelines - Scenarios with many concurrent processes

Behavior: - Waits up to 120s for tokens (patient, but not forever) - Aggressive penalty on 429 (halves rate) - Slow recovery (2% per success) - Can drop to 5% of max_requests under stress

Parameters:

Name	Type	Description	Default
max_requests	int	Maximum requests allowed in the time window. Calculate based on your quota and expected concurrent processes. Default assumes ~5 processes sharing a 100 req/min quota.	20
time_window	float	Time window in seconds for the rate limit.	60.0

Returns:

Type	Description
RateLimitConfig	RateLimitConfig with conservative settings.

Example

Quota of 100 req/min, expect ~5 processes

config = RateLimitConfig.conservative_preset(max_requests=20)

Quota of 200 req/min, expect ~4 processes

config = RateLimitConfig.conservative_preset(max_requests=50)

Source code in src/stkai/_config.py

@classmethod
def conservative_preset(
    cls,
    max_requests: int = 20,
    time_window: float = 60.0,
) -> RateLimitConfig:
    """
    Conservative rate limiting preset.

    Prioritizes stability over throughput. Best for:
    - Critical batch jobs
    - CI/CD pipelines
    - Scenarios with many concurrent processes

    Behavior:
    - Waits up to 120s for tokens (patient, but not forever)
    - Aggressive penalty on 429 (halves rate)
    - Slow recovery (2% per success)
    - Can drop to 5% of max_requests under stress

    Args:
        max_requests: Maximum requests allowed in the time window.
            Calculate based on your quota and expected concurrent processes.
            Default assumes ~5 processes sharing a 100 req/min quota.
        time_window: Time window in seconds for the rate limit.

    Returns:
        RateLimitConfig with conservative settings.

    Example:
        >>> # Quota of 100 req/min, expect ~5 processes
        >>> config = RateLimitConfig.conservative_preset(max_requests=20)

        >>> # Quota of 200 req/min, expect ~4 processes
        >>> config = RateLimitConfig.conservative_preset(max_requests=50)
    """
    return cls(
        enabled=True,
        strategy="adaptive",
        max_requests=max_requests,
        time_window=time_window,
        max_wait_time=120.0,
        min_rate_floor=0.05,
        penalty_factor=0.5,
        recovery_factor=0.02,
    )

balanced_preset classmethod

balanced_preset(max_requests: int = 40, time_window: float = 60.0) -> RateLimitConfig

Balanced rate limiting preset (recommended).

Sensible defaults for most use cases. Best for: - General batch processing - 2-5 concurrent processes - When unsure which preset to use

Behavior: - Waits up to 45s for tokens - Moderate penalty on 429 (30% reduction) - Medium recovery (5% per success) - Can drop to 10% of max_requests under stress

Parameters:

Name	Type	Description	Default
max_requests	int	Maximum requests allowed in the time window. Calculate based on your quota and expected concurrent processes. Default assumes ~2-5 processes sharing a 100 req/min quota.	40
time_window	float	Time window in seconds for the rate limit.	60.0

Returns:

Type	Description
RateLimitConfig	RateLimitConfig with balanced settings.

Example

Quota of 100 req/min, expect ~2 processes

config = RateLimitConfig.balanced_preset(max_requests=50)

Use defaults (good for typical scenarios)

config = RateLimitConfig.balanced_preset()

Source code in src/stkai/_config.py

@classmethod
def balanced_preset(
    cls,
    max_requests: int = 40,
    time_window: float = 60.0,
) -> RateLimitConfig:
    """
    Balanced rate limiting preset (recommended).

    Sensible defaults for most use cases. Best for:
    - General batch processing
    - 2-5 concurrent processes
    - When unsure which preset to use

    Behavior:
    - Waits up to 45s for tokens
    - Moderate penalty on 429 (30% reduction)
    - Medium recovery (5% per success)
    - Can drop to 10% of max_requests under stress

    Args:
        max_requests: Maximum requests allowed in the time window.
            Calculate based on your quota and expected concurrent processes.
            Default assumes ~2-5 processes sharing a 100 req/min quota.
        time_window: Time window in seconds for the rate limit.

    Returns:
        RateLimitConfig with balanced settings.

    Example:
        >>> # Quota of 100 req/min, expect ~2 processes
        >>> config = RateLimitConfig.balanced_preset(max_requests=50)

        >>> # Use defaults (good for typical scenarios)
        >>> config = RateLimitConfig.balanced_preset()
    """
    return cls(
        enabled=True,
        strategy="adaptive",
        max_requests=max_requests,
        time_window=time_window,
        max_wait_time=45.0,
        min_rate_floor=0.1,
        penalty_factor=0.3,
        recovery_factor=0.05,
    )

optimistic_preset classmethod

optimistic_preset(max_requests: int = 80, time_window: float = 60.0) -> RateLimitConfig

Optimistic rate limiting preset.

Prioritizes throughput over stability. Best for: - Interactive/CLI usage - Single-process scenarios - When external retry logic exists

Behavior: - Waits up to 20s for tokens - Light penalty on 429 (15% reduction) - Fast recovery (10% per success) - Never drops below 30% of max_requests

Parameters:

Name	Type	Description	Default
max_requests	int	Maximum requests allowed in the time window. Calculate based on your quota. Default assumes single process using ~80% of a 100 req/min quota.	80
time_window	float	Time window in seconds for the rate limit.	60.0

Returns:

Type	Description
RateLimitConfig	RateLimitConfig with optimistic settings.

Example

Quota of 100 req/min, single process

config = RateLimitConfig.optimistic_preset(max_requests=80)

Quota of 200 req/min, want maximum throughput

config = RateLimitConfig.optimistic_preset(max_requests=180)

Source code in src/stkai/_config.py

@classmethod
def optimistic_preset(
    cls,
    max_requests: int = 80,
    time_window: float = 60.0,
) -> RateLimitConfig:
    """
    Optimistic rate limiting preset.

    Prioritizes throughput over stability. Best for:
    - Interactive/CLI usage
    - Single-process scenarios
    - When external retry logic exists

    Behavior:
    - Waits up to 20s for tokens
    - Light penalty on 429 (15% reduction)
    - Fast recovery (10% per success)
    - Never drops below 30% of max_requests

    Args:
        max_requests: Maximum requests allowed in the time window.
            Calculate based on your quota. Default assumes single process
            using ~80% of a 100 req/min quota.
        time_window: Time window in seconds for the rate limit.

    Returns:
        RateLimitConfig with optimistic settings.

    Example:
        >>> # Quota of 100 req/min, single process
        >>> config = RateLimitConfig.optimistic_preset(max_requests=80)

        >>> # Quota of 200 req/min, want maximum throughput
        >>> config = RateLimitConfig.optimistic_preset(max_requests=180)
    """
    return cls(
        enabled=True,
        strategy="adaptive",
        max_requests=max_requests,
        time_window=time_window,
        max_wait_time=20.0,
        min_rate_floor=0.3,
        penalty_factor=0.15,
        recovery_factor=0.1,
    )

Helper Classes

ConfigEntry dataclass

A configuration field with its resolved value and source.

Represents a single configuration entry with metadata about where the value came from. Used by explain_data() for structured output.

Attributes:

Name	Type	Description
name	str	The field name (e.g., "request_timeout").
value	Any	The resolved value.
source	str	Where the value came from: - "default": Hardcoded default value - "env:VAR_NAME": Environment variable - "CLI": StackSpot CLI (oscli) - "configure": Set via STKAI.configure()

Example

entry = ConfigEntry("request_timeout", 60, "configure") entry.name 'request_timeout' entry.value 60 entry.source 'configure' entry.formatted_value '60'

Source code in src/stkai/_config.py

@dataclass(frozen=True)
class ConfigEntry:
    """
    A configuration field with its resolved value and source.

    Represents a single configuration entry with metadata about where
    the value came from. Used by explain_data() for structured output.

    Attributes:
        name: The field name (e.g., "request_timeout").
        value: The resolved value.
        source: Where the value came from:
            - "default": Hardcoded default value
            - "env:VAR_NAME": Environment variable
            - "CLI": StackSpot CLI (oscli)
            - "configure": Set via STKAI.configure()

    Example:
        >>> entry = ConfigEntry("request_timeout", 60, "configure")
        >>> entry.name
        'request_timeout'
        >>> entry.value
        60
        >>> entry.source
        'configure'
        >>> entry.formatted_value
        '60'
    """

    name: str
    value: Any
    source: str

    @property
    def formatted_value(self) -> str:
        """
        Return value formatted for display.

        Masks sensitive fields (e.g., client_secret) showing only
        first and last 4 characters, and truncates long strings.

        Returns:
            Formatted string representation of the value.

        Examples:
            >>> ConfigEntry("client_secret", "super-secret-key", "configure").formatted_value
            'supe********-key'
            >>> ConfigEntry("client_secret", "short", "configure").formatted_value
            '********t'
        """
        # Mask sensitive fields
        if self.name in ("client_secret",) and self.value is not None:
            secret = str(self.value)
            # Long secrets: show first 4 and last 4 chars
            if len(secret) >= 12:
                return f"{secret[:4]}********{secret[-4:]}"
            # Short secrets: show last 1/3 of chars
            if len(secret) >= 3:
                visible = max(1, len(secret) // 3)
                return f"********{secret[-visible:]}"
            return "********"

        # Handle None
        if self.value is None:
            return "None"

        # Convert to string and truncate if needed
        str_value = str(self.value)
        max_length = 50
        if len(str_value) > max_length:
            return str_value[: max_length - 3] + "..."

        return str_value

Attributes

formatted_value property

formatted_value: str

Return value formatted for display.

Masks sensitive fields (e.g., client_secret) showing only first and last 4 characters, and truncates long strings.

Returns:

Type	Description
str	Formatted string representation of the value.

Examples:

>>> ConfigEntry("client_secret", "super-secret-key", "configure").formatted_value
'supe********-key'
>>> ConfigEntry("client_secret", "short", "configure").formatted_value
'********t'

Error Classes

ConfigEnvVarError

Bases: ValueError

Raised when an environment variable has an invalid value.

Source code in src/stkai/_config.py

class ConfigEnvVarError(ValueError):
    """Raised when an environment variable has an invalid value."""

    def __init__(
        self,
        env_var: str,
        value: str,
        expected_type: str,
        cause: Exception | None = None,
    ):
        self.env_var = env_var
        self.value = value
        self.expected_type = expected_type
        super().__init__(f"Invalid value for {env_var}: '{value}' (expected {expected_type})")
        self.__cause__ = cause

ConfigValidationError

Bases: ValueError

Raised when a configuration value fails validation.

Source code in src/stkai/_config.py

class ConfigValidationError(ValueError):
    """Raised when a configuration value fails validation."""

    def __init__(
        self,
        field: str,
        value: Any,
        message: str,
        section: str | None = None,
    ):
        self.field = field
        self.value = value
        self.section = section
        prefix = f"[{section}] " if section else ""
        super().__init__(f"{prefix}Invalid value for '{field}': {value!r}. {message}")