Skip to content

ZaguanLabs/zaguan-sdk-python

Repository files navigation

🚀 Zaguan Python SDK

The official Python SDK for Zaguan CoreX

PyPI version Python 3.8+ License Tests

One API. 15+ Providers. 500+ Models. Infinite Possibilities.

InstallationQuick StartDocumentationExamplesSupport


🌟 What is Zaguan?

Zaguan CoreX is an enterprise-grade AI gateway that provides unified access to multiple AI providers through a single, OpenAI-compatible API. Think of it as your universal adapter for AI services.

Why Zaguan?

  • 🔌 One API for Everything - Switch between OpenAI, Anthropic, Google, and 12+ other providers without changing code
  • 💰 Built-in Credits System - Track usage, set limits, and manage costs across all providers
  • 🎯 OpenAI Compatible - Drop-in replacement for OpenAI SDK with zero code changes
  • 🚀 Production Ready - Enterprise-grade reliability, monitoring, and support
  • 🔒 Secure & Compliant - SOC 2, GDPR, and HIPAA ready infrastructure

Installation

pip install zaguan-sdk

🚀 Quick Start

Basic Chat Completion

from zaguan_sdk import ZaguanClient, ChatRequest, Message

# Initialize the client
client = ZaguanClient(
    base_url="https://api.zaguanai.com",
    api_key="your-api-key"
)

# Simple chat completion
response = client.chat(ChatRequest(
    model="openai/gpt-4o-mini",
    messages=[Message(role="user", content="What is Python?")]
))

print(response.choices[0].message.content)

Streaming Responses

# Stream responses in real-time
for chunk in client.chat_stream(ChatRequest(
    model="openai/gpt-4o-mini",
    messages=[Message(role="user", content="Tell me a story")]
)):
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Async Support

import asyncio
from zaguan_sdk import AsyncZaguanClient

async def main():
    async with AsyncZaguanClient(
        base_url="https://api.zaguanai.com",
        api_key="your-api-key"
    ) as client:
        response = await client.chat(ChatRequest(
            model="anthropic/claude-3-5-sonnet",
            messages=[Message(role="user", content="Hello!")]
        ))
        print(response.choices[0].message.content)

asyncio.run(main())

Multi-Provider Usage

# Switch providers without changing code
models = [
    "openai/gpt-4o",
    "anthropic/claude-3-5-sonnet",
    "google/gemini-2.0-flash",
    "deepseek/deepseek-chat"
]

for model in models:
    response = client.chat(ChatRequest(
        model=model,
        messages=[Message(role="user", content="Hi!")]
    ))
    print(f"{model}: {response.choices[0].message.content}")

✨ Features

🎯 Core Capabilities

Feature Description
🔄 Sync & Async Both ZaguanClient and AsyncZaguanClient for any use case
🔌 OpenAI Compatible Drop-in replacement - change 3 lines, keep everything else
🌐 Multi-Provider Access OpenAI, Anthropic, Google, DeepSeek, and 12+ more
📡 Streaming Real-time response streaming with cancellation support
🛡️ Type Safe Full type hints and Pydantic validation for all models
⚡ Production Ready Comprehensive error handling, retries, and timeouts

📦 Complete API Coverage

💬 Chat & Completions

  • ✅ Chat completions
  • ✅ Streaming responses
  • ✅ Function calling
  • ✅ Tool use
  • ✅ Vision (multimodal)

🧠 Embeddings & Search

  • ✅ Text embeddings
  • ✅ Batch embeddings
  • ✅ Semantic search ready

🎨 Image Generation

  • ✅ DALL-E 2 & 3
  • ✅ Image editing
  • ✅ Image variations

🎙️ Audio Processing

  • ✅ Speech-to-text (Whisper)
  • ✅ Audio translation
  • ✅ Text-to-speech (6 voices)

🔍 Content Safety

  • ✅ Content moderation
  • ✅ Policy compliance
  • ✅ Safety scores

💰 Credits & Usage

  • ✅ Balance tracking
  • ✅ Usage history
  • ✅ Cost analytics

🎯 100% coverage of major OpenAI-compatible endpoints

📚 Examples

Embeddings for Semantic Search

from zaguan_sdk import EmbeddingRequest

# Create embeddings
response = client.create_embeddings(EmbeddingRequest(
    model="openai/text-embedding-3-small",
    input=["Python is great", "I love coding"]
))

for embedding in response.data:
    print(f"Embedding: {embedding.embedding[:5]}...")  # First 5 dimensions

Audio Transcription

# Transcribe audio file
transcription = client.create_transcription(
    file_path="meeting.mp3",
    model="whisper-1",
    language="en"
)
print(transcription.text)

Image Generation

