API Reference

CygnisAI Python SDK ~~~~~~~~~~~~~~~~~~~

Simple, fast access to CygnisAI language models.

Quick-start::

import cygnisai_sdk_python as cygnis

cygnis.configure(api_key="YOUR_KEY")
model = cygnis.GenerativeModel("alpha2")
response = model.generate_content("Give me a Python tip.")
print(response.text)

Streaming::

stream = model.generate_content("Tell me a story.", stream=True)
async for token in stream:
    print(token, end="", flush=True)

AuthenticationError

Bases: CygnisAIError

Raised on HTTP 401 / 403.

Source code in cygnisai_sdk_python/client.py

class AuthenticationError(CygnisAIError):
    """Raised on HTTP 401 / 403."""

ChatRequest

Bases: BaseModel

Payload sent to POST /v3/chat.

Source code in cygnisai_sdk_python/models.py

class ChatRequest(BaseModel):
    """Payload sent to POST /v3/chat."""

    model: str = Field(..., min_length=1, description="Model name, e.g. 'alpha2'.")
    prompt: str = Field(..., min_length=1, description="User prompt / question.")
    messages: Optional[List[Message]] = Field(
        None, description="Conversation history (optional)."
    )
    stream: bool = Field(False, description="Enable SSE streaming.")

    @field_validator("messages", mode="before")
    @classmethod
    def _messages_not_empty_list(cls, v: Any) -> Any:
        if isinstance(v, list) and len(v) == 0:
            return None
        return v

ChatResponse

Bases: BaseModel

Successful non-streaming response from POST /v3/chat.

Source code in cygnisai_sdk_python/models.py

class ChatResponse(BaseModel):
    """Successful non-streaming response from POST /v3/chat."""

    id: uuid.UUID = Field(..., description="Unique response identifier.")
    response: str = Field(..., description="Full model reply.")
    latency_ms: int = Field(..., ge=0, description="Round-trip latency in milliseconds.")
    redacted: bool = Field(..., description="Whether the reply was moderated.")
    usage: UsageInfo = Field(..., description="Token-usage statistics.")

    @field_validator("usage", mode="before")
    @classmethod
    def _coerce_usage(cls, v: Any) -> Any:
        if isinstance(v, dict):
            return UsageInfo(**v)
        return v

CygnisAIClient

Async HTTP client for the CygnisAI API.

Can be used as an async context manager::

async with CygnisAIClient(api_key="...") as client:
    response = await client.chat(request)

Or managed manually::

client = CygnisAIClient(api_key="...")
try:
    ...
finally:
    await client.close()

Source code in cygnisai_sdk_python/client.py

