fix: use Portkey native Python SDK

pyproject.toml

@@ -131,7 +131,7 @@ nebius = []
 neosantara = []
 openai = []
 openrouter = []
-portkey = []
+portkey = ["portkey-ai"]
 qiniu = []
 requesty = []
 sambanova = []

src/any_llm/providers/portkey/portkey.py

@@ -12,6 +12,16 @@
 from any_llm.types.completion import ChatCompletion, ChatCompletionChunk, CompletionParams
 from any_llm.utils.structured_output import get_json_schema, is_structured_output_type
 
+try:
+    import portkey_ai  # type: ignore[import-not-found]
+
+    AsyncPortkey: Any = portkey_ai.AsyncPortkey
+except ImportError as e:  # pragma: no cover - exercised by the package guard
+    AsyncPortkey = None
+    MISSING_PACKAGES_ERROR: ImportError | None = e
+else:
+    MISSING_PACKAGES_ERROR = None
+
 
 class PortkeyProvider(XMLReasoningOpenAIProvider):
     """Portkey provider for accessing 200+ LLMs through Portkey's AI Gateway."""
@@ -30,6 +40,17 @@ class PortkeyProvider(XMLReasoningOpenAIProvider):
     SUPPORTS_LIST_MODELS = True
 
     _DEFAULT_REASONING_EFFORT = None
+    MISSING_PACKAGES_ERROR = MISSING_PACKAGES_ERROR
+    client: Any
+
+    @override
+    def _init_client(self, api_key: str | None = None, api_base: str | None = None, **kwargs: Any) -> None:
+        """Initialize Portkey's native async client."""
+        self.client = AsyncPortkey(
+            api_key=api_key,
+            base_url=api_base or self.API_BASE,
+            **kwargs,
+        )
 
     @staticmethod
     @override

tests/unit/providers/test_portkey_provider.py

@@ -1,9 +1,26 @@
 from dataclasses import dataclass
+from typing import Any
+
+import pytest
 
 from any_llm.providers.portkey.portkey import PortkeyProvider
 from any_llm.types.completion import CompletionParams
 
 
+def test_portkey_uses_native_async_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The provider initializes Portkey's native SDK instead of OpenAI's client."""
+    calls: dict[str, Any] = {}
+
+    class FakeAsyncPortkey:
+        def __init__(self, **kwargs: Any) -> None:
+            calls.update(kwargs)
+
+    monkeypatch.setattr("any_llm.providers.portkey.portkey.AsyncPortkey", FakeAsyncPortkey)
+    PortkeyProvider(api_key="test-key", api_base="https://example.test/v1")
+
+    assert calls == {"api_key": "test-key", "base_url": "https://example.test/v1"}
+
+
 def test_convert_completion_params_with_dataclass_response_format() -> None:
     """Test that dataclass response_format is converted to JSON schema format."""