from zaguan_sdk import ImageGenerationRequest

# Generate image with DALL-E
response = client.create_image(ImageGenerationRequest(
    prompt="A serene mountain landscape at sunset",
    model="dall-e-3",
    size="1024x1024",
    quality="hd"
))
print(response.data[0].url)

Content Moderation

from zaguan_sdk import ModerationRequest

# Check content safety
result = client.create_moderation(ModerationRequest(
    input="Your content here"
))

if result.results[0].flagged:
    print("Content flagged:", result.results[0].categories)

📁 More examples in examples/ directory

✨ Advanced Features

Anthropic Messages API (Native)

Access Anthropic's extended thinking and native API features:

from zaguan_sdk import (
    AnthropicMessagesRequest,
    AnthropicMessage,
    AnthropicThinkingConfig
)

# Use extended thinking for complex reasoning
request = AnthropicMessagesRequest(
    model="anthropic/claude-3-5-sonnet",
    messages=[
        AnthropicMessage(role="user", content="Explain quantum computing")
    ],
    max_tokens=2048,
    thinking=AnthropicThinkingConfig(
        type="enabled",
        budget_tokens=5000  # Internal reasoning budget
    )
)

response = client.messages(request)

# Process thinking and response blocks
for block in response.content:
    if block.type == "thinking":
        print(f"🧠 Thinking: {block.thinking}")
    elif block.type == "text":
        print(f"💬 Response: {block.text}")

Streaming Utilities

Accumulate streaming responses easily:

from zaguan_sdk import StreamAccumulator

accumulator = StreamAccumulator()

for chunk in client.chat_stream(request):
    accumulator.add_chunk(chunk)
    # Print as we go
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

# Get the complete message
message = accumulator.get_message()

Retry Logic

Built-in retry with exponential backoff:

from zaguan_sdk import RetryConfig, with_retry

retry_config = RetryConfig(
    max_retries=5,
    initial_delay=1.0,
    exponential_base=2.0,
    jitter=True
)

@with_retry(retry_config)
def make_request():
    return client.chat(request)

response = make_request()  # Automatically retries on failures

Observability

Track metrics and logs:

from zaguan_sdk import MetricsCollector, LoggingHook

# Collect metrics
metrics = MetricsCollector()

# After requests...
summary = metrics.get_summary()
print(f"Success rate: {summary['success_rate']:.2%}")
print(f"Average latency: {summary['average_latency_ms']:.2f}ms")
print(f"Total cost: ${summary['total_cost']:.6f}")

📚 Learn more: Advanced Features Guide

🌐 Supported Providers

Access 15+ AI providers through one unified API:

Provider Models Specialties
OpenAI GPT-4o, GPT-4, GPT-3.5 Chat, embeddings, images, audio
Anthropic Claude 3.5 Sonnet, Opus Long context, analysis
Google Gemini 2.0, Gemini Pro Multimodal, reasoning
DeepSeek DeepSeek-V3, Reasoner Coding, reasoning
Alibaba Qwen 2.5 Multilingual
xAI Grok 2 Real-time data
Perplexity Sonar Search-augmented
Cohere Command R+ Enterprise RAG
Groq Llama 3, Mixtral Ultra-fast inference
And more...

🔗 Full provider list: docs/SDK/SDK_PROVIDER_FEATURES.md

📖 Documentation

📘 Getting Started

🎯 Advanced Features

🏗️ Architecture

🧪 Development

🚀 Migration from OpenAI SDK

Switching from OpenAI SDK? It's just 3 lines:

# Before (OpenAI SDK)
from openai import OpenAI
client = OpenAI(api_key="sk-...")

# After (Zaguan SDK)
from zaguan_sdk import ZaguanClient
client = ZaguanClient(
    base_url="https://api.zaguanai.com",
    api_key="your-zaguan-key"
)

# Everything else stays exactly the same! 🎉
response = client.chat.completions.create(...)

🛠️ Development

Setup Development Environment

# Clone repository
git clone https://github.com/ZaguanLabs/zaguan-sdk-python.git
cd zaguan-sdk-python

# Install dependencies
pip install -r requirements.txt

# Install in development mode
pip install -e .

Running Tests

# Run all tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=zaguan_sdk --cov-report=html

# Run specific test file
pytest tests/test_client.py -v

Building and Publishing

# Build package
make build

# Run tests
make test

# Publish to PyPI (maintainers only)
make publish

🤝 Support

Getting Help

Community

📄 License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

🙏 Acknowledgments

Built with ❤️ using:

  • httpx - Modern HTTP client
  • pydantic - Data validation and type safety
  • pytest - Testing framework

⬆ Back to Top

Made with ❤️ by Zaguan Labs