class CygnisAIClient:
    """
    Async HTTP client for the CygnisAI API.

    Can be used as an async context manager::

        async with CygnisAIClient(api_key="...") as client:
            response = await client.chat(request)

    Or managed manually::

        client = CygnisAIClient(api_key="...")
        try:
            ...
        finally:
            await client.close()
    """

    def __init__(
        self,
        api_key: str,
        base_url: str = DEFAULT_BASE_URL,
        timeout: float = DEFAULT_TIMEOUT,
        max_retries: int = DEFAULT_MAX_RETRIES,
        **httpx_kwargs: Any,
    ) -> None:
        if not api_key:
            raise ValueError("api_key must not be empty.")

        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.max_retries = max(0, max_retries)

        default_headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "cygnisai-python-sdk/1.0.0",
        }
        headers = {**default_headers, **httpx_kwargs.pop("headers", {})}

        self._client = httpx.AsyncClient(
            base_url=self.base_url,
            timeout=timeout,
            headers=headers,
            **httpx_kwargs,
        )

    # ------------------------------------------------------------------
    # Context manager support
    # ------------------------------------------------------------------

    async def __aenter__(self) -> "CygnisAIClient":
        return self

    async def __aexit__(self, *_: Any) -> None:
        await self.close()

    async def close(self) -> None:
        """Release the underlying HTTP connection pool."""
        if not self._client.is_closed:
            await self._client.aclose()

    # ------------------------------------------------------------------
    # Internal: retry logic
    # ------------------------------------------------------------------

    async def _post_with_retry(
        self,
        path: str,
        payload: dict[str, Any],
    ) -> httpx.Response:
        """POST *path* with exponential-backoff retries on transient errors."""
        import asyncio

        last_exc: Exception | None = None
        for attempt in range(self.max_retries + 1):
            try:
                resp = await self._client.post(path, json=payload)
                if resp.status_code not in _RETRY_STATUS_CODES or attempt == self.max_retries:
                    return resp
                wait = 2 ** attempt
                logger.warning(
                    "CygnisAI: HTTP %s on attempt %d/%d — retrying in %ds",
                    resp.status_code, attempt + 1, self.max_retries + 1, wait,
                )
                await asyncio.sleep(wait)
            except httpx.RequestError as exc:
                last_exc = exc
                if attempt == self.max_retries:
                    break
                wait = 2 ** attempt
                logger.warning(
                    "CygnisAI: Network error on attempt %d/%d — retrying in %ds: %s",
                    attempt + 1, self.max_retries + 1, wait, exc,
                )
                await asyncio.sleep(wait)

        raise NetworkError(
            f"Request failed after {self.max_retries + 1} attempts: {last_exc}",
            error_details=str(last_exc),
        )

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def chat(self, request: ChatRequest) -> ChatResponse:
        """
        Send a chat request and return the complete response.

        Args:
            request: A :class:`~cygnisai_sdk_python.models.ChatRequest` instance.

        Returns:
            :class:`~cygnisai_sdk_python.models.ChatResponse`

        Raises:
            AuthenticationError: On 401/403.
            RateLimitError: On 429.
            ServerError: On 5xx.
            NetworkError: On connection failure.
            ResponseValidationError: If the response body doesn't match the schema.
            CygnisAIError: For any other API error.
        """
        request.stream = False
        payload = request.model_dump(exclude_none=True)

        try:
            raw = await self._post_with_retry("/v3/chat", payload)
        except NetworkError:
            raise

        if raw.status_code != 200:
            try:
                body = raw.json()
                msg = _extract_error_message(body)
            except Exception:
                body = {}
                msg = raw.text
            raise _map_status_error(
                raw.status_code,
                f"API error (HTTP {raw.status_code}): {msg}",
                body,
            )

        try:
            return ChatResponse(**raw.json())
        except (ValidationError, Exception) as exc:
            raise ResponseValidationError(
                f"Failed to parse API response: {exc}",
                error_details=str(exc),
            ) from exc

    async def chat_stream(
        self, request: ChatRequest
    ) -> AsyncGenerator[str, None]:
        """
        Send a chat request and yield tokens as they arrive (SSE).

        Args:
            request: A :class:`~cygnisai_sdk_python.models.ChatRequest` instance.

        Yields:
            str – individual text tokens from the model.

        Raises:
            CygnisAIError (and subclasses): as documented on :meth:`chat`.
        """
        request.stream = True
        payload = request.model_dump(exclude_none=True)

        async with self._client.stream("POST", "/v3/chat", json=payload) as response:
            if response.status_code != 200:
                error_bytes = await response.aread()
                try:
                    body = json.loads(error_bytes.decode())
                    msg = _extract_error_message(body)
                except Exception:
                    body = {}
                    msg = error_bytes.decode()
                raise _map_status_error(
                    response.status_code,
                    f"API error (HTTP {response.status_code}): {msg}",
                    body,
                )

            async for raw_line in response.aiter_lines():
                line = raw_line.strip()
                if not line:
                    continue

                if line.startswith("data:"):
                    content = line[5:].strip()

                    if content == "[DONE]":
                        return

                    try:
                        data = json.loads(content)
                    except json.JSONDecodeError:
                        # Yield raw text if it isn't valid JSON
                        yield content
                        continue

                    if not isinstance(data, dict):
                        yield str(data)
                        continue

                    # Surface server-side errors embedded in the stream
                    if "error" in data or "detail" in data:
                        err_key = "error" if "error" in data else "detail"
                        err_val = data[err_key]
                        if isinstance(err_val, list):
                            msgs = [
                                f"{e.get('loc', ['?'])[-1]}: {e.get('msg', '')}"
                                for e in err_val
                                if isinstance(e, dict)
                            ]
                            err_val = "; ".join(msgs)
                        raise CygnisAIError(
                            f"Server error in stream: {err_val}",
                            error_details=data,
                        )

                    token = data.get("response") or data.get("text") or ""
                    if token:
                        yield token

