oats.provider

AI provider configuration and model management.

oats.provider.models

Model definitions and registry.

class oats.provider.models.Model(id, provider_id, name, description='', context_length=262100, max_output_tokens=262100, supports_tools=True, supports_vision=False, supports_streaming=True, cost_per_input_token=0.0, cost_per_output_token=0.0, extra=<factory>)[source]

Bases: object

Definition of an AI model.

id: Unique model identifier (e.g. “gpt-4o”).

provider_id: The provider this model belongs to (e.g. “openai”, “anthropic”).

name: Human-readable display name.

description: Short description of the model’s capabilities.

context_length: Maximum number of tokens in the input context window.

max_output_tokens: Maximum number of tokens the model can generate in a single response.

supports_tools: Whether the model supports function/tool calling.

supports_vision: Whether the model supports image/vision input.

supports_streaming: Whether the model supports streaming responses.

cost_per_input_token: Cost per input token in USD.

cost_per_output_token: Cost per output token in USD.

extra: Additional provider-specific metadata.

id: str

provider_id: str

name: str

description: str = ''

context_length: int = 262100

max_output_tokens: int = 262100

supports_tools: bool = True

supports_vision: bool = False

supports_streaming: bool = True

cost_per_input_token: float = 0.0

cost_per_output_token: float = 0.0

extra: dict[str, Any]

property litellm_model: str

Get the LiteLLM model identifier.

Maps the internal provider/model combination to the format expected by LiteLLM (e.g. “azure/gpt-4o”, “gemini/gemini-2.0-flash”).

Returns:: The model string to pass to LiteLLM’s acompletion().

__init__(id, provider_id, name, description='', context_length=262100, max_output_tokens=262100, supports_tools=True, supports_vision=False, supports_streaming=True, cost_per_input_token=0.0, cost_per_output_token=0.0, extra=<factory>)

class oats.provider.models.ModelRegistry[source]

Bases: object

Registry of available models.

Maintains a central index of all registered AI models, keyed by provider_id/model_id. Initialized with built-in model definitions on construction.

__init__()[source]: Initialize the registry and load built-in models.

register(model)[source]

Register a model in the registry.

Return type:: None
Parameters:: model – The model definition to register.

get(provider_id, model_id)[source]

Get a model by provider and model ID.

Return type:

Model | None

Parameters:

provider_id – The provider identifier (e.g. “openai”).
model_id – The model identifier (e.g. “gpt-4o”).

Returns:

The Model if found, or None.

list(provider_id=None)[source]

List all models, optionally filtered by provider.

Return type:: list[Model]
Parameters:: provider_id – If provided, only return models from this provider.
Returns:: A list of Model instances.

list_by_provider()[source]

List models grouped by provider.

Return type:: dict[str, list[Model]]
Returns:: A dict mapping provider_id to a list of Model instances.

oats.provider.models.get_model_registry()[source]

Get the global model registry.

Lazily initializes a singleton ModelRegistry on first call.

Return type:: ModelRegistry
Returns:: The global ModelRegistry instance.

oats.provider.models.get_model(provider_id, model_id)[source]

Get a model by provider and model ID.

Return type:

Model | None

Parameters:

provider_id – The provider identifier (e.g. “openai”).
model_id – The model identifier (e.g. “gpt-4o”).

Returns:

The Model if found, or None.

oats.provider.models.list_models(provider_id=None)[source]

List all available models.

Return type:: list[Model]
Parameters:: provider_id – If provided, only return models from this provider.
Returns:: A list of all registered Model instances.

oats.provider.provider

AI Provider abstraction using LiteLLM.

Provides: - Multi-provider LLM calls via LiteLLM - Retry with exponential backoff and jitter for transient errors - Text-based tool call parsing for open-source models (Qwen, Hermes) - Streaming support

class oats.provider.provider.Message(**data)[source]

Bases: BaseModel

A chat message in a conversation.

role: The role of the message sender (“system”, “user”, “assistant”, “tool”).

content: The message content as a string or a list of content blocks (for multimodal).

name: Optional name for the message sender.

tool_call_id: ID of the tool call this message is responding to.

tool_calls: List of tool calls made by the assistant.

role: str

content: str | list[dict[str, Any]]

name: str | None

tool_call_id: str | None

tool_calls: list[dict[str, Any]] | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.provider.provider.ToolDefinition(**data)[source]

Bases: BaseModel

Definition of a tool for the LLM.

name: The tool’s function name.

description: Human-readable description of what the tool does.

parameters: JSON Schema describing the tool’s input parameters.

strict: Whether to enforce strict schema validation.

name: str

description: str

parameters: dict[str, Any]

strict: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.provider.provider.CompletionRequest(**data)[source]

Bases: BaseModel

Request for a completion from an LLM provider.

messages: The conversation history as a list of Message objects.

model: Optional model ID override. Falls back to config default.

provider_id: Optional provider ID override.

tools: Optional list of tool definitions for function calling.

temperature: Sampling temperature for generation.

