Add OVHcloud AI Endpoints provider (#3188)

Co-authored-by: Douwe Maan <[email protected]>

Author: eliasto

README.md

@@ -39,7 +39,7 @@ We built Pydantic AI with one simple aim: to bring that FastAPI feeling to GenAI
 [Pydantic Validation](https://docs.pydantic.dev/latest/) is the validation layer of the OpenAI SDK, the Google ADK, the Anthropic SDK, LangChain, LlamaIndex, AutoGPT, Transformers, CrewAI, Instructor and many more. _Why use the derivative when you can go straight to the source?_ :smiley:
 
 2. **Model-agnostic**:
-Supports virtually every [model](https://ai.pydantic.dev/models/overview) and provider: OpenAI, Anthropic, Gemini, DeepSeek, Grok, Cohere, Mistral, and Perplexity; Azure AI Foundry, Amazon Bedrock, Google Vertex AI, Ollama, LiteLLM, Groq, OpenRouter, Together AI, Fireworks AI, Cerebras, Hugging Face, GitHub, Heroku, Vercel, Nebius. If your favorite model or provider is not listed, you can easily implement a [custom model](https://ai.pydantic.dev/models/overview#custom-models).
+Supports virtually every [model](https://ai.pydantic.dev/models/overview) and provider: OpenAI, Anthropic, Gemini, DeepSeek, Grok, Cohere, Mistral, and Perplexity; Azure AI Foundry, Amazon Bedrock, Google Vertex AI, Ollama, LiteLLM, Groq, OpenRouter, Together AI, Fireworks AI, Cerebras, Hugging Face, GitHub, Heroku, Vercel, Nebius, OVHcloud. If your favorite model or provider is not listed, you can easily implement a [custom model](https://ai.pydantic.dev/models/overview#custom-models).
 
 3. **Seamless Observability**:
 Tightly [integrates](https://ai.pydantic.dev/logfire) with [Pydantic Logfire](https://pydantic.dev/logfire), our general-purpose OpenTelemetry observability platform, for real-time debugging, evals-based performance monitoring, and behavior, tracing, and cost tracking. If you already have an observability platform that supports OTel, you can [use that too](https://ai.pydantic.dev/logfire#alternative-observability-backends).

docs/api/providers.md

@@ -43,3 +43,5 @@
 ::: pydantic_ai.providers.litellm.LiteLLMProvider
 
 ::: pydantic_ai.providers.nebius.NebiusProvider
+
+::: pydantic_ai.providers.ovhcloud.OVHcloudProvider

docs/index.md

@@ -14,7 +14,7 @@ We built Pydantic AI with one simple aim: to bring that FastAPI feeling to GenAI
 [Pydantic Validation](https://docs.pydantic.dev/latest/) is the validation layer of the OpenAI SDK, the Google ADK, the Anthropic SDK, LangChain, LlamaIndex, AutoGPT, Transformers, CrewAI, Instructor and many more. _Why use the derivative when you can go straight to the source?_ :smiley:
 
 2. **Model-agnostic**:
-Supports virtually every [model](models/overview.md) and provider: OpenAI, Anthropic, Gemini, DeepSeek, Grok, Cohere, Mistral, and Perplexity; Azure AI Foundry, Amazon Bedrock, Google Vertex AI, Ollama, LiteLLM, Groq, OpenRouter, Together AI, Fireworks AI, Cerebras, Hugging Face, GitHub, Heroku, Vercel, Nebius. If your favorite model or provider is not listed, you can easily implement a [custom model](models/overview.md#custom-models).
+Supports virtually every [model](models/overview.md) and provider: OpenAI, Anthropic, Gemini, DeepSeek, Grok, Cohere, Mistral, and Perplexity; Azure AI Foundry, Amazon Bedrock, Google Vertex AI, Ollama, LiteLLM, Groq, OpenRouter, Together AI, Fireworks AI, Cerebras, Hugging Face, GitHub, Heroku, Vercel, Nebius, OVHcloud. If your favorite model or provider is not listed, you can easily implement a [custom model](models/overview.md#custom-models).
 
 3. **Seamless Observability**:
 Tightly [integrates](logfire.md) with [Pydantic Logfire](https://pydantic.dev/logfire), our general-purpose OpenTelemetry observability platform, for real-time debugging, evals-based performance monitoring, and behavior, tracing, and cost tracking. If you already have an observability platform that supports OTel, you can [use that too](logfire.md#alternative-observability-backends).

docs/models/openai.md

@@ -724,3 +724,37 @@ result = agent.run_sync('What is the capital of France?')
 print(result.output)
 #> The capital of France is Paris.
 ```
+
+### OVHcloud AI Endpoints
+
+To use OVHcloud AI Endpoints, you need to create a new API key. To do so, go to the [OVHcloud manager](https://ovh.com/manager), then in Public Cloud > AI Endpoints > API keys. Click on `Create a new API key` and copy your new key.
+
+You can explore the [catalog](https://endpoints.ai.cloud.ovh.net/catalog) to find which models are available.
+
+You can set the `OVHCLOUD_API_KEY` environment variable and use [`OVHcloudProvider`][pydantic_ai.providers.ovhcloud.OVHcloudProvider] by name:
+
+```python
+from pydantic_ai import Agent
+
+agent = Agent('ovhcloud:gpt-oss-120b')
+result = agent.run_sync('What is the capital of France?')
+print(result.output)
+#> The capital of France is Paris.
+```
+
+If you need to configure the provider, you can use the [`OVHcloudProvider`][pydantic_ai.providers.ovhcloud.OVHcloudProvider] class:
+
+```python
+from pydantic_ai import Agent
+from pydantic_ai.models.openai import OpenAIChatModel
+from pydantic_ai.providers.ovhcloud import OVHcloudProvider
+
+model = OpenAIChatModel(
+    'gpt-oss-120b',
+    provider=OVHcloudProvider(api_key='your-api-key'),
+)
+agent = Agent(model)
+result = agent.run_sync('What is the capital of France?')
+print(result.output)
+#> The capital of France is Paris.
+```

docs/models/overview.md

@@ -29,6 +29,7 @@ In addition, many providers are compatible with the OpenAI API, and can be used
 - [Cerebras](openai.md#cerebras)
 - [LiteLLM](openai.md#litellm)
 - [Nebius AI Studio](openai.md#nebius-ai-studio)
+- [OVHcloud AI Endpoints](openai.md#ovhcloud-ai-endpoints)
 
 Pydantic AI also comes with [`TestModel`](../api/models/test.md) and [`FunctionModel`](../api/models/function.md)
 for testing and development.

pydantic_ai_slim/pydantic_ai/models/__init__.py

@@ -688,6 +688,7 @@ def infer_model(model: Model | KnownModelName | str) -> Model:  # noqa: C901
         'vercel',
         'litellm',
         'nebius',
+        'ovhcloud',
     ):
         from .openai import OpenAIChatModel
 

pydantic_ai_slim/pydantic_ai/models/openai.py

@@ -285,6 +285,7 @@ def __init__(
             'vercel',
             'litellm',
             'nebius',
+            'ovhcloud',
         ]
         | Provider[AsyncOpenAI] = 'openai',
         profile: ModelProfileSpec | None = None,
@@ -314,6 +315,7 @@ def __init__(
             'vercel',
             'litellm',
             'nebius',
+            'ovhcloud',
         ]
         | Provider[AsyncOpenAI] = 'openai',
         profile: ModelProfileSpec | None = None,
@@ -342,6 +344,7 @@ def __init__(
             'vercel',
             'litellm',
             'nebius',
+            'ovhcloud',
         ]
         | Provider[AsyncOpenAI] = 'openai',
         profile: ModelProfileSpec | None = None,
@@ -903,7 +906,9 @@ def __init__(
         self,
         model_name: OpenAIModelName,
         *,
-        provider: Literal['openai', 'deepseek', 'azure', 'openrouter', 'grok', 'fireworks', 'together', 'nebius']
+        provider: Literal[
+            'openai', 'deepseek', 'azure', 'openrouter', 'grok', 'fireworks', 'together', 'nebius', 'ovhcloud'
+        ]
         | Provider[AsyncOpenAI] = 'openai',
         profile: ModelProfileSpec | None = None,
         settings: ModelSettings | None = None,

pydantic_ai_slim/pydantic_ai/providers/__init__.py

@@ -146,6 +146,10 @@ def infer_provider_class(provider: str) -> type[Provider[Any]]:  # noqa: C901
         from .nebius import NebiusProvider
 
         return NebiusProvider
+    elif provider == 'ovhcloud':
+        from .ovhcloud import OVHcloudProvider
+
+        return OVHcloudProvider
     else:  # pragma: no cover
         raise ValueError(f'Unknown provider: {provider}')
 

pydantic_ai_slim/pydantic_ai/providers/ovhcloud.py

@@ -0,0 +1,95 @@
+from __future__ import annotations as _annotations
+
+import os
+from typing import overload
+
+import httpx
+
+from pydantic_ai import ModelProfile
+from pydantic_ai.exceptions import UserError
+from pydantic_ai.models import cached_async_http_client
+from pydantic_ai.profiles.deepseek import deepseek_model_profile
+from pydantic_ai.profiles.harmony import harmony_model_profile
+from pydantic_ai.profiles.meta import meta_model_profile
+from pydantic_ai.profiles.mistral import mistral_model_profile
+from pydantic_ai.profiles.openai import OpenAIJsonSchemaTransformer, OpenAIModelProfile
+from pydantic_ai.profiles.qwen import qwen_model_profile
+from pydantic_ai.providers import Provider
+
+try:
+    from openai import AsyncOpenAI
+except ImportError as _import_error:  # pragma: no cover
+    raise ImportError(
+        'Please install the `openai` package to use OVHcloud AI Endpoints provider.'
+        'You can use the `openai` optional group — `pip install "pydantic-ai-slim[openai]"`'
+    ) from _import_error
+
+
+class OVHcloudProvider(Provider[AsyncOpenAI]):
+    """Provider for OVHcloud AI Endpoints."""
+
+    @property
+    def name(self) -> str:
+        return 'ovhcloud'
+
+    @property
+    def base_url(self) -> str:
+        return 'https://oai.endpoints.kepler.ai.cloud.ovh.net/v1'
+
+    @property
+    def client(self) -> AsyncOpenAI:
+        return self._client
+
+    def model_profile(self, model_name: str) -> ModelProfile | None:
+        model_name = model_name.lower()
+
+        prefix_to_profile = {
+            'llama': meta_model_profile,
+            'meta-': meta_model_profile,
+            'deepseek': deepseek_model_profile,
+            'mistral': mistral_model_profile,
+            'gpt': harmony_model_profile,
+            'qwen': qwen_model_profile,
+        }
+
+        profile = None
+        for prefix, profile_func in prefix_to_profile.items():
+            if model_name.startswith(prefix):
+                profile = profile_func(model_name)
+
+        # As the OVHcloud AI Endpoints API is OpenAI-compatible, let's assume we also need OpenAIJsonSchemaTransformer.
+        return OpenAIModelProfile(json_schema_transformer=OpenAIJsonSchemaTransformer).update(profile)
+
+    @overload
+    def __init__(self) -> None: ...
+
+    @overload
+    def __init__(self, *, api_key: str) -> None: ...
+
+    @overload
+    def __init__(self, *, api_key: str, http_client: httpx.AsyncClient) -> None: ...
+
+    @overload
+    def __init__(self, *, openai_client: AsyncOpenAI | None = None) -> None: ...
+
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        openai_client: AsyncOpenAI | None = None,
+        http_client: httpx.AsyncClient | None = None,
+    ) -> None:
+        api_key = api_key or os.getenv('OVHCLOUD_API_KEY')
+        if not api_key and openai_client is None:
+            raise UserError(
+                'Set the `OVHCLOUD_API_KEY` environment variable or pass it via '
+                '`OVHcloudProvider(api_key=...)` to use OVHcloud AI Endpoints provider.'
+            )
+
+        if openai_client is not None:
+            self._client = openai_client
+        elif http_client is not None:
+            self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=http_client)
+        else:
+            http_client = cached_async_http_client(provider='ovhcloud')
+            self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=http_client)

tests/providers/test_ovhcloud.py

@@ -0,0 +1,109 @@
+import re
+
+import httpx
+import pytest
+from pytest_mock import MockerFixture
+
+from pydantic_ai._json_schema import InlineDefsJsonSchemaTransformer
+from pydantic_ai.exceptions import UserError
+from pydantic_ai.profiles.deepseek import deepseek_model_profile
+from pydantic_ai.profiles.harmony import harmony_model_profile
+from pydantic_ai.profiles.meta import meta_model_profile
+from pydantic_ai.profiles.mistral import mistral_model_profile
+from pydantic_ai.profiles.openai import OpenAIJsonSchemaTransformer
+from pydantic_ai.profiles.qwen import qwen_model_profile
+
+from ..conftest import TestEnv, try_import
+
+with try_import() as imports_successful:
+    import openai
+
+    from pydantic_ai.providers.ovhcloud import OVHcloudProvider
+
+
+pytestmark = [
+    pytest.mark.skipif(not imports_successful(), reason='openai not installed'),
+    pytest.mark.vcr,
+    pytest.mark.anyio,
+]
+
+
+def test_ovhcloud_provider():
+    provider = OVHcloudProvider(api_key='your-api-key')
+    assert provider.name == 'ovhcloud'
+    assert provider.base_url == 'https://oai.endpoints.kepler.ai.cloud.ovh.net/v1'
+    assert isinstance(provider.client, openai.AsyncOpenAI)
+    assert provider.client.api_key == 'your-api-key'
+
+
+def test_ovhcloud_provider_need_api_key(env: TestEnv) -> None:
+    env.remove('OVHCLOUD_API_KEY')
+    with pytest.raises(
+        UserError,
+        match=re.escape(
+            'Set the `OVHCLOUD_API_KEY` environment variable or pass it via '
+            '`OVHcloudProvider(api_key=...)` to use OVHcloud AI Endpoints provider.'
+        ),
+    ):
+        OVHcloudProvider()
+
+
+def test_ovhcloud_pass_openai_client() -> None:
+    openai_client = openai.AsyncOpenAI(api_key='your-api-key')
+    provider = OVHcloudProvider(openai_client=openai_client)
+    assert provider.client == openai_client
+
+
+def test_ovhcloud_pass_http_client():
+    http_client = httpx.AsyncClient()
+    provider = OVHcloudProvider(api_key='your-api-key', http_client=http_client)
+    assert isinstance(provider.client, openai.AsyncOpenAI)
+    assert provider.client.api_key == 'your-api-key'
+
+
+def test_ovhcloud_model_profile(mocker: MockerFixture):
+    provider = OVHcloudProvider(api_key='your-api-key')
+
+    ns = 'pydantic_ai.providers.ovhcloud'
+
+    # Mock all profile functions
+    deepseek_mock = mocker.patch(f'{ns}.deepseek_model_profile', wraps=deepseek_model_profile)
+    harmony_mock = mocker.patch(f'{ns}.harmony_model_profile', wraps=harmony_model_profile)
+    meta_mock = mocker.patch(f'{ns}.meta_model_profile', wraps=meta_model_profile)
+    mistral_mock = mocker.patch(f'{ns}.mistral_model_profile', wraps=mistral_model_profile)
+    qwen_mock = mocker.patch(f'{ns}.qwen_model_profile', wraps=qwen_model_profile)
+
+    # Test deepseek provider
+    profile = provider.model_profile('DeepSeek-R1-Distill-Llama-70B')
+    deepseek_mock.assert_called_with('deepseek-r1-distill-llama-70b')
+    assert profile is not None
+    assert profile.json_schema_transformer == OpenAIJsonSchemaTransformer
+
+    # Test harmony (for openai gpt-oss) provider
+    profile = provider.model_profile('gpt-oss-120b')
+    harmony_mock.assert_called_with('gpt-oss-120b')
+    assert profile is not None
+    assert profile.json_schema_transformer == OpenAIJsonSchemaTransformer
+
+    # Test meta provider
+    meta_profile = provider.model_profile('Llama-3.3-70B-Instruct')
+    meta_mock.assert_called_with('llama-3.3-70b-instruct')
+    assert meta_profile is not None
+    assert meta_profile.json_schema_transformer == InlineDefsJsonSchemaTransformer
+
+    # Test mistral provider
+    profile = provider.model_profile('Mistral-Small-3.2-24B-Instruct-2506')
+    mistral_mock.assert_called_with('mistral-small-3.2-24b-instruct-2506')
+    assert profile is not None
+    assert profile.json_schema_transformer == OpenAIJsonSchemaTransformer
+
+    # Test qwen provider
+    qwen_profile = provider.model_profile('Qwen3-32B')
+    qwen_mock.assert_called_with('qwen3-32b')
+    assert qwen_profile is not None
+    assert qwen_profile.json_schema_transformer == InlineDefsJsonSchemaTransformer
+
+    # Test unknown provider
+    unknown_profile = provider.model_profile('unknown-model')
+    assert unknown_profile is not None
+    assert unknown_profile.json_schema_transformer == OpenAIJsonSchemaTransformer