chat(request) async

Send a chat request and return the complete response.

Parameters:

Name	Type	Description	Default
request	ChatRequest	A :class:~cygnisai_sdk_python.models.ChatRequest instance.	required

Returns:

Type	Description
ChatResponse	class:~cygnisai_sdk_python.models.ChatResponse

Raises:

Type	Description
AuthenticationError	On 401/403.
RateLimitError	On 429.
ServerError	On 5xx.
NetworkError	On connection failure.
ResponseValidationError	If the response body doesn't match the schema.
CygnisAIError	For any other API error.

Source code in cygnisai_sdk_python/client.py

async def chat(self, request: ChatRequest) -> ChatResponse:
    """
    Send a chat request and return the complete response.

    Args:
        request: A :class:`~cygnisai_sdk_python.models.ChatRequest` instance.

    Returns:
        :class:`~cygnisai_sdk_python.models.ChatResponse`

    Raises:
        AuthenticationError: On 401/403.
        RateLimitError: On 429.
        ServerError: On 5xx.
        NetworkError: On connection failure.
        ResponseValidationError: If the response body doesn't match the schema.
        CygnisAIError: For any other API error.
    """
    request.stream = False
    payload = request.model_dump(exclude_none=True)

    try:
        raw = await self._post_with_retry("/v3/chat", payload)
    except NetworkError:
        raise

    if raw.status_code != 200:
        try:
            body = raw.json()
            msg = _extract_error_message(body)
        except Exception:
            body = {}
            msg = raw.text
        raise _map_status_error(
            raw.status_code,
            f"API error (HTTP {raw.status_code}): {msg}",
            body,
        )

    try:
        return ChatResponse(**raw.json())
    except (ValidationError, Exception) as exc:
        raise ResponseValidationError(
            f"Failed to parse API response: {exc}",
            error_details=str(exc),
        ) from exc

chat_stream(request) async

Send a chat request and yield tokens as they arrive (SSE).

Parameters:

Name	Type	Description	Default
request	ChatRequest	A :class:~cygnisai_sdk_python.models.ChatRequest instance.	required

Yields:

Type	Description
AsyncGenerator[str, None]	str – individual text tokens from the model.

Raises:

Type	Description
CygnisAIError (and subclasses)	as documented on :meth:chat.

Source code in cygnisai_sdk_python/client.py

async def chat_stream(
    self, request: ChatRequest
) -> AsyncGenerator[str, None]:
    """
    Send a chat request and yield tokens as they arrive (SSE).

    Args:
        request: A :class:`~cygnisai_sdk_python.models.ChatRequest` instance.

    Yields:
        str – individual text tokens from the model.

    Raises:
        CygnisAIError (and subclasses): as documented on :meth:`chat`.
    """
    request.stream = True
    payload = request.model_dump(exclude_none=True)

    async with self._client.stream("POST", "/v3/chat", json=payload) as response:
        if response.status_code != 200:
            error_bytes = await response.aread()
            try:
                body = json.loads(error_bytes.decode())
                msg = _extract_error_message(body)
            except Exception:
                body = {}
                msg = error_bytes.decode()
            raise _map_status_error(
                response.status_code,
                f"API error (HTTP {response.status_code}): {msg}",
                body,
            )

        async for raw_line in response.aiter_lines():
            line = raw_line.strip()
            if not line:
                continue

            if line.startswith("data:"):
                content = line[5:].strip()

                if content == "[DONE]":
                    return

                try:
                    data = json.loads(content)
                except json.JSONDecodeError:
                    # Yield raw text if it isn't valid JSON
                    yield content
                    continue

                if not isinstance(data, dict):
                    yield str(data)
                    continue

                # Surface server-side errors embedded in the stream
                if "error" in data or "detail" in data:
                    err_key = "error" if "error" in data else "detail"
                    err_val = data[err_key]
                    if isinstance(err_val, list):
                        msgs = [
                            f"{e.get('loc', ['?'])[-1]}: {e.get('msg', '')}"
                            for e in err_val
                            if isinstance(e, dict)
                        ]
                        err_val = "; ".join(msgs)
                    raise CygnisAIError(
                        f"Server error in stream: {err_val}",
                        error_details=data,
                    )

                token = data.get("response") or data.get("text") or ""
                if token:
                    yield token

