Providers¶

Base Provider¶

Base provider interface.

All provider adapters must implement this interface.

class agent.providers.base.BaseProvider(api_key=None, base_url=None, timeout=120.0, max_retries=2, **kwargs)[source]¶

Bases: ABC

Base class for all provider adapters.

Each provider must implement the core methods to handle request conversion, response normalization, and streaming.

Parameters:

api_key (str | None)
base_url (str | None)
timeout (float)
max_retries (int)
kwargs (Any)

name: str = 'base'¶

capabilities: ProviderCapabilities = ProviderCapabilities(streaming=True, tools=True, structured_output=True, json_mode=True, vision=False, system_messages=True, batch=False, native_schema_output=False, max_context_tokens=None, max_output_tokens=None)¶

abstractmethod run(request)[source]¶

Execute a synchronous request.

Parameters:: request (AgentRequest) – Normalized agent request
Return type:: AgentResponse
Returns:: Normalized agent response

abstractmethod async run_async(request)[source]¶

Execute an asynchronous request.

Parameters:: request (AgentRequest) – Normalized agent request
Return type:: AgentResponse
Returns:: Normalized agent response

abstractmethod stream(request)[source]¶

Execute a streaming request.

Parameters:: request (AgentRequest) – Normalized agent request
Yields:: Normalized stream events
Return type:: Iterator[StreamEvent]

abstractmethod stream_async(request)[source]¶

Execute an asynchronous streaming request.

Subclasses implement this as an async generator (async def with yield). Callers should iterate with async for, not await.

Parameters:: request (AgentRequest) – Normalized agent request
Yields:: Normalized stream events
Return type:: AsyncIterator[StreamEvent]

supports_tools()[source]¶

Check if provider supports tool calling.

Return type:: bool

supports_structured_output()[source]¶

Check if provider supports structured output.

Return type:: bool

supports_vision()[source]¶

Check if provider supports vision/images.

Return type:: bool

supports_streaming()[source]¶

Check if provider supports streaming.

Return type:: bool

supports_json_mode()[source]¶

Check if provider supports JSON mode.

Return type:: bool

supports_native_schema()[source]¶

Check if provider supports native schema-enforced output.

Return type:: bool

validate_config()[source]¶

Validate provider configuration.

Return type:: list[str]
Returns:: List of validation error messages (empty if valid)

Provider Registry¶

Provider registry.

Manages registration and instantiation of provider adapters.

class agent.providers.registry.ProviderRegistry[source]¶

Bases: object

Registry for provider adapters.

classmethod register(name, provider_class, aliases=None)[source]¶

Parameters:

name (str) – Provider name (e.g., “openai”)
provider_class (type[BaseProvider]) – Provider class
aliases (list[str] | None) – Optional list of aliases

Return type:

None

classmethod get_class(name)[source]¶

Get a provider class by name.

Parameters:: name (str) – Provider name or alias
Return type:: type[BaseProvider]
Returns:: Provider class
Raises:: ProviderError – If provider not found

classmethod create(name, **kwargs)[source]¶

Create a provider instance.

Parameters:

name (str) – Provider name or alias
**kwargs (Any) – Provider configuration

Return type:

BaseProvider

Returns:

Provider instance

classmethod list_providers()[source]¶

List all registered provider names.

Return type:: list[str]

classmethod is_registered(name)[source]¶

Check if a provider is registered.

Parameters:: name (str)
Return type:: bool

agent.providers.registry.get_provider(name, **kwargs)[source]¶

Get a provider instance.

This is the main entry point for getting providers. Automatically loads provider modules on first access.

Parameters:

name (str) – Provider name (e.g., “openai”, “anthropic”)
**kwargs (Any) – Provider configuration

Return type:

BaseProvider

Returns:

Provider instance

OpenAI¶

OpenAI provider adapter.

Supports OpenAI API and compatible endpoints.

class agent.providers.openai.OpenAIProvider(api_key=None, base_url=None, timeout=120.0, max_retries=2, **kwargs)[source]¶

