docs
Python client library for the Portkey API - Control Panel for AI Apps with unified API signature, automated fallbacks, retries, load balancing, semantic caching, virtual keys, and comprehensive observability features.
npx @tessl/cli install tessl/pypi-portkey-ai@1.14.0Python client library for the Portkey API - Control Panel for AI Apps. Built on top of the OpenAI SDK, allowing seamless integration of Portkey's advanced features including AI Gateway capabilities (unified API signature, automated fallbacks & retries, load balancing, semantic caching, virtual keys), and comprehensive observability features (logging, request tracing, custom metadata, feedback collection, analytics).
pip install portkey-aifrom portkey_ai import Portkey, AsyncPortkeyFor OpenAI compatibility:
from portkey_ai import openaiFor specific components:
from portkey_ai import (
Config,
LLMOptions,
Modes,
ProviderTypes,
createHeaders
)from portkey_ai import Portkey
# Initialize client with API key and virtual key
portkey = Portkey(
api_key="PORTKEY_API_KEY",
virtual_key="VIRTUAL_KEY"
)
# Make a chat completion request
chat_completion = portkey.chat.completions.create(
messages=[{"role": "user", "content": "Say this is a test"}],
model="gpt-4"
)
print(chat_completion)import asyncio
from portkey_ai import AsyncPortkey
async def main():
portkey = AsyncPortkey(
api_key="PORTKEY_API_KEY",
virtual_key="VIRTUAL_KEY"
)
chat_completion = await portkey.chat.completions.create(
messages=[{"role": "user", "content": "Say this is a test"}],
model="gpt-4"
)
print(chat_completion)
asyncio.run(main())Portkey's architecture centers around two main client classes that provide comprehensive AI API management:
The dual client pattern ensures both synchronous and asynchronous support across all operations, while maintaining full OpenAI SDK compatibility.
Primary client classes and configuration utilities for initializing and managing Portkey connections with comprehensive provider support and advanced routing capabilities.
class Portkey:
def __init__(
self,
*,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
virtual_key: Optional[str] = None,
config: Optional[Union[Mapping, str]] = None,
provider: Optional[str] = None,
trace_id: Optional[str] = None,
metadata: Union[Optional[dict[str, str]], str] = None,
**kwargs
) -> None: ...
class AsyncPortkey:
def __init__(self, **kwargs) -> None: ...
def createHeaders(**kwargs) -> dict: ...OpenAI-compatible chat completion API with support for all major providers, streaming, function calling, and advanced Portkey features like fallbacks and load balancing.
class ChatCompletion:
completions: Completions
class Completions:
def create(
self,
*,
messages: Iterable[dict],
model: Optional[str] = "portkey-default",
**kwargs
) -> Union[ChatCompletions, Iterator[ChatCompletionChunk]]: ...
messages: ChatCompletionsMessages
class AsyncChatCompletion:
completions: AsyncCompletionsLegacy text completion API for non-chat models, supporting streaming and all OpenAI-compatible parameters with Portkey enhancements.
class Completion:
def create(
self,
*,
prompt: Union[str, List[str]],
model: str,
**kwargs
) -> Union[TextCompletion, Iterator[TextCompletionChunk]]: ...
class AsyncCompletion:
async def create(
self,
*,
prompt: Union[str, List[str]],
model: str,
**kwargs
) -> Union[TextCompletion, AsyncIterator[TextCompletionChunk]]: ...Advanced prompt management system with templating, rendering, versioning, and prompt execution capabilities.
class Prompts:
def render(self, **kwargs): ...
def completions(self, **kwargs): ...
class AsyncPrompts:
async def render(self, **kwargs): ...
async def completions(self, **kwargs): ...
class Generations:
def create(self, **kwargs): ...
class AsyncGenerations:
async def create(self, **kwargs): ...Text embedding generation for semantic similarity, search, and machine learning applications with support for multiple providers.
class Embeddings:
def create(
self,
*,
input: str,
model: Optional[str] = "portkey-default",
dimensions: Union[int, NotGiven] = NOT_GIVEN,
encoding_format: Union[str, NotGiven] = NOT_GIVEN,
user: Union[str, NotGiven] = NOT_GIVEN,
**kwargs
) -> CreateEmbeddingResponse: ...
class AsyncEmbeddings:
async def create(self, **kwargs) -> CreateEmbeddingResponse: ...Model management and information retrieval for accessing available AI models across different providers.
class Models:
def list(self, **kwargs) -> ModelList: ...
def retrieve(
self,
model: str,
*,
timeout: Union[float, NotGiven] = NOT_GIVEN,
**kwargs
) -> Model: ...
def delete(
self,
model: str,
*,
timeout: Union[float, NotGiven] = NOT_GIVEN,
**kwargs
) -> ModelDeleted: ...
class AsyncModels:
async def list(self, **kwargs) -> ModelList: ...
async def retrieve(self, model: str, **kwargs) -> Model: ...
async def delete(self, model: str, **kwargs) -> ModelDeleted: ...Image generation, audio processing (speech-to-text, text-to-speech, translation), and content moderation capabilities with OpenAI compatibility.
class Images:
def generate(self, **kwargs): ...
def create_variation(self, **kwargs): ...
def edit(self, **kwargs): ...
class Audio:
transcriptions: Transcriptions
translations: Translations
speech: Speech
class Moderations:
def create(self, **kwargs): ...File upload, management, and processing capabilities including support for assistants, fine-tuning, and batch operations.
class MainFiles:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def delete(self, **kwargs): ...
class Uploads:
def create(self, **kwargs): ...
parts: PartsOpenAI Assistants API implementation with thread management, message handling, and tool execution support.
class Assistants:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...
class Threads:
def create(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...
messages: Messages
runs: RunsVector database operations for retrieval-augmented generation (RAG) applications with file management and batch processing.
class VectorStores:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...
files: VectorFiles
file_batches: VectorFileBatchesModel fine-tuning capabilities with job management, checkpoint handling, and training monitoring.
class FineTuning:
jobs: Jobs
checkpoints: Checkpoints
class Jobs:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def cancel(self, **kwargs): ...Batch API for processing large volumes of requests efficiently with cost optimization and result management.
class Batches:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def cancel(self, **kwargs): ...Administrative functions for managing users, workspaces, invitations, and organizational settings.
class Admin:
users: Users
invites: Invites
workspaces: Workspaces
class Users:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...Administration & User Management
Advanced configuration system for managing provider settings, routing rules, fallback strategies, and load balancing configurations.
class Configs:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...Secure API key and virtual key management for credential protection and access control.
class ApiKeys:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...
class VirtualKeys:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...API Key & Virtual Key Management
Comprehensive logging, request tracing, analytics, and monitoring capabilities with custom metadata and performance metrics.
class Logs:
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
class Responses:
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
input_items: InputItems
output_items: OutputItems
class Labels:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...User feedback collection and data organization capabilities for improving AI applications and gathering insights.
class Feedback:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
class Collections:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...AI evaluation framework with automated grading, test execution, and performance measurement capabilities.
class Evals:
runs: EvalsRuns
class EvalsRuns:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
class Alpha:
graders: Graders
class Graders:
def create(self, **kwargs): ...
def list(self, **kwargs): ...Container-based deployment and content management capabilities for scalable AI applications.
class Containers:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
def update(self, **kwargs): ...
def delete(self, **kwargs): ...
files: ContainersFiles
class Content:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...Container & Content Management
Multi-part upload system for large files with chunked upload support and integrity verification.
class Uploads:
def create(
self,
*,
bytes: int,
filename: str,
mime_type: str,
purpose: Any,
**kwargs
) -> Upload: ...
def upload_file_chunked(
self,
*,
file: Union[os.PathLike[str], bytes],
mime_type: str,
purpose: Any,
**kwargs
) -> Any: ...
def complete(
self,
upload_id: str,
*,
part_ids: List[str],
**kwargs
) -> Upload: ...
parts: Parts
class AsyncUploads:
async def create(self, **kwargs) -> Upload: ...
async def upload_file_chunked(self, **kwargs) -> Any: ...
async def complete(self, upload_id: str, **kwargs) -> Upload: ...
parts: AsyncPartsReal-time audio and WebSocket-based AI interactions for building conversational applications with low-latency voice communication.
class BetaRealtime:
def connect(
self,
*,
model: str,
websocket_connection_options: WebsocketConnectionOptions = {},
**kwargs
) -> RealtimeConnectionManager: ...
sessions: BetaSessions
class BetaSessions:
def create(
self,
*,
model: Any = "portkey-default",
modalities: Union[List[Any], NotGiven] = NOT_GIVEN,
voice: Union[Any, NotGiven] = NOT_GIVEN,
instructions: Union[str, NotGiven] = NOT_GIVEN,
**kwargs
) -> SessionCreateResponse: ...
class AsyncBetaRealtime:
def connect(self, **kwargs) -> AsyncRealtimeConnectionManager: ...
sessions: AsyncBetaSessionsManagement of AI providers and third-party integrations with workspace-level and model-level configuration.
class Providers:
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...
class Integrations:
workspaces: IntegrationsWorkspaces
models: IntegrationsModels
class IntegrationsWorkspaces:
def create(self, **kwargs): ...
def list(self, **kwargs): ...
def retrieve(self, **kwargs): ...Provider & Integration Management
Pre-built integrations with popular AI frameworks including LangChain and LlamaIndex callback handlers.
from portkey_ai.langchain import PortkeyLangchainCallbackHandler
from portkey_ai.llamaindex import PortkeyLlamaCallbackHandlerclass Config(TypedDict):
"""Configuration object for Portkey requests"""
...
class LLMOptions(TypedDict):
"""LLM-specific options and parameters"""
...
class Modes(Enum):
"""Available Portkey modes"""
SINGLE = "single"
FALLBACK = "fallback"
LOADBALANCE = "loadbalance"
AB_TEST = "ab_test"
class ProviderTypes(Enum):
"""Supported AI providers"""
OPENAI = "openai"
ANTHROPIC = "anthropic"
AZURE_OPENAI = "azure-openai"
COHERE = "cohere"
HUGGINGFACE = "huggingface"
# ... and 35+ more providers
class RetrySettings(TypedDict):
"""Retry configuration settings"""
attempts: int
on_status_codes: List[int]class PortkeyResponse:
"""Base response class for all Portkey operations"""
...
class ChatCompletions:
"""Chat completion response"""
id: str
object: str
created: int
model: str
choices: List[dict]
usage: dict
class ChatCompletionChunk:
"""Streaming chat completion chunk"""
id: str
object: str
created: int
model: str
choices: List[dict]
class TextCompletion:
"""Text completion response"""
id: str
object: str
created: int
model: str
choices: List[dict]
usage: dict
class Message:
"""Message object for chat completions"""
role: str
content: strclass APIStatusError(Exception):
"""Base exception for API status errors"""
...
class BadRequestError(APIStatusError):
"""400 Bad Request error"""
...
class AuthenticationError(APIStatusError):
"""401 Authentication error"""
...
class PermissionDeniedError(APIStatusError):
"""403 Permission denied error"""
...
class NotFoundError(APIStatusError):
"""404 Not found error"""
...
class RateLimitError(APIStatusError):
"""429 Rate limit error"""
...
class InternalServerError(APIStatusError):
"""500 Internal server error"""
...