close() async

Release the underlying HTTP connection pool.

Source code in cygnisai_sdk_python/client.py

async def close(self) -> None:
    """Release the underlying HTTP connection pool."""
    if not self._client.is_closed:
        await self._client.aclose()

CygnisAIError

Bases: Exception

Base exception for all CygnisAI SDK errors.

Source code in cygnisai_sdk_python/client.py

class CygnisAIError(Exception):
    """Base exception for all CygnisAI SDK errors."""

    def __init__(
        self,
        message: str,
        status_code: Optional[int] = None,
        error_details: Optional[Any] = None,
    ) -> None:
        super().__init__(message)
        self.message = message
        self.status_code = status_code
        self.error_details = error_details

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"{self.__class__.__name__}("
            f"message={self.message!r}, "
            f"status_code={self.status_code!r})"
        )

ErrorResponse

Bases: BaseModel

Standardised error envelope returned by the API.

Source code in cygnisai_sdk_python/models.py

class ErrorResponse(BaseModel):
    """Standardised error envelope returned by the API."""

    code: str = Field(..., description="Machine-readable error code.")
    message: str = Field(..., description="Human-readable error message.")
    details: Optional[List[ErrorDetail]] = Field(
        None, description="Validation details when applicable."
    )

GenerativeModel

High-level interface for a CygnisAI language model.

Uses the global client configured via :func:configure.

Example::

model = GenerativeModel("alpha2")
response = model.generate_content("Hello!")
print(response.text)

Source code in cygnisai_sdk_python/__init__.py

class GenerativeModel:
    """
    High-level interface for a CygnisAI language model.

    Uses the global client configured via :func:`configure`.

    Example::

        model = GenerativeModel("alpha2")
        response = model.generate_content("Hello!")
        print(response.text)
    """

    def __init__(self, model_name: str) -> None:
        if not model_name:
            raise ValueError("model_name must not be empty.")
        self.model_name = model_name

    def _get_client(self) -> CygnisAIClient:
        if _global_client is None:
            raise CygnisAIError(
                "CygnisAI client is not configured. "
                "Call cygnis.configure(api_key='YOUR_KEY') first."
            )
        return _global_client

    def generate_content(
        self,
        prompt: str,
        messages: Optional[List[Message]] = None,
        stream: bool = False,
    ) -> Any:
        """
        Generate text from a prompt.

        Args:
            prompt: The question or instruction for the model.
            messages: Optional conversation history.
            stream: If ``True``, returns a :class:`GenerativeStreamResponse`
                    that you can ``async for`` over.

        Returns:
            :class:`GenerativeResponse` (blocking) or
            :class:`GenerativeStreamResponse` (streaming).
        """
        client = self._get_client()
        request = ChatRequest(
            model=self.model_name,
            prompt=prompt,
            messages=messages,
            stream=stream,
        )

        if stream:
            return GenerativeStreamResponse(client.chat_stream(request=request))

        # Synchronous execution via asyncio.run — works in scripts & notebooks
        async def _run() -> ChatResponse:
            return await client.chat(request=request)

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop and loop.is_running():
            # Inside an async context (e.g. Jupyter) — caller must await
            raise RuntimeError(
                "generate_content() cannot be called synchronously inside a running "
                "event loop. Use `await model.async_generate_content(...)` instead, "
                "or call it from a regular (non-async) context."
            )

        full_response = asyncio.run(_run())
        return GenerativeResponse(text=full_response.response, full_response=full_response)

    async def async_generate_content(
        self,
        prompt: str,
        messages: Optional[List[Message]] = None,
        stream: bool = False,
    ) -> Any:
        """
        Async version of :meth:`generate_content` for use inside ``async`` code.

        Example::

            response = await model.async_generate_content("Hello!")
        """
        client = self._get_client()
        request = ChatRequest(
            model=self.model_name,
            prompt=prompt,
            messages=messages,
            stream=stream,
        )

        if stream:
            return GenerativeStreamResponse(client.chat_stream(request=request))

        full_response = await client.chat(request=request)
        return GenerativeResponse(text=full_response.response, full_response=full_response)

