zaguan-sdk 0.2.0

Official Python SDK for Zaguan CoreX

🚀 Zaguan Python SDK

The official Python SDK for Zaguan CoreX

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

Installation • Quick Start • Documentation • Examples • Support

---

🌟 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

• Quick Start Guide - Get up and running in 5 minutes
• Complete Examples - Real-world usage examples
• API Reference - Full type documentation

🎯 Advanced Features

• Provider-Specific Examples - Use unique features from each provider
• Advanced Features Guide - Streaming, retry logic, observability, and more
• Anthropic Messages API - Native Anthropic API with extended thinking

🏗️ Architecture

• SDK Design Overview - Architecture and design principles
• HTTP Contract - Wire-level API specification
• Provider Features - Provider-specific capabilities

🧪 Development

• Testing Guide - Testing strategy and best practices
• Contributing - How to contribute to the SDK

🚀 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

• 📧 Email: [e-mail]
• 🌐 Website: https://zaguanai.com
• 📚 Documentation: https://zaguanai.com/docs
• 🐛 Issues: GitHub Issues

Community

• 💬 Join our Discord community (coming soon)
• 🐦 Follow us on Twitter (coming soon)
• 📝 Read our Blog (coming soon)

📄 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