DEV Community

Malik Taimoor Awan
Malik Taimoor Awan

Posted on

The Requests library for AI one Unified Python SDK for every LLM provider

#ai #api #llm #python

UniversalAI

The Requests library for AI — one unified SDK for every LLM provider. Write once, run anywhere.

pip install universal-ai
Enter fullscreen mode Exit fullscreen mode 
from universal_ai import AI

ai = AI(provider="openai", model="gpt-4o")
response = await ai.chat("What is quantum computing?")
print(response.content)
Enter fullscreen mode Exit fullscreen mode

Why UniversalAI?

Building AI applications today means juggling multiple provider SDKs, each with different APIs, error handling, and quirks. UniversalAI gives you one clean interface that works across all major providers:

  • Same code works with OpenAI, Anthropic, Gemini, Ollama, Groq, Mistral, OpenRouter, HuggingFace, and Azure OpenAI
  • Switch providers by changing one string — no code rewrite
  • Built-in resilience with retry, caching, rate limiting, and circuit breaker middleware
  • Tool calling works identically across all providers that support it

Features

Feature Description
9 Providers OpenAI, Anthropic, Gemini, Ollama, Groq, Mistral, OpenRouter, HuggingFace, Azure OpenAI
Async-first Full async/await with synchronous wrappers for scripts and notebooks
Streaming Real-time token streaming from any provider
Tool Calling @tool decorator with automatic execution loop
Middleware Retry, cache, rate limit, circuit breaker, cost tracking, logging
Routing Fallback, round-robin, lowest latency, lowest cost strategies
Vision Image-aware chat with OpenAI, Anthropic, Gemini
Audio Transcription (Whisper) and TTS with OpenAI
Image Generation DALL-E 3 support
Embeddings OpenAI, Gemini, Mistral, HuggingFace, Azure, OpenRouter
RAG Built-in retrieval-augmented generation with chunking and vector store
Agents Multi-agent orchestration with coordinator pattern
Context Safety Automatic validation and optional truncation
Cost Tracking Per-request and cumulative cost estimation
CLI Full-featured uai command-line tool

Quick Start

Installation 

# Core SDK (auto-detects available providers)
pip install universal-ai

# With specific provider support
pip install universal-ai[openai]
pip install universal-ai[anthropic]
pip install universal-ai[gemini]
pip install universal-ai[ollama]

# Everything
pip install universal-ai[all]
Enter fullscreen mode Exit fullscreen mode

Basic Usage 

import asyncio
from universal_ai import AI

async def main():
    # Auto-detect provider from environment
    ai = AI()

    # Chat
    response = await ai.chat("Explain quantum computing in one sentence")
    print(response.content)

    # Streaming
    async for chunk in ai.stream("Write a haiku about programming"):
        print(chunk.delta, end="", flush=True)

    # Embeddings
    embed_response = await ai.embed("Hello, world!")
    print(f"Embedding dimensions: {len(embed_response.vector)}")

asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

With Specific Provider 

from universal_ai import AI

# OpenAI
ai = AI(provider="openai", model="gpt-4o")
response = await ai.chat("Hello!")

# Anthropic
ai = AI(provider="anthropic", model="claude-sonnet-4-20250514")
response = await ai.chat("Hello!")

# Local Ollama
ai = AI(provider="ollama", model="llama3")
response = await ai.chat("Hello!")
Enter fullscreen mode Exit fullscreen mode

Synchronous Usage 

from universal_ai import AI

ai = AI(provider="openai", model="gpt-4o")

# Synchronous wrappers for scripts/notebooks
response = ai.chat_sync("Hello!")
print(response.content)

# Sync streaming (returns full text)
text = ai.stream_sync("Tell me a joke")
print(text)
Enter fullscreen mode Exit fullscreen mode

Tool Calling

Define tools with the @tool decorator and let the AI use them:

from universal_ai import AI, tool