async_generate_content(prompt, messages=None, stream=False) async

Async version of :meth:generate_content for use inside async code.

Example::

response = await model.async_generate_content("Hello!")

Source code in cygnisai_sdk_python/__init__.py

async def async_generate_content(
    self,
    prompt: str,
    messages: Optional[List[Message]] = None,
    stream: bool = False,
) -> Any:
    """
    Async version of :meth:`generate_content` for use inside ``async`` code.

    Example::

        response = await model.async_generate_content("Hello!")
    """
    client = self._get_client()
    request = ChatRequest(
        model=self.model_name,
        prompt=prompt,
        messages=messages,
        stream=stream,
    )

    if stream:
        return GenerativeStreamResponse(client.chat_stream(request=request))

    full_response = await client.chat(request=request)
    return GenerativeResponse(text=full_response.response, full_response=full_response)

generate_content(prompt, messages=None, stream=False)

Generate text from a prompt.

Parameters:

Name	Type	Description	Default
prompt	str	The question or instruction for the model.	required
messages	Optional[List[Message]]	Optional conversation history.	None
stream	bool	If True, returns a :class:GenerativeStreamResponse that you can async for over.	False

Returns:

Type	Description
Any	class:GenerativeResponse (blocking) or
Any	class:GenerativeStreamResponse (streaming).

Source code in cygnisai_sdk_python/__init__.py

def generate_content(
    self,
    prompt: str,
    messages: Optional[List[Message]] = None,
    stream: bool = False,
) -> Any:
    """
    Generate text from a prompt.

    Args:
        prompt: The question or instruction for the model.
        messages: Optional conversation history.
        stream: If ``True``, returns a :class:`GenerativeStreamResponse`
                that you can ``async for`` over.

    Returns:
        :class:`GenerativeResponse` (blocking) or
        :class:`GenerativeStreamResponse` (streaming).
    """
    client = self._get_client()
    request = ChatRequest(
        model=self.model_name,
        prompt=prompt,
        messages=messages,
        stream=stream,
    )

    if stream:
        return GenerativeStreamResponse(client.chat_stream(request=request))

    # Synchronous execution via asyncio.run — works in scripts & notebooks
    async def _run() -> ChatResponse:
        return await client.chat(request=request)

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    if loop and loop.is_running():
        # Inside an async context (e.g. Jupyter) — caller must await
        raise RuntimeError(
            "generate_content() cannot be called synchronously inside a running "
            "event loop. Use `await model.async_generate_content(...)` instead, "
            "or call it from a regular (non-async) context."
        )

    full_response = asyncio.run(_run())
    return GenerativeResponse(text=full_response.response, full_response=full_response)

GenerativeResponse

Wraps a completed :class:~cygnisai_sdk_python.models.ChatResponse.

Source code in cygnisai_sdk_python/__init__.py

class GenerativeResponse:
    """Wraps a completed :class:`~cygnisai_sdk_python.models.ChatResponse`."""

    def __init__(self, text: str, full_response: Optional[ChatResponse] = None) -> None:
        self.text = text
        self._full_response = full_response

    def __str__(self) -> str:
        return self.text

    def __repr__(self) -> str:
        snippet = self.text[:60].replace("\n", " ")
        return f"GenerativeResponse(text={snippet!r})"

    @property
    def full_response(self) -> Optional[ChatResponse]:
        """The raw :class:`~cygnisai_sdk_python.models.ChatResponse`, if available."""
        return self._full_response

    @property
    def usage(self) -> Optional[UsageInfo]:
        """Token-usage statistics, if available."""
        return self._full_response.usage if self._full_response else None

    @property
    def latency_ms(self) -> Optional[int]:
        """Round-trip latency reported by the API."""
        return self._full_response.latency_ms if self._full_response else None

full_response property

The raw :class:~cygnisai_sdk_python.models.ChatResponse, if available.

latency_ms property

Round-trip latency reported by the API.

usage property

Token-usage statistics, if available.

GenerativeStreamResponse

Wraps an async token generator from :meth:CygnisAIClient.chat_stream.

Source code in cygnisai_sdk_python/__init__.py