Bases: BaseProvider

OpenAI provider adapter.

Supports GPT-4, GPT-4o, GPT-3.5-turbo and other OpenAI models. Also works with OpenAI-compatible APIs.

Parameters:

api_key (str | None)
base_url (str | None)
timeout (float)
max_retries (int)
kwargs (Any)

name: str = 'openai'¶

capabilities: ProviderCapabilities = ProviderCapabilities(streaming=True, tools=True, structured_output=True, json_mode=True, vision=True, system_messages=True, batch=True, native_schema_output=True, max_context_tokens=None, max_output_tokens=None)¶

run(request)[source]¶

Execute a synchronous request.

Parameters:: request (AgentRequest)
Return type:: AgentResponse

async run_async(request)[source]¶

Execute an asynchronous request.

Parameters:: request (AgentRequest)
Return type:: AgentResponse

stream(request)[source]¶

Execute a streaming request.

Parameters:: request (AgentRequest)
Return type:: Iterator[StreamEvent]

async stream_async(request)[source]¶

Execute an async streaming request.

Parameters:: request (AgentRequest)
Return type:: AsyncIterator[StreamEvent]

Anthropic¶

Gemini¶

Google Gemini provider adapter.

class agent.providers.gemini.GeminiProvider(api_key=None, base_url=None, timeout=120.0, max_retries=2, **kwargs)[source]¶

Bases: BaseProvider

Google Gemini provider adapter.

Supports Gemini Pro, Gemini Flash, and other Gemini models.

Parameters:

api_key (str | None)
base_url (str | None)
timeout (float)
max_retries (int)
kwargs (Any)

name: str = 'gemini'¶

capabilities: ProviderCapabilities = ProviderCapabilities(streaming=True, tools=True, structured_output=True, json_mode=True, vision=True, system_messages=True, batch=False, native_schema_output=True, max_context_tokens=None, max_output_tokens=None)¶

run(request)[source]¶

Execute a synchronous request.

Parameters:: request (AgentRequest)
Return type:: AgentResponse

async run_async(request)[source]¶

Execute an asynchronous request.

Parameters:: request (AgentRequest)
Return type:: AgentResponse

stream(request)[source]¶

Execute a streaming request.

Parameters:: request (AgentRequest)
Return type:: Iterator[StreamEvent]

async stream_async(request)[source]¶

Execute an async streaming request.

Parameters:: request (AgentRequest)
Return type:: AsyncIterator[StreamEvent]

DeepSeek¶

DeepSeek provider adapter.

Uses OpenAI-compatible API.

class agent.providers.deepseek.DeepSeekProvider(api_key=None, base_url=None, timeout=120.0, max_retries=2, **kwargs)[source]¶

Bases: BaseProvider

DeepSeek provider adapter.

Uses the OpenAI-compatible API with DeepSeek’s endpoint.

Parameters:

api_key (str | None)
base_url (str | None)
timeout (float)
max_retries (int)
kwargs (Any)

name: str = 'deepseek'¶

capabilities: ProviderCapabilities = ProviderCapabilities(streaming=True, tools=True, structured_output=True, json_mode=True, vision=False, system_messages=True, batch=False, native_schema_output=False, max_context_tokens=None, max_output_tokens=None)¶

DEEPSEEK_BASE_URL = 'https://api.deepseek.com/v1'¶

run(request)[source]¶

Execute a synchronous request.

Parameters:: request (AgentRequest)
Return type:: AgentResponse

async run_async(request)[source]¶

Execute an asynchronous request.

Parameters:: request (AgentRequest)
Return type:: AgentResponse

stream(request)[source]¶

Execute a streaming request.

Parameters:: request (AgentRequest)
Return type:: Iterator[StreamEvent]

async stream_async(request)[source]¶

Execute an async streaming request.

Parameters:: request (AgentRequest)
Return type:: AsyncIterator[StreamEvent]