@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get current weather for a city."""
    # In a real app, call a weather API
    return f"Weather in {city}: 22°{unit[0].upper()}, sunny"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

ai = AI(provider="openai", model="gpt-4o")

# The AI will automatically call your tools
response = await ai.chat(
    "What's the weather in Paris? Also calculate 15 * 23.",
    tools=[get_weather, calculate]
)
print(response.content)
Enter fullscreen mode Exit fullscreen mode

Manual Tool Execution 

from universal_ai import AI, tool

@tool
def search(query: str) -> str:
    """Search the web."""
    return f"Results for: {query}"

ai = AI(provider="openai", model="gpt-4o")
ai.register_tool(search)

# Tools are auto-executed in the tool loop
response = await ai.chat("Search for Python tutorials")
Enter fullscreen mode Exit fullscreen mode

Conversations

Multi-turn conversations with automatic history management:

from universal_ai import AI

ai = AI(provider="openai", model="gpt-4o")

# Create a conversation
conv = ai.conversation(
    system_prompt="You are a helpful cooking assistant.",
    max_turns=20
)

# Send messages
response = await conv.send(message="What should I cook for dinner?")
print(response.content)

response = await conv.send(message="Can you give me a recipe?")
print(response.content)

# Access history
print(f"Turn count: {conv.turn_count}")
print(f"Messages: {len(conv.history)}")

# Reset
conv.reset()
Enter fullscreen mode Exit fullscreen mode

Configuration

Environment Variables 

# Provider selection
export UNIVERSALAI_PROVIDER=openai
export UNIVERSALAI_MODEL=gpt-4o

# API keys (provider-specific)
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...
export GEMINI_API_KEY=...
export GROQ_API_KEY=gsk_...
export MISTRAL_API_KEY=...
export OPENROUTER_API_KEY=sk-or-...
export HF_API_KEY=hf_...

# Azure OpenAI
export AZURE_OPENAI_API_KEY=...
export AZURE_OPENAI_API_BASE=https://your-resource.openai.azure.com
export AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4o

# Ollama (local)
export OLLAMA_HOST=http://localhost:11434
Enter fullscreen mode Exit fullscreen mode

Config File 

# ~/.config/universalai/config.yaml
provider: openai
model: gpt-4o
temperature: 0.7
max_tokens: 4096
timeout: 30
max_retries: 3
auto_truncate: true

fallback_providers:
  - anthropic
  - gemini

provider_api_keys:
  openai: sk-...
  anthropic: sk-ant-...
Enter fullscreen mode Exit fullscreen mode

Programmatic Configuration 

from universal_ai import AI, Config

config = Config(
    provider="openai",
    model="gpt-4o",
    temperature=0.7,
    max_tokens=4096,
    timeout=30,
    max_retries=3,
    auto_truncate=True,
    provider_api_keys={
        "openai": "sk-...",
        "anthropic": "sk-ant-...",
    }
)

ai = AI(config=config)
Enter fullscreen mode Exit fullscreen mode

Middleware

Add resilience and observability to your requests:

from universal_ai import AI
from universal_ai.middleware import (
    RetryMiddleware,
    CacheMiddleware,
    RateLimitMiddleware,
    CircuitBreakerMiddleware,
    CostTrackingMiddleware,
    LoggingMiddleware,
)

ai = AI(provider="openai", model="gpt-4o")

# Add middleware in order (executed top to bottom)
ai.add_middleware(LoggingMiddleware())
ai.add_middleware(CostTrackingMiddleware())
ai.add_middleware(RetryMiddleware(max_retries=3, base_delay=1.0))
ai.add_middleware(CacheMiddleware(ttl=300))
ai.add_middleware(RateLimitMiddleware(requests_per_minute=60))
ai.add_middleware(CircuitBreakerMiddleware(failure_threshold=5))

# All requests now go through the middleware pipeline
response = await ai.chat("Hello!")
Enter fullscreen mode Exit fullscreen mode

Middleware Reference

Middleware Purpose Key Options
RetryMiddleware Retry failed requests max_retries, base_delay, max_delay, jitter
CacheMiddleware Cache responses ttl, backend (memory/sqlite/redis)
RateLimitMiddleware Limit request rate requests_per_minute, burst
CircuitBreakerMiddleware Stop cascading failures failure_threshold, recovery_timeout
CostTrackingMiddleware Track API costs —
LoggingMiddleware Log requests/responses log_level

Routing Strategies

Automatically select the best provider:

from universal_ai import AI
from universal_ai.router import (
    Router,
    FallbackStrategy,
    RoundRobinStrategy,
    LowestLatencyStrategy,
    LowestCostStrategy,
)

# Configure fallback in config
config = Config(
    provider="openai",
    fallback_providers=["anthropic", "gemini"]
)

ai = AI(config=config)

# Or use router directly
router = Router(
    providers=["openai", "anthropic", "gemini"],
    strategy=FallbackStrategy()
)
Enter fullscreen mode Exit fullscreen mode

Strategy Options

Strategy Behavior
FallbackStrategy Try first provider, failover to next on error
RoundRobinStrategy Distribute requests evenly across providers
LowestLatencyStrategy Always use the fastest responding provider
LowestCostStrategy Always use the cheapest provider

RAG (Retrieval-Augmented Generation)

Build knowledge-base powered chat:

from universal_ai import AI
from universal_ai.rag import RAG, TextLoader, DirectoryLoader

# Initialize RAG
rag = RAG(chunk_size=500, chunk_overlap=50, top_k=3)

# Add content
rag.add_text("Python is a high-level programming language...")
rag.add_document(Document(content="...", source="docs.txt"))
rag.add_folder("./knowledge_base")
rag.add_url("https://example.com/article.txt")
rag.add_github("owner/repo")

# Search
chunks = await rag.search("What is Python?")
for chunk in chunks:
    print(f"Score: {chunk.content[:50]}...")

# Use with AI
ai = AI(provider="openai", model="gpt-4o")
augmented_request = await rag.augment_request(chat_request)
response = await ai.chat(augmented_request)
Enter fullscreen mode Exit fullscreen mode

Audio & Image

Transcription (Whisper) 

ai = AI(provider="openai", model="gpt-4o")

# Transcribe audio file
text = await ai.transcribe("audio.mp3")
print(text)

# Transcribe from bytes
text = await ai.transcribe(audio_bytes)
Enter fullscreen mode Exit fullscreen mode

Text-to-Speech 

# Generate speech
audio_bytes = await ai.speak("Hello, world!", voice="alloy")
with open("output.mp3", "wb") as f:
    f.write(audio_bytes)
Enter fullscreen mode Exit fullscreen mode

Image Generation 

# Generate image
urls = await ai.image("A sunset over mountains", size="1024x1024")
print(urls[0])  # URL to generated image
Enter fullscreen mode Exit fullscreen mode

CLI Usage

UniversalAI includes a full-featured command-line tool:

# Chat interactively
uai chat

# Chat with specific provider
uai chat -p openai -m gpt-4o

# Send a single message
uai chat "What is machine learning?"

# List available providers
uai providers

# Run diagnostics
uai doctor

# Manage configuration
uai config show
uai config set provider openai
uai config set-api-key openai

# Benchmark providers
uai benchmark --iterations 10

# Start local API server
uai serve --port 8000
Enter fullscreen mode Exit fullscreen mode

Provider Details

OpenAI 

ai = AI(provider="openai", model="gpt-4o")

# Features: Chat, Streaming, Vision, Tools, Embeddings, Audio, Image Gen
# Requires: OPENAI_API_KEY
Enter fullscreen mode Exit fullscreen mode

Anthropic 

ai = AI(provider="anthropic", model="claude-sonnet-4-20250514")

# Features: Chat, Streaming, Vision, Tools
# Requires: ANTHROPIC_API_KEY
Enter fullscreen mode Exit fullscreen mode

Gemini 

ai = AI(provider="gemini", model="gemini-2.0-flash")

# Features: Chat, Streaming, Vision, Tools, Embeddings
# Requires: GEMINI_API_KEY
Enter fullscreen mode Exit fullscreen mode

Ollama (Local) 

ai = AI(provider="ollama", model="llama3")

# Features: Chat, Streaming, Embeddings
# Requires: Ollama running locally
# Install: https://ollama.ai
Enter fullscreen mode Exit fullscreen mode

Groq 

ai = AI(provider="groq", model="llama-3.1-70b-versatile")

# Features: Chat, Streaming, Tools
# Requires: GROQ_API_KEY
Enter fullscreen mode Exit fullscreen mode

Mistral 

ai = AI(provider="mistral", model="mistral-large-latest")

# Features: Chat, Streaming, Tools, Embeddings
# Requires: MISTRAL_API_KEY
Enter fullscreen mode Exit fullscreen mode

OpenRouter 

ai = AI(provider="openrouter", model="openai/gpt-4o")

# Features: Chat, Streaming, Vision, Tools, Embeddings
# Requires: OPENROUTER_API_KEY
Enter fullscreen mode Exit fullscreen mode

HuggingFace 

ai = AI(provider="huggingface", model="meta-llama/Llama-2-7b-chat-hf")

# Features: Chat, Streaming, Embeddings
# Requires: HF_API_KEY
Enter fullscreen mode Exit fullscreen mode

Azure OpenAI 

ai = AI(provider="azure", model="gpt-4o")

# Features: Chat, Streaming, Vision, Tools, Embeddings
# Requires: AZURE_OPENAI_API_KEY, AZURE_OPENAI_API_BASE
Enter fullscreen mode Exit fullscreen mode

Error Handling 

from universal_ai import AI
from universal_ai.exceptions import (
    AuthenticationError,
    RateLimitError,
    ContextWindowExceededError,
    ProviderError,
    TimeoutError,
)

ai = AI(provider="openai", model="gpt-4o")

try:
    response = await ai.chat("Hello!")
except AuthenticationError as e:
    print(f"Invalid API key: {e}")
except RateLimitError as e:
    print(f"Rate limited, retry after: {e.retry_after}s")
except ContextWindowExceededError as e:
    print(f"Context too long: {e.estimated_tokens} > {e.context_window}")
except ProviderError as e:
    print(f"Provider error: {e}")
except TimeoutError:
    print("Request timed out")
Enter fullscreen mode Exit fullscreen mode

Context Window Safety

UniversalAI validates that messages fit within the provider's context window:

from universal_ai import AI, Config

# Option 1: Raise error if too long (default)
config = Config(auto_truncate=False)
ai = AI(config=config)

# Option 2: Auto-truncate to fit
config = Config(auto_truncate=True)
ai = AI(config=config)
Enter fullscreen mode Exit fullscreen mode

Cost Estimation 

ai = AI(provider="openai", model="gpt-4o")

# Estimate cost before sending
estimated_cost = ai.estimate_cost("Hello, world!")
print(f"Estimated cost: ${estimated_cost:.6f}")

# Track actual costs with middleware
from universal_ai.middleware import CostTrackingMiddleware

cost_middleware = CostTrackingMiddleware()
ai.add_middleware(cost_middleware)

response = await ai.chat("Hello!")
print(f"Actual cost: ${response.usage.estimated_cost:.6f}")
print(f"Total cost: ${cost_middleware.total_cost:.6f}")
Enter fullscreen mode Exit fullscreen mode

Sync Wrappers

For scripts and notebooks where you can't use async:

Async Method Sync Wrapper
await ai.chat(...) ai.chat_sync(...)
async for chunk in ai.stream(...) ai.stream_sync(...)
await ai.embed(...) ai.embed_sync(...)
await ai.chat_with_tools(...) ai.chat_with_tools_sync(...)
await ai.chat_json(...) ai.chat_json_sync(...)

Examples

See the examples/ directory for complete working examples:

  • basic_chat.py - Simple chat usage
  • streaming.py - Real-time streaming
  • tool_calling.py - Tool definition and execution
  • middleware_demo.py - Middleware configuration
  • rag_demo.py - RAG with document loading
  • multi_provider.py - Provider switching

Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines.

# Clone the repo
git clone https://github.com/6t9xstar/universal-ai.git
cd universal-ai

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run linting
ruff check .

# Run type checking
mypy .
Enter fullscreen mode Exit fullscreen mode

License

MIT License - see LICENSE for details.

Top comments (0)