fluxtokens 1.0.0

Official FluxTokens API SDK - Access GPT-4.1, Gemini 2.5 and more at 30% lower cost

FluxTokens Python SDK

Official Python SDK for the FluxTokens API

Access GPT-4.1, Gemini 2.5 Flash and more at 30% lower cost than competitors.

Website · Documentation · Dashboard

---

Installation

pip install fluxtokens

Quick Start

from fluxtokens import FluxTokens

client = FluxTokens(api_key="sk-flux-your-api-key")

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"},
    ],
)

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

Available Models

Model	Provider	Input Price	Output Price	Max Tokens	Vision	Audio	Video
gpt-4.1-mini	OpenAI	$0.28/1M	$1.12/1M	16,384	✅	❌	❌
gpt-4.1-nano	OpenAI	$0.07/1M	$0.28/1M	16,384	✅	❌	❌
gemini-2.5-flash	Google	$0.21/1M	$1.75/1M	65,536	✅	✅	✅

Usage Examples

Basic Chat Completion

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "What is the capital of France?"}],
    temperature=0.7,
    max_tokens=256,
)

print(response.choices[0].message.content)
# Output: "The capital of France is Paris."

Streaming Responses

stream = client.chat.completions.create(
    model="gemini-2.5-flash",
    messages=[{"role": "user", "content": "Write a haiku about programming."}],
    stream=True,
)

for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

Async Client

import asyncio
from fluxtokens import AsyncFluxTokens

async def main():
    client = AsyncFluxTokens(api_key="sk-flux-...")
    
    response = await client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    
    print(response.choices[0].message.content)
    await client.close()

asyncio.run(main())

Async Streaming

async def stream_response():
    async with AsyncFluxTokens(api_key="sk-flux-...") as client:
        stream = await client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=[{"role": "user", "content": "Tell me a story"}],
            stream=True,
        )
        
        async for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                print(content, end="", flush=True)

asyncio.run(stream_response())

Vision (Image Analysis)

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What is in this image?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/image.jpg",
                        "detail": "high",
                    },
                },
            ],
        }
    ],
    max_tokens=500,
)

Audio Input (Gemini only)

import base64

with open("audio.mp3", "rb") as f:
    audio_data = base64.b64encode(f.read()).decode()

response = client.chat.completions.create(
    model="gemini-2.5-flash",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Transcribe this audio:"},
                {
                    "type": "input_audio",
                    "input_audio": {
                        "data": audio_data,
                        "format": "mp3",
                    },
                },
            ],
        }
    ],
)

Video Analysis (Gemini only)

response = client.chat.completions.create(
    model="gemini-2.5-flash",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe what happens in this video:"},
                {
                    "type": "video_url",
                    "video_url": {"url": "https://example.com/video.mp4"},
                },
            ],
        }
    ],
    max_tokens=1000,
)

Context Manager

with FluxTokens(api_key="sk-flux-...") as client:
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    print(response.choices[0].message.content)
# Client is automatically closed

List Available Models

for model in client.models.list():
    print(f"{model.name} ({model.provider})")
    print(f"  Input: ${model.input_price}/1M tokens")
    print(f"  Output: ${model.output_price}/1M tokens")
    print(f"  Vision: {'✅' if model.supports_vision else '❌'}")

Error Handling

from fluxtokens import (
    FluxTokens,
    AuthenticationError,
    RateLimitError,
    InsufficientBalanceError,
    BadRequestError,
)

try:
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": "Hello"}],
    )
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limit exceeded, retry after: {e.retry_after}")
except InsufficientBalanceError:
    print("Please add credits at https://fluxtokens.io/dashboard/billing")
except BadRequestError as e:
    print(f"Invalid request: {e.message}")

Configuration

client = FluxTokens(
    api_key="sk-flux-...",
    base_url="https://api.fluxtokens.io",  # Custom base URL
    timeout=60.0,  # Request timeout in seconds
    max_retries=3,  # Max retries on rate limit/server errors
)

Migration from OpenAI SDK

- from openai import OpenAI
+ from fluxtokens import FluxTokens

- client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
+ client = FluxTokens(api_key=os.environ["FLUXTOKENS_API_KEY"])

response = client.chat.completions.create(
-    model="gpt-4o-mini",
+    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "Hello!"}],
)

Requirements

• Python 3.8+
• httpx >= 0.25.0
• pydantic >= 2.0.0

Support

• 📧 Email: [e-mail]
• 💬 Discord: Join our community
• 📖 Docs: fluxtokens.io/docs

License

MIT © FluxTokens