max_tokens: Maximum number of tokens to generate.

top_p: Nucleus sampling threshold.

stop: List of stop sequences to halt generation.

stream: Whether to stream the response.

debug_context: Optional metadata for debugging and tracing.

messages: list[Message]

model: str | None

provider_id: str | None

tools: list[ToolDefinition] | None

temperature: float | None

max_tokens: int | None

top_p: float | None

stop: list[str] | None

stream: bool

debug_context: dict[str, Any] | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.provider.provider.ToolCall(**data)[source]

Bases: BaseModel

A tool call from the LLM.

id: Unique identifier for this tool call.

name: The name of the tool/function to invoke.

arguments: JSON string of the tool’s input arguments.

id: str

name: str

arguments: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.provider.provider.CompletionResponse(**data)[source]

Bases: BaseModel

Response from a non-streamed completion.

content: The text content of the response, or None if tool calls were made.

tool_calls: List of tool calls requested by the model.

finish_reason: Reason the model stopped generating (e.g. “stop”, “tool_calls”).

usage: Token usage statistics (prompt, completion, total).

model: The model that generated the response.

content: str | None

tool_calls: list[ToolCall] | None

finish_reason: str | None

usage: dict[str, int] | None

model: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.provider.provider.CompletionChunk(**data)[source]

Bases: BaseModel

A streaming chunk from a completion.

content: A fragment of text content, or None.

tool_calls: Tool calls assembled from the stream, if any.

finish_reason: Reason the stream ended, if this is the final chunk.

content: str | None

tool_calls: list[ToolCall] | None

finish_reason: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.provider.provider.Provider(id, name, config, models=<factory>)[source]

Bases: object

An AI provider (Anthropic, OpenAI, etc.).

Wraps LiteLLM to provide a unified interface for calling LLMs across multiple providers. Handles authentication, model resolution, retry logic, and tool call parsing.

id: Unique provider identifier (e.g. “openai”, “anthropic”).

name: Human-readable provider name.

config: Provider-specific configuration (API keys, base URLs, etc.).

models: List of models available from this provider.

id: str

name: str

config: ProviderConfig

models: list[Model]

is_configured()[source]

Check if the provider has required configuration.

For OpenWebUI providers, attempts login and retrieves an API token. For Ollama, always returns True (no API key required). For all other providers, checks that an API key is set.

Return type:: bool
Returns:: True if the provider is ready to make API calls.

async complete(request, verbose=True)[source]

Get a completion from the provider with retry logic.

Retries transient errors (rate limits, 5xx, timeouts) with exponential backoff + jitter. Publishes request and response events to the event bus for observability.

Return type:

CompletionResponse

Parameters:

request – The completion request containing messages, model, tools, etc.
verbose – If True, log the request details.

Returns:

A CompletionResponse with content, tool calls, and usage stats.

Raises:

Exception – If all retry attempts are exhausted.

async stream(request, verbose=False)[source]

Stream a completion from the provider with retry logic.

Yields chunks as they arrive from the LLM. For tool calls that arrive as text content (open-source models), the final chunk will contain the parsed tool calls. Publishes request and response events to the event bus.

Return type:

AsyncIterator[CompletionChunk]

Parameters:

request – The completion request (stream=True is enforced).
verbose – If True, log the request details.

Yields:

CompletionChunk objects containing content fragments, tool calls, and finish reasons.

Raises:

Exception – If all retry attempts are exhausted.

__init__(id, name, config, models=<factory>)

class oats.provider.provider.ProviderRegistry[source]

Bases: object

Registry of available AI providers.

Maintains a central index of all registered providers, keyed by provider ID. Initialized with built-in provider definitions on first access via get_provider_registry().

__init__()[source]: Initialize an empty provider registry.

register(provider)[source]

Register a provider in the registry.

Return type:: None
Parameters:: provider – The provider to register.

get(provider_id)[source]

Get a provider by ID.

Return type:: Provider | None
Parameters:: provider_id – The provider identifier (e.g. “openai”).
Returns:: The Provider if found, or None.

list()[source]

List all registered providers.

Return type:: list[Provider]
Returns:: A list of all Provider instances.

list_configured()[source]

List only configured providers.

Return type:: list[Provider]
Returns:: A list of Provider instances that have valid configuration.

oats.provider.provider.get_provider_registry()[source]

Get the global provider registry, initializing from config if needed.

Lazily initializes a singleton ProviderRegistry on first call, populating it with built-in provider definitions from the config.

Return type:: ProviderRegistry
Returns:: The global ProviderRegistry instance.

oats.provider.provider.get_provider(provider_id=None)[source]

Get a provider by ID, or the default provider.

If provider_id is None, uses the default provider from the config.

Return type:: Provider
Parameters:: provider_id – The provider identifier. If None, uses the config default.
Returns:: The configured Provider instance.
Raises:: ValueError – If the provider is not found or not configured.

oats.provider.provider.list_providers()[source]

List all available providers.

Return type:: list[Provider]
Returns:: A list of all registered Provider instances.