API reference¶
Generated from the apogee-ai-providers source with mkdocstrings. Every symbol below is exported from apogee_ai_providers, so it is part of the supported public surface.
Other¶
Python
AnthropicChatProvider(credentials: ProviderCredentials, *, config: AnthropicConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IChatCompletionProvider against Anthropic Messages API.
Uses x-api-key and anthropic-version headers (no Bearer).
Source code in apogee_ai_providers/infrastructure/providers/anthropic/anthropic_chat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: AnthropicConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or AnthropicConfig()
merged_headers = {
"x-api-key": credentials.api_key,
"anthropic-version": self._config.api_version,
**credentials.extra_headers,
}
effective = replace(
credentials,
base_url=credentials.base_url or self._config.base_url,
extra_headers=merged_headers,
)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider="anthropic",
credentials=self._credentials,
auth_scheme="none",
)
async
¶
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
Source code in apogee_ai_providers/infrastructure/providers/anthropic/anthropic_chat_provider.py
async
¶
Python
stream(request: ChatRequest) -> AsyncIterator[ChatChunk]
Source code in apogee_ai_providers/infrastructure/providers/anthropic/anthropic_chat_provider.py
Python
async def stream(self, request: ChatRequest) -> AsyncIterator[ChatChunk]:
payload = request_to_anthropic_payload(replace(request, stream=True))
state = AnthropicStreamState(default_model=request.model)
async for raw in self._http.stream_sse("/messages", payload):
chunk = anthropic_event_to_chunk(raw, state=state)
if chunk is not None:
yield chunk
dataclass
¶
Python
AsyncHttpClient(*, provider: str, credentials: ProviderCredentials, max_retries: int = 2, backoff_base: float = 0.5, auth_scheme: str = 'bearer')
Thin async HTTP client. One instance per provider adapter.
auth_scheme: "bearer" (Authorization: Bearer
Source code in apogee_ai_providers/infrastructure/http/async_http_client.py
Python
def __init__(
self,
*,
provider: str,
credentials: ProviderCredentials,
max_retries: int = 2,
backoff_base: float = 0.5,
auth_scheme: str = "bearer",
) -> None:
"""auth_scheme: "bearer" (Authorization: Bearer <key>) or "none" (extra_headers only)."""
self._provider = provider
self._credentials = credentials
self._max_retries = max_retries
self._backoff_base = backoff_base
self._auth_scheme = auth_scheme
self._client = httpx.AsyncClient(
base_url=credentials.base_url or "",
timeout=credentials.timeout,
headers=self._default_headers(),
)
async
¶
async
¶
Python
post_json(url: str, payload: Mapping[str, Any], *, extra_headers: Mapping[str, str] | None = None) -> dict[str, Any]
Source code in apogee_ai_providers/infrastructure/http/async_http_client.py
async
¶
Python
post_bytes(url: str, payload: Mapping[str, Any], *, extra_headers: Mapping[str, str] | None = None) -> bytes
POST a JSON payload, return the raw response body (e.g. audio bytes).
Source code in apogee_ai_providers/infrastructure/http/async_http_client.py
Python
async def post_bytes(
self,
url: str,
payload: Mapping[str, Any],
*,
extra_headers: Mapping[str, str] | None = None,
) -> bytes:
"""POST a JSON payload, return the raw response body (e.g. audio bytes)."""
response = await self._request_with_retry(
"POST", url, json=payload, headers=extra_headers
)
return response.content
async
¶
Python
post_multipart(url: str, *, data: Mapping[str, Any] | None = None, files: Mapping[str, tuple[str, bytes, str]] | None = None, extra_headers: Mapping[str, str] | None = None) -> dict[str, Any]
POST a multipart/form-data request — used by STT endpoints.
We open a one-shot AsyncClient without the JSON default Content-Type so httpx can negotiate the multipart boundary itself.
Source code in apogee_ai_providers/infrastructure/http/async_http_client.py
Python
async def post_multipart(
self,
url: str,
*,
data: Mapping[str, Any] | None = None,
files: Mapping[str, tuple[str, bytes, str]] | None = None,
extra_headers: Mapping[str, str] | None = None,
) -> dict[str, Any]:
"""POST a multipart/form-data request — used by STT endpoints.
We open a one-shot AsyncClient without the JSON default Content-Type
so httpx can negotiate the multipart boundary itself.
"""
attempt = 0
# Build auth headers without Content-Type so httpx can set multipart.
headers: dict[str, str] = {}
if self._auth_scheme == "bearer":
headers["Authorization"] = f"Bearer {self._credentials.api_key}"
headers.update(self._credentials.extra_headers)
if extra_headers:
headers.update(extra_headers)
base_url = str(self._client.base_url) or self._credentials.base_url or ""
timeout = self._credentials.timeout
async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as fresh:
while True:
try:
response = await fresh.post(
url,
data=dict(data or {}),
files=dict(files or {}),
headers=headers,
)
if response.status_code == 429 and attempt < self._max_retries:
retry_after = self._parse_retry_after(response)
attempt += 1
await asyncio.sleep(
retry_after or self._backoff_base * (2 ** (attempt - 1))
)
continue
if 500 <= response.status_code < 600 and attempt < self._max_retries:
attempt += 1
await asyncio.sleep(self._backoff_base * (2 ** (attempt - 1)))
continue
self._raise_for_status(response)
return response.json()
except (httpx.TimeoutException, httpx.TransportError) as exc:
attempt += 1
if attempt > self._max_retries:
raise self._map_transport_error(exc) from exc
await asyncio.sleep(self._backoff_base * (2 ** (attempt - 1)))
async
¶
Python
stream_sse(url: str, payload: Mapping[str, Any], *, extra_headers: Mapping[str, str] | None = None) -> AsyncIterator[str]
Yields raw SSE data: lines (without the data: prefix).
Source code in apogee_ai_providers/infrastructure/http/async_http_client.py
Python
async def stream_sse(
self,
url: str,
payload: Mapping[str, Any],
*,
extra_headers: Mapping[str, str] | None = None,
) -> AsyncIterator[str]:
"""Yields raw SSE `data:` lines (without the `data: ` prefix)."""
headers = dict(extra_headers or {})
headers.setdefault("Accept", "text/event-stream")
attempt = 0
while True:
try:
async with self._client.stream(
"POST", url, json=payload, headers=headers
) as response:
self._raise_for_status(response)
async for line in response.aiter_lines():
if not line:
continue
if line.startswith("data: "):
yield line[6:]
elif line.startswith("data:"):
yield line[5:]
return
except (httpx.TimeoutException, httpx.TransportError) as exc:
attempt += 1
if attempt > self._max_retries:
raise self._map_transport_error(exc) from exc
await asyncio.sleep(self._backoff_base * (2 ** (attempt - 1)))
Python
AzureOpenAIChatProvider(credentials: ProviderCredentials, *, config: AzureOpenAIConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IChatCompletionProvider against Azure OpenAI.
Source code in apogee_ai_providers/infrastructure/providers/azure/azure_chat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: AzureOpenAIConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or AzureOpenAIConfig()
if not credentials.base_url:
raise ValueError(
"Azure OpenAI requires `base_url` to be set on ProviderCredentials "
"(e.g. https://<resource>.openai.azure.com)."
)
self._credentials = credentials
# Azure uses a custom header; we always start with auth_scheme="none"
# on the underlying client and inject our own header below.
self._http = http_client or AsyncHttpClient(
provider=self.provider_name,
credentials=self._auth_credentials(credentials),
auth_scheme="bearer" if self._config.auth_scheme == "bearer" else "none",
)
async
¶
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
Source code in apogee_ai_providers/infrastructure/providers/azure/azure_chat_provider.py
Python
async def complete(self, request: ChatRequest) -> ChatResponse:
deployment = self._resolve_deployment(request)
payload = request_to_openai_payload(replace(request, stream=False))
data = await self._http.post_json(self._path_for(deployment), payload)
response = openai_response_to_domain(data)
return replace(response, provider=self.provider_name)
async
¶
Python
stream(request: ChatRequest) -> AsyncIterator[ChatChunk]
Source code in apogee_ai_providers/infrastructure/providers/azure/azure_chat_provider.py
Python
async def stream(self, request: ChatRequest) -> AsyncIterator[ChatChunk]:
deployment = self._resolve_deployment(request)
payload = request_to_openai_payload(replace(request, stream=True))
async for raw in self._http.stream_sse(self._path_for(deployment), payload):
chunk = openai_chunk_to_domain(raw)
if chunk is not None:
yield chunk
dataclass
¶
Python
AzureOpenAIConfig(api_version: str = '2024-10-21', default_deployment: str | None = None, auth_scheme: str = 'api-key')
class-attribute
instance-attribute
¶
Either "api-key" (header api-key:
Python
BedrockChatProvider(credentials: ProviderCredentials, *, config: BedrockConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IChatCompletionProvider against AWS Bedrock Converse API.
Source code in apogee_ai_providers/infrastructure/providers/bedrock/bedrock_chat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: BedrockConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or BedrockConfig()
effective = replace(
credentials,
base_url=credentials.base_url or self._config.base_url,
)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider="bedrock",
credentials=self._credentials,
auth_scheme="bearer",
)
async
¶
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
Source code in apogee_ai_providers/infrastructure/providers/bedrock/bedrock_chat_provider.py
Python
async def complete(self, request: ChatRequest) -> ChatResponse:
payload = request_to_bedrock_payload(replace(request, stream=False))
model_id = quote(request.model, safe="")
path = f"/model/{model_id}/converse"
data = await self._http.post_json(path, payload)
return bedrock_response_to_domain(data, model=request.model)
async
¶
Python
stream(request: ChatRequest) -> AsyncIterator[ChatChunk]
Source code in apogee_ai_providers/infrastructure/providers/bedrock/bedrock_chat_provider.py
Python
async def stream(self, request: ChatRequest) -> AsyncIterator[ChatChunk]:
raise NotImplementedError(
"Bedrock streaming uses AWS Event Stream (binary framing); not yet supported. "
"Use `complete()` instead."
)
# hint for type-checker: make this an async generator
yield # type: ignore[unreachable] # pragma: no cover
dataclass
¶
Holds Bedrock runtime endpoint + default model.
The Converse API is region-scoped. Default region is us-east-1.
Override via env var AI_PROVIDER_URL_BEDROCK (full base URL) or pass
credentials.base_url.
dataclass
¶
Python
ChatChoice(index: int, message: ChatMessage, finish_reason: FinishReason | None = None)
dataclass
¶
Python
ChatChunk(id: str, model: str, delta: str = '', tool_calls: list[ToolCall] = list(), finish_reason: FinishReason | None = None)
dataclass
¶
Python
ChatMessage(role: MessageRole, content: str | None = None, name: str | None = None, tool_calls: list[ToolCall] = list(), tool_call_id: str | None = None)
dataclass
¶
Python
ChatRequest(model: str, messages: list[ChatMessage], tools: list[ToolDefinition] = list(), tool_choice: str | dict[str, Any] | None = None, thinking: ThinkingConfig | None = None, stream: bool = False, temperature: float | None = None, top_p: float | None = None, max_tokens: int | None = None, stop: list[str] | None = None, response_format: dict[str, Any] | None = None, seed: int | None = None, user: str | None = None, metadata: dict[str, Any] = dict())
class-attribute
instance-attribute
¶
Python
tools: list[ToolDefinition] = field(default_factory=list)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
dataclass
¶
Python
ChatResponse(id: str, model: str, choices: list[ChatChoice], usage: TokenUsage = TokenUsage(), provider: str = '')
class-attribute
instance-attribute
¶
Python
usage: TokenUsage = field(default_factory=TokenUsage)
Python
DeepSeekChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
DeepSeekConfig(base_url: str = 'https://api.deepseek.com/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'deepseek-chat')
Bases: OpenAICompatibleConfig
dataclass
¶
Python
EmbeddingRequest(model: str, inputs: list[str], dimensions: int | None = None, encoding_format: str = 'float', user: str | None = None)
One or more inputs to embed against a single model.
dataclass
¶
Python
EmbeddingResponse(model: str, provider: str, embeddings: list[list[float]], usage: TokenUsage)
Bases: str, Enum
Python
GeminiChatProvider(credentials: ProviderCredentials, *, config: GeminiConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IChatCompletionProvider against Google Generative Language API.
Authentication uses the x-goog-api-key header (no Bearer).
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_chat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: GeminiConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or GeminiConfig()
merged_headers = {
"x-goog-api-key": credentials.api_key,
**credentials.extra_headers,
}
effective = replace(
credentials,
base_url=credentials.base_url or self._config.base_url,
extra_headers=merged_headers,
)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider="gemini",
credentials=self._credentials,
auth_scheme="none",
)
async
¶
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_chat_provider.py
async
¶
Python
stream(request: ChatRequest) -> AsyncIterator[ChatChunk]
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_chat_provider.py
Python
async def stream(self, request: ChatRequest) -> AsyncIterator[ChatChunk]:
payload = request_to_gemini_payload(replace(request, stream=True))
path = f"/models/{request.model}:streamGenerateContent?alt=sse"
async for raw in self._http.stream_sse(path, payload):
chunk = gemini_chunk_to_domain(raw, model=request.model)
if chunk is not None:
yield chunk
dataclass
¶
Python
GeminiEmbeddingProvider(credentials: ProviderCredentials, *, config: GeminiConfig | None = None, http_client: AsyncHttpClient | None = None)
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_embedding_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: GeminiConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or GeminiConfig()
merged_headers = {
"x-goog-api-key": credentials.api_key,
**credentials.extra_headers,
}
effective = replace(
credentials,
base_url=credentials.base_url or self._config.base_url,
extra_headers=merged_headers,
)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider=self.provider_name,
credentials=self._credentials,
auth_scheme="none",
)
async
¶
async
¶
Python
embed(request: EmbeddingRequest) -> EmbeddingResponse
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_embedding_provider.py
Python
async def embed(self, request: EmbeddingRequest) -> EmbeddingResponse:
if not request.inputs:
raise ValueError("EmbeddingRequest.inputs cannot be empty")
model_id = quote(request.model, safe="")
url = f"/models/{model_id}:batchEmbedContents"
payload: dict[str, Any] = {
"requests": [
{
"model": f"models/{request.model}",
"content": {"parts": [{"text": text}]},
**(
{"outputDimensionality": request.dimensions}
if request.dimensions is not None
else {}
),
}
for text in request.inputs
]
}
data = await self._http.post_json(url, payload)
embeddings = [
[float(v) for v in (entry.get("values") or [])]
for entry in (data.get("embeddings") or [])
]
usage_meta = data.get("usageMetadata") or {}
return EmbeddingResponse(
model=request.model,
provider=self.provider_name,
embeddings=embeddings,
usage=TokenUsage(
prompt_tokens=int(usage_meta.get("promptTokenCount", 0)),
completion_tokens=0,
total_tokens=int(usage_meta.get("totalTokenCount", 0)),
),
)
Python
GeminiMultimodalProvider(credentials: ProviderCredentials, *, config: GeminiConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IMultimodalProvider using Gemini's generateContent.
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_multimodal_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: GeminiConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or GeminiConfig()
merged_headers = {
"x-goog-api-key": credentials.api_key,
**credentials.extra_headers,
}
effective = replace(
credentials,
base_url=credentials.base_url or self._config.base_url,
extra_headers=merged_headers,
)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider="gemini",
credentials=self._credentials,
auth_scheme="none",
)
async
¶
async
¶
Python
generate(request: MultimodalRequest) -> MultimodalResponse
Source code in apogee_ai_providers/infrastructure/providers/gemini/gemini_multimodal_provider.py
Python
async def generate(self, request: MultimodalRequest) -> MultimodalResponse:
parts = [self._to_part(inp) for inp in request.inputs]
payload: dict[str, Any] = {"contents": [{"role": "user", "parts": parts}]}
if request.instructions:
payload["systemInstruction"] = {
"role": "system",
"parts": [{"text": request.instructions}],
}
gen_config: dict[str, Any] = {}
if request.max_tokens is not None:
gen_config["maxOutputTokens"] = request.max_tokens
if request.temperature is not None:
gen_config["temperature"] = request.temperature
if gen_config:
payload["generationConfig"] = gen_config
path = f"/models/{request.model}:generateContent"
data = await self._http.post_json(path, payload)
text = ""
for cand in data.get("candidates") or []:
for part in (cand.get("content") or {}).get("parts") or []:
if "text" in part:
text += str(part["text"])
usage_meta = data.get("usageMetadata") or {}
return MultimodalResponse(
id=str(data.get("responseId") or ""),
model=str(data.get("modelVersion") or request.model),
output_text=text,
usage=TokenUsage(
prompt_tokens=int(usage_meta.get("promptTokenCount", 0)),
completion_tokens=int(usage_meta.get("candidatesTokenCount", 0)),
total_tokens=int(usage_meta.get("totalTokenCount", 0)),
reasoning_tokens=int(usage_meta.get("thoughtsTokenCount", 0)),
),
provider="gemini",
)
Python
HuggingFaceChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
HuggingFaceConfig(base_url: str = 'https://router.huggingface.co/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'meta-llama/Llama-3.3-70B-Instruct')
Bases: OpenAIConfig
Python
KimiChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
KimiConfig(base_url: str = 'https://api.moonshot.cn/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'moonshot-v1-8k')
Bases: OpenAICompatibleConfig
Bases: str, Enum
Python
MetaChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
MetaConfig(base_url: str = 'https://api.llama.com/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'Llama-3.3-70B-Instruct')
Bases: OpenAIConfig
Bases: str, Enum
dataclass
¶
Python
MultimodalInput(modality: Modality, text: str | None = None, url: str | None = None, data_b64: str | None = None, mime_type: str | None = None)
dataclass
¶
Python
MultimodalRequest(model: str, inputs: list[MultimodalInput], instructions: str | None = None, max_tokens: int | None = None, temperature: float | None = None, metadata: dict[str, Any] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
Python
MultimodalResponse(id: str, model: str, output_text: str, usage: TokenUsage = TokenUsage(), provider: str = '')
class-attribute
instance-attribute
¶
Python
usage: TokenUsage = field(default_factory=TokenUsage)
Python
NvidiaChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
NvidiaConfig(base_url: str = 'https://integrate.api.nvidia.com/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'meta/llama-3.3-70b-instruct')
Bases: OpenAIConfig
dataclass
¶
Python
OpenAIAudioConfig(base_url: str = 'https://api.openai.com/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'gpt-4o-mini', speech_path: str = '/audio/speech', transcription_path: str = '/audio/transcriptions')
Bases: OpenAIConfig
Python
OpenAIAudioProvider(credentials: ProviderCredentials, *, config: OpenAIAudioConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements ITextToSpeechProvider + ISpeechToTextProvider for OpenAI.
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_audio_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAIAudioConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or OpenAIAudioConfig()
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=self._config.base_url)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider=self.provider_name, credentials=self._credentials
)
async
¶
async
¶
Python
synthesize(request: TextToSpeechRequest) -> TextToSpeechResponse
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_audio_provider.py
Python
async def synthesize(self, request: TextToSpeechRequest) -> TextToSpeechResponse:
payload = {
"model": request.model,
"input": request.text,
"voice": request.voice,
"response_format": request.audio_format,
}
if request.speed is not None:
payload["speed"] = request.speed
audio_bytes = await self._http.post_bytes(self._config.speech_path, payload)
return TextToSpeechResponse(
audio=audio_bytes,
audio_format=request.audio_format,
model=request.model,
provider=self.provider_name,
)
async
¶
Python
transcribe(request: SpeechToTextRequest) -> SpeechToTextResponse
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_audio_provider.py
Python
async def transcribe(self, request: SpeechToTextRequest) -> SpeechToTextResponse:
filename = f"audio.{request.audio_format}"
mime = _audio_mime(request.audio_format)
data: dict[str, str] = {"model": request.model}
if request.language:
data["language"] = request.language
if request.prompt:
data["prompt"] = request.prompt
result = await self._http.post_multipart(
self._config.transcription_path,
data=data,
files={"file": (filename, request.audio, mime)},
)
return SpeechToTextResponse(
text=str(result.get("text", "")),
model=request.model,
provider=self.provider_name,
language=str(result.get("language")) if result.get("language") else None,
duration_seconds=(
float(result["duration"]) if "duration" in result else None
),
)
Python
OpenAIChatProvider(credentials: ProviderCredentials, *, config: OpenAIConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IChatCompletionProvider against OpenAI's /chat/completions.
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_chat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAIConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or OpenAIConfig()
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=self._config.base_url)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider="openai", credentials=self._credentials
)
async
¶
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_chat_provider.py
async
¶
Python
stream(request: ChatRequest) -> AsyncIterator[ChatChunk]
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_chat_provider.py
Python
async def stream(self, request: ChatRequest) -> AsyncIterator[ChatChunk]:
payload = request_to_openai_payload(replace(request, stream=True))
async for raw in self._http.stream_sse(self._config.chat_completions_path, payload):
chunk = openai_chunk_to_domain(raw)
if chunk is not None:
yield chunk
Python
OpenAICompatibleChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAIChatProvider
Subclass that re-stamps responses/chunks with the upstream provider name.
Subclasses set :attr:provider_name and override :attr:_default_config.
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
async
¶
Python
stream(request: ChatRequest)
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
dataclass
¶
dataclass
¶
Python
OpenAIConfig(base_url: str = 'https://api.openai.com/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'gpt-4o-mini')
class-attribute
instance-attribute
¶
dataclass
¶
Python
OpenAIEmbeddingConfig(base_url: str = 'https://api.openai.com/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'gpt-4o-mini', embeddings_path: str = '/embeddings')
Bases: OpenAIConfig
Python
OpenAIEmbeddingProvider(credentials: ProviderCredentials, *, config: OpenAIEmbeddingConfig | None = None, http_client: AsyncHttpClient | None = None)
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_embedding_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAIEmbeddingConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or OpenAIEmbeddingConfig()
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=self._config.base_url)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider=self.provider_name, credentials=self._credentials
)
async
¶
async
¶
Python
embed(request: EmbeddingRequest) -> EmbeddingResponse
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_embedding_provider.py
Python
async def embed(self, request: EmbeddingRequest) -> EmbeddingResponse:
if not request.inputs:
raise ValueError("EmbeddingRequest.inputs cannot be empty")
payload: dict[str, Any] = {
"model": request.model,
"input": request.inputs,
"encoding_format": request.encoding_format,
}
if request.dimensions is not None:
payload["dimensions"] = request.dimensions
if request.user:
payload["user"] = request.user
data = await self._http.post_json(self._config.embeddings_path, payload)
return _openai_embeddings_to_domain(data, provider=self.provider_name)
Python
OpenAIMultimodalProvider(credentials: ProviderCredentials, *, config: OpenAIConfig | None = None, http_client: AsyncHttpClient | None = None)
Implements IMultimodalProvider. Supports text + image inputs natively.
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_multimodal_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAIConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
self._config = config or OpenAIConfig()
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=self._config.base_url)
self._credentials = effective
self._http = http_client or AsyncHttpClient(
provider="openai", credentials=self._credentials
)
async
¶
async
¶
Python
generate(request: MultimodalRequest) -> MultimodalResponse
Source code in apogee_ai_providers/infrastructure/providers/openai/openai_multimodal_provider.py
Python
async def generate(self, request: MultimodalRequest) -> MultimodalResponse:
content_parts = [self._to_content_part(inp) for inp in request.inputs]
messages: list[dict[str, Any]] = []
if request.instructions:
messages.append({"role": "system", "content": request.instructions})
messages.append({"role": "user", "content": content_parts})
payload: dict[str, Any] = {
"model": request.model,
"messages": messages,
}
if request.max_tokens is not None:
payload["max_tokens"] = request.max_tokens
if request.temperature is not None:
payload["temperature"] = request.temperature
data = await self._http.post_json(self._config.chat_completions_path, payload)
choices = data.get("choices") or []
text = ""
if choices:
text = str((choices[0].get("message") or {}).get("content") or "")
usage = data.get("usage") or {}
return MultimodalResponse(
id=str(data.get("id", "")),
model=str(data.get("model", "")),
output_text=text,
usage=TokenUsage(
prompt_tokens=int(usage.get("prompt_tokens", 0)),
completion_tokens=int(usage.get("completion_tokens", 0)),
total_tokens=int(usage.get("total_tokens", 0)),
),
provider="openai",
)
Python
OpenRouterChatProvider(credentials: ProviderCredentials, *, config: OpenRouterConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAIChatProvider
Thin wrapper: OpenAI-compatible endpoint at openrouter.ai.
Source code in apogee_ai_providers/infrastructure/providers/openrouter/openrouter_chat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenRouterConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or OpenRouterConfig()
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
# Override provider string stored by the HTTP client for error attribution.
self._http._provider = "openrouter" # noqa: SLF001
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
dataclass
¶
Python
OpenRouterConfig(base_url: str = 'https://openrouter.ai/api/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'openai/gpt-4o-mini')
Bases: OpenAIConfig
class-attribute
instance-attribute
¶
Bases: str, Enum
dataclass
¶
Python
ProviderCredentials(api_key: str, base_url: str | None = None, organization: str | None = None, project: str | None = None, timeout: float = 60.0, extra_headers: dict[str, str] = dict())
class-attribute
instance-attribute
¶
Builds a chat provider from a Provider value or canonical string.
staticmethod
¶
Python
build(provider: Provider | str, credentials: ProviderCredentials) -> IChatCompletionProvider
Source code in apogee_ai_providers/infrastructure/factory/provider_factory.py
Python
@staticmethod
def build(
provider: Provider | str,
credentials: ProviderCredentials,
) -> IChatCompletionProvider:
key = provider.value if isinstance(provider, Provider) else str(provider).lower()
if key == Provider.OPENAI.value:
return OpenAIChatProvider(credentials)
if key == Provider.GEMINI.value:
return GeminiChatProvider(credentials)
if key == Provider.BEDROCK.value:
return BedrockChatProvider(credentials)
if key == Provider.ANTHROPIC.value:
return AnthropicChatProvider(credentials)
if key == Provider.OPENROUTER.value:
return OpenRouterChatProvider(credentials)
if key == Provider.AZURE.value:
return AzureOpenAIChatProvider(credentials)
if key in _OPENAI_COMPAT_PROVIDERS:
return _OPENAI_COMPAT_PROVIDERS[key](credentials)
if key in STUB_PROVIDERS:
return STUB_PROVIDERS[key](credentials)
raise ValueError(f"Unknown provider: {key}")
Python
QwenChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
QwenConfig(base_url: str = 'https://dashscope.aliyuncs.com/compatible-mode/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'qwen-plus')
Bases: OpenAICompatibleConfig
Alibaba DashScope OpenAI-compatible endpoint.
dataclass
¶
Python
SpeechToTextRequest(model: str, audio: bytes, audio_format: str = 'wav', language: str | None = None, prompt: str | None = None)
Transcribe spoken audio to text.
dataclass
¶
Python
SpeechToTextResponse(text: str, model: str, provider: str, language: str | None = None, duration_seconds: float | None = None)
Transcript text plus optional metadata returned by the provider.
dataclass
¶
Python
TextToSpeechRequest(model: str, text: str, voice: str = 'alloy', audio_format: str = 'mp3', speed: float | None = None)
Synthesise spoken audio from text.
dataclass
¶
dataclass
¶
dataclass
¶
Python
TokenUsage(prompt_tokens: int = 0, completion_tokens: int = 0, total_tokens: int = 0, reasoning_tokens: int = 0)
dataclass
¶
dataclass
¶
Python
XAIChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
XAIConfig(base_url: str = 'https://api.x.ai/v1', chat_completions_path: str = '/chat/completions', default_model: str = 'grok-2-latest')
Bases: OpenAICompatibleConfig
Python
ZAIChatProvider(credentials: ProviderCredentials, *, config: OpenAICompatibleConfig | None = None, http_client: AsyncHttpClient | None = None)
Bases: OpenAICompatibleChatProvider
Source code in apogee_ai_providers/infrastructure/providers/openai_compat/openai_compat_provider.py
Python
def __init__(
self,
credentials: ProviderCredentials,
*,
config: OpenAICompatibleConfig | None = None,
http_client: AsyncHttpClient | None = None,
) -> None:
cfg = config or self._default_config
effective = credentials
if credentials.base_url is None:
effective = replace(credentials, base_url=cfg.base_url)
super().__init__(effective, config=cfg, http_client=http_client)
self._http._provider = self.provider_name # noqa: SLF001
dataclass
¶
Python
ZAIConfig(base_url: str = 'https://api.z.ai/api/paas/v4', chat_completions_path: str = '/chat/completions', default_model: str = 'glm-4-plus')
Bases: OpenAICompatibleConfig
Z.ai (BigModel / GLM) OpenAI-compatible endpoint.
Other · DTOs¶
Bases: BaseModel
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
tool_calls: list[ToolCallDTO] = Field(default_factory=list)
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
tool_calls: list[ToolCallDTO] = Field(default_factory=list)
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
tools: list[ToolDefinitionDTO] = Field(default_factory=list)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
usage: TokenUsageDTO = Field(default_factory=TokenUsageDTO)
Bases: BaseModel
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
usage: TokenUsageDTO = Field(default_factory=TokenUsageDTO)
Bases: BaseModel
Other · Exceptions¶
Python
ProviderAuthError(message: str, *, provider: str, status: int | None = None, code: str | None = None, raw: Any = None)
Python
ProviderError(message: str, *, provider: str, status: int | None = None, code: str | None = None, raw: Any = None)
Bases: Exception
Raised when a provider call fails.
Attributes:
| Name | Type | Description |
|---|---|---|
provider |
Provider identifier (e.g. "openai"). |
|
status |
HTTP status code, if available. |
|
code |
Provider-specific error code, if available. |
|
raw |
Raw response payload, if available. |
Source code in apogee_ai_providers/domain/exceptions/provider_error.py
Bases: ProviderError
Source code in apogee_ai_providers/domain/exceptions/provider_rate_limit_error.py
Python
ProviderTimeoutError(message: str, *, provider: str, status: int | None = None, code: str | None = None, raw: Any = None)
Python
ProviderValidationError(message: str, *, provider: str, status: int | None = None, code: str | None = None, raw: Any = None)
Other · Protocols (ports)¶
Bases: Protocol
Async chat-completion contract. Any provider implements both methods.
async
¶
Python
complete(request: ChatRequest) -> ChatResponse
Python
stream(request: ChatRequest) -> AsyncIterator[ChatChunk]
Bases: Protocol
async
¶
Python
embed(request: EmbeddingRequest) -> EmbeddingResponse
async
¶
Bases: Protocol
async
¶
Python
generate(request: MultimodalRequest) -> MultimodalResponse
Bases: Protocol
async
¶
Python
transcribe(request: SpeechToTextRequest) -> SpeechToTextResponse
async
¶
Bases: Protocol
async
¶
Python
synthesize(request: TextToSpeechRequest) -> TextToSpeechResponse
async
¶
Other · Use cases¶
Python
ExecuteChatCompletionUseCase(provider: IChatCompletionProvider)
Source code in apogee_ai_providers/application/use_cases/execute_chat_completion_use_case.py
async
¶
Python
execute(request: ChatRequestDTO) -> ChatResponseDTO
Source code in apogee_ai_providers/application/use_cases/execute_chat_completion_use_case.py
Python
ExecuteMultimodalUseCase(provider: IMultimodalProvider)
Source code in apogee_ai_providers/application/use_cases/execute_multimodal_use_case.py
async
¶
Python
execute(request: MultimodalRequestDTO) -> MultimodalResponseDTO
Source code in apogee_ai_providers/application/use_cases/execute_multimodal_use_case.py
Python
StreamChatCompletionUseCase(provider: IChatCompletionProvider)
Source code in apogee_ai_providers/application/use_cases/stream_chat_completion_use_case.py
async
¶
Python
execute(request: ChatRequestDTO) -> AsyncIterator[ChatChunkDTO]