class GenerativeStreamResponse:
    """Wraps an async token generator from :meth:`CygnisAIClient.chat_stream`."""

    def __init__(self, async_generator: AsyncGenerator[str, None]) -> None:
        self._async_generator = async_generator

    def __aiter__(self):
        return self._async_generator.__aiter__()

Message

Bases: BaseModel

A single turn in a conversation.

Source code in cygnisai_sdk_python/models.py

class Message(BaseModel):
    """A single turn in a conversation."""

    role: Role = Field(..., description="Author role: 'user', 'assistant', or 'system'.")
    content: str = Field(..., min_length=1, description="Text content of the message.")

    model_config = {"use_enum_values": True}

NetworkError

Bases: CygnisAIError

Raised when the request never reaches the server.

Source code in cygnisai_sdk_python/client.py

class NetworkError(CygnisAIError):
    """Raised when the request never reaches the server."""

RateLimitError

Bases: CygnisAIError

Raised on HTTP 429.

Source code in cygnisai_sdk_python/client.py

class RateLimitError(CygnisAIError):
    """Raised on HTTP 429."""

ResponseValidationError

Bases: CygnisAIError

Raised when the API response doesn't match the expected schema.

Source code in cygnisai_sdk_python/client.py

class ResponseValidationError(CygnisAIError):
    """Raised when the API response doesn't match the expected schema."""

Role

Bases: str, Enum

Valid message roles.

Source code in cygnisai_sdk_python/models.py

class Role(str, Enum):
    """Valid message roles."""
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"

ServerError

Bases: CygnisAIError

Raised on HTTP 5xx.

Source code in cygnisai_sdk_python/client.py

class ServerError(CygnisAIError):
    """Raised on HTTP 5xx."""

UsageInfo

Bases: BaseModel

Token-usage breakdown returned by the API.

Source code in cygnisai_sdk_python/models.py

class UsageInfo(BaseModel):
    """Token-usage breakdown returned by the API."""

    prompt_tokens: Optional[int] = Field(None, ge=0)
    completion_tokens: Optional[int] = Field(None, ge=0)
    total_tokens: Optional[int] = Field(None, ge=0)

    # Keep extra fields from the API without failing validation
    model_config = {"extra": "allow"}

configure(api_key=None, base_url=None, timeout=30.0, max_retries=3)

Initialise the global CygnisAI client.

Call this once at startup before using :class:GenerativeModel.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	Your CygnisAI API key. Falls back to the CYGNIS_API_KEY environment variable if not provided.	None
base_url	Optional[str]	Override the default API base URL.	None
timeout	float	HTTP timeout in seconds (default 30).	30.0
max_retries	int	Number of automatic retries on transient errors (default 3).	3

Raises:

Type	Description
ValueError	If no API key is found.

Source code in cygnisai_sdk_python/__init__.py

def configure(
    api_key: Optional[str] = None,
    base_url: Optional[str] = None,
    timeout: float = 30.0,
    max_retries: int = 3,
) -> None:
    """
    Initialise the global CygnisAI client.

    Call this once at startup before using :class:`GenerativeModel`.

    Args:
        api_key: Your CygnisAI API key. Falls back to the ``CYGNIS_API_KEY``
                 environment variable if not provided.
        base_url: Override the default API base URL.
        timeout: HTTP timeout in seconds (default 30).
        max_retries: Number of automatic retries on transient errors (default 3).

    Raises:
        ValueError: If no API key is found.
    """
    global _global_client, _global_api_key, _global_base_url

    resolved_key = api_key or os.environ.get("CYGNIS_API_KEY")
    if not resolved_key:
        raise ValueError(
            "No API key provided. Pass api_key= or set the CYGNIS_API_KEY "
            "environment variable."
        )

    _global_api_key = resolved_key
    if base_url:
        _global_base_url = base_url

    # Cleanly close the previous client if one exists
    if _global_client is not None:
        try:
            asyncio.run(_global_client.close())
        except RuntimeError:
            pass  # Already inside a running loop — leak is acceptable here

    _global_client = CygnisAIClient(
        api_key=_global_api_key,
        base_url=_global_base_url,
        timeout=timeout,
        max_retries=max_retries,
    )