"""Synchronous and asynchronous Taimoe Platform client."""
from __future__ import annotations
import asyncio
import time
import uuid
from types import TracebackType
from typing import Any
import httpx
from taimoe.platform.errors import DeadlineExceededError, TaimoeAPIError
from taimoe.platform.errors._base import _Retryable
from taimoe.platform.resources.agents import AgentsResource
from taimoe.platform.resources.knowledge_bases import KnowledgeBasesResource
from taimoe.platform.resources.observability import ObservabilityResource
from taimoe.platform.resources.runtimes import RuntimesResource
from taimoe.platform.version import __version__
_MAX_BACKOFF_SECONDS = 30.0
[docs]
class TaimoeClient:
"""REST client for the Taimoe Platform.
Follows Google ADK style: one client class; async operations use the
``*_async`` suffix on the same class instead of a separate
``AsyncTaimoeClient``.
The client owns a long-lived ``httpx.Client`` and ``httpx.AsyncClient``
so connection pools and keep-alive are reused across requests. Use as a
context manager or call :meth:`close` / :meth:`aclose` when done.
Retries are opt-in via ``max_retries`` (default 2). The client retries
network errors and HTTP 408/429/5xx with exponential backoff, honouring
the server's ``Retry-After`` header when present. Set ``max_retries=0``
to disable retries entirely.
Every request carries an ``X-Request-ID`` header for audit correlation.
A UUID4 is generated automatically; callers can override per request by
passing ``request_id=...`` through the resource layer.
"""
def __init__(
self,
*,
base_url: str,
api_key: str | None = None,
org: str = "default",
timeout: float | httpx.Timeout = 10.0,
max_retries: int = 2,
) -> None:
if max_retries < 0:
raise ValueError("max_retries must be >= 0")
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.org = org
self.timeout = timeout
self.max_retries = max_retries
headers = self._default_headers()
self._http = httpx.Client(base_url=self.base_url, timeout=timeout, headers=headers)
self._http_async = httpx.AsyncClient(
base_url=self.base_url, timeout=timeout, headers=headers
)
self.runtimes = RuntimesResource(self)
self.agents = AgentsResource(self)
self.knowledge_bases = KnowledgeBasesResource(self)
self.observability = ObservabilityResource(self)
def _request(
self,
method: str,
path: str,
*,
request_id: str | None = None,
**kwargs: Any,
) -> Any:
"""Synchronous internal request helper."""
headers = self._merge_request_headers(kwargs.pop("headers", None), request_id)
attempts = self.max_retries + 1
for attempt in range(attempts):
try:
response = self._http.request(method, path, headers=headers, **kwargs)
except httpx.RequestError as exc:
if attempt + 1 < attempts:
time.sleep(self._backoff(attempt))
continue
raise self._transport_error(path, exc) from exc
if response.is_error:
error = TaimoeAPIError.from_response(response)
if isinstance(error, _Retryable) and attempt + 1 < attempts:
time.sleep(self._retry_delay(response, attempt))
continue
raise error
return self._parse_body(response)
raise RuntimeError(f"_request loop exited without resolution for {path}")
async def _request_async(
self,
method: str,
path: str,
*,
request_id: str | None = None,
**kwargs: Any,
) -> Any:
"""Asynchronous internal request helper."""
headers = self._merge_request_headers(kwargs.pop("headers", None), request_id)
attempts = self.max_retries + 1
for attempt in range(attempts):
try:
response = await self._http_async.request(
method, path, headers=headers, **kwargs
)
except httpx.RequestError as exc:
if attempt + 1 < attempts:
await asyncio.sleep(self._backoff(attempt))
continue
raise self._transport_error(path, exc) from exc
if response.is_error:
error = TaimoeAPIError.from_response(response)
if isinstance(error, _Retryable) and attempt + 1 < attempts:
await asyncio.sleep(self._retry_delay(response, attempt))
continue
raise error
return self._parse_body(response)
raise RuntimeError(f"_request_async loop exited without resolution for {path}")
@staticmethod
def _parse_body(response: httpx.Response) -> Any:
if not response.content:
return None
return response.json()
@staticmethod
def _transport_error(path: str, exc: httpx.RequestError) -> TaimoeAPIError:
"""Wrap transport-level failures (connect / read / timeout) as a
canonical retryable error so callers never see raw httpx exceptions."""
if isinstance(exc, httpx.TimeoutException):
return DeadlineExceededError(
f"Request to {path} timed out: {exc}", status_code=0
)
return DeadlineExceededError(
f"Request to {path} failed at transport: {exc}", status_code=0
)
def _default_headers(self) -> dict[str, str]:
headers = {
"User-Agent": f"taimoe-platform/{__version__}",
"X-Taimoe-Org": self.org,
}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
return headers
def _merge_request_headers(
self, caller_headers: dict[str, str] | None, request_id: str | None
) -> dict[str, str]:
merged = dict(caller_headers) if caller_headers else {}
merged.setdefault("X-Request-ID", request_id or str(uuid.uuid4()))
return merged
def _backoff(self, attempt: int) -> float:
return min(2.0**attempt, _MAX_BACKOFF_SECONDS)
def _retry_delay(self, response: httpx.Response, attempt: int) -> float:
retry_after = response.headers.get("Retry-After")
if retry_after:
try:
return min(float(retry_after), _MAX_BACKOFF_SECONDS)
except ValueError:
pass
return self._backoff(attempt)
[docs]
def close(self) -> None:
"""Close the underlying synchronous HTTP client."""
self._http.close()
[docs]
async def aclose(self) -> None:
"""Close the underlying asynchronous HTTP client."""
await self._http_async.aclose()
def __enter__(self) -> TaimoeClient:
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
self.close()
async def __aenter__(self) -> TaimoeClient:
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
await self.aclose()