Writing Robust, Resilient Python Clients for MindRouter

from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url="https://mindrouter.example.com/v1",
    api_key="mr2_your_api_key_here",
)

async def stream_chat(prompt: str) -> str:
    stream = await async_client.chat.completions.create(
        model="openai/gpt-oss-120b",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )

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

    return full_response

import openai

async def stream_with_recovery(messages: list, max_retries: int = 3) -> str:
    """Stream a response with retry on mid-stream disconnect."""
    for attempt in range(max_retries):
        collected = ""
        try:
            stream = await async_client.chat.completions.create(
                model="openai/gpt-oss-120b",
                messages=messages,
                stream=True,
            )
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    collected += delta
                    yield delta
            return  # stream completed successfully

        except (openai.APIConnectionError, openai.APITimeoutError):
            if attempt == max_retries - 1:
                raise
            # Append partial response and ask model to continue
            if collected:
                messages = messages + [
                    {"role": "assistant", "content": collected},
                    {"role": "user", "content": "Continue from where you left off."},
                ]
            await asyncio.sleep(2 ** attempt)  # backoff before retry

texts = [
    "Machine learning is a subset of artificial intelligence.",
    "Deep learning uses neural networks with many layers.",
    "Natural language processing handles human language.",
]

response = client.embeddings.create(
    model="Qwen/Qwen3-Embedding-8B",
    input=texts,
)

for i, item in enumerate(response.data):
    print(f"Text {i}: {len(item.embedding)} dimensions")

{
  "object": "list",
  "data": [
    {
      "object": "embedding",
      "index": 0,
      "embedding": [0.0123, -0.0456, 0.0789, ...]
    }
  ],
  "model": "Qwen/Qwen3-Embedding-8B",
  "usage": {
    "prompt_tokens": 10,
    "total_tokens": 10
  }
}

import httpx

# The OpenAI SDK doesn't have a built-in rerank method,
# so use httpx directly
response = httpx.post(
    "https://mindrouter.example.com/v1/rerank",
    headers={"Authorization": "Bearer mr2_your_api_key_here"},
    json={
        "model": "Qwen/Qwen3-Reranker-8B",
        "query": "What is the capital of France?",
        "documents": [
            "Paris is the capital and most populous city of France.",
            "Berlin is the capital of Germany.",
            "The Eiffel Tower is located in Paris.",
            "London is the capital of England.",
        ],
        "top_n": 3,
        "return_documents": True,
    },
    timeout=60.0,
)

results = response.json()
for result in results["results"]:
    print(f"Score: {result['relevance_score']:.4f} — {result['document'][:60]}...")

response = httpx.post(
    "https://mindrouter.example.com/v1/score",
    headers={"Authorization": "Bearer mr2_your_api_key_here"},
    json={
        "model": "Qwen/Qwen3-Reranker-8B",
        "text_1": "What is machine learning?",
        "text_2": [
            "Machine learning is a branch of AI that learns from data.",
            "The weather today is sunny and warm.",
        ],
    },
    timeout=60.0,
)

# Handle image generation with appropriate timeout
import httpx

client = OpenAI(
    base_url="https://mindrouter.example.com/v1",
    api_key="mr2_your_api_key_here",
    timeout=httpx.Timeout(300.0, connect=5.0),  # 5 min for slow diffusion
)

try:
    response = client.images.generate(
        model="black-forest-labs/FLUX.2-dev",
        prompt="A robot painting a landscape",
        size="1024x1024",
    )
except openai.PermissionDeniedError as e:
    # 403: content policy violation or feature not enabled
    print(f"Denied: {e.message}")
except openai.BadRequestError as e:
    # 400: invalid size, missing prompt, etc.
    print(f"Bad request: {e.message}")

{
  "object": "list",
  "data": [
    {
      "id": "openai/gpt-oss-120b",
      "object": "model",
      "created": 1713400000,
      "owned_by": "mindrouter",
      "capabilities": {
        "multimodal": false,
        "embeddings": false,
        "structured_output": true,
        "thinking": true,
        "tools": true
      },
      "backends": ["aspen1-gpu1", "aspen3-gpu1"],
      "context_length": 131072,
      "model_max_context": 131072,
      "parameter_count": "120B",
      "quantization": null,
      "family": "gpt"
    }
  ]
}

import openai
import logging

logger = logging.getLogger(__name__)

# Errors worth Treing
RETRYABLE_ERRORS = (
    openai.RateLimitError,       # 429
    openai.APITimeoutError,      # timeout
    openai.APIConnectionError,   # network
    openai.InternalServerError,  # 500/502/503/504
)

# Errors that will never succeed on retry
FATAL_ERRORS = (
    openai.BadRequestError,          # 400
    openai.AuthenticationError,      # 401
    openai.PermissionDeniedError,    # 403
    openai.NotFoundError,            # 404
    openai.UnprocessableEntityError, # 422
)

def call_with_error_handling(func, *args, **kwargs):
    """Call an OpenAI SDK method with proper error classification."""
    try:
        return func(*args, **kwargs)
    except FATAL_ERRORS as e:
        logger.error(f"Non-retryable error ({e.status_code}): {e.message}")
        raise  # don't retry — fix the request
    except openai.RateLimitError as e:
        logger.warning(f"Rate limited: {e.message}")
        raise  # caller should implement backoff
    except openai.InternalServerError as e:
        logger.warning(f"Server error ({e.status_code}): {e.message}")
        raise  # caller should retry with backoff
    except (openai.APIConnectionError, openai.APITimeoutError) as e:
        logger.warning(f"Connection issue: {e}")
        raise  # caller should retry with backoff

pip install tenacity

import openai
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type,
    before_sleep_log,
)
import logging

logger = logging.getLogger(__name__)

@retry(
    wait=wait_random_exponential(min=1, max=60),
    stop=stop_after_attempt(6),
    retry=retry_if_exception_type((
        openai.RateLimitError,
        openai.APITimeoutError,
        openai.APIConnectionError,
        openai.InternalServerError,
    )),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def resilient_chat(client, **kwargs):
    """Chat completion with tenacity-managed retries."""
    return client.chat.completions.create(**kwargs)

# Usage
response = resilient_chat(
    client,
    model="openai/gpt-oss-120b",
    messages=[{"role": "user", "content": "hello"}],
)

import random
import time
import openai

def chat_with_backoff(
    client: OpenAI,
    messages: list,
    model: str = "openai/gpt-oss-120b",
    max_retries: int = 6,
    initial_delay: float = 1.0,
    max_delay: float = 60.0,
    backoff_factor: float = 2.0,
) -> openai.types.chat.ChatCompletion:
    """
    Chat completion with exponential backoff and full jitter.

    Jitter prevents the "thundering herd" problem where many clients
    retry simultaneously after a shared failure, overwhelming the server.
    """
    delay = initial_delay

    for attempt in range(max_retries + 1):
        try:
            return client.chat.completions.create(
                model=model,
                messages=messages,
            )
        except openai.RateLimitError as e:
            if attempt == max_retries:
                raise
            # Full jitter: uniform random between 0 and calculated delay
            jittered_delay = random.uniform(0, delay)
            print(f"Rate limited (attempt {attempt + 1}/{max_retries + 1}). "
                  f"Waiting {jittered_delay:.1f}s...")
            time.sleep(jittered_delay)
            delay = min(delay * backoff_factor, max_delay)

        except openai.InternalServerError as e:
            if attempt == max_retries:
                raise
            # Server errors get shorter initial backoff since MindRouter
            # already retried internally
            jittered_delay = random.uniform(0, delay)
            print(f"Server error {e.status_code} (attempt {attempt + 1}). "
                  f"Waiting {jittered_delay:.1f}s...")
            time.sleep(jittered_delay)
            delay = min(delay * backoff_factor, max_delay)

        except (openai.APIConnectionError, openai.APITimeoutError) as e:
            if attempt == max_retries:
                raise
            jittered_delay = random.uniform(0, delay)
            time.sleep(jittered_delay)
            delay = min(delay * backoff_factor, max_delay)

import asyncio
import openai
import time
import random

class AdaptiveClient:
    """
    Dynamically adjusts concurrency based on server responses.

    - Halves concurrency on 429 (rate limit)
    - Reduces by 25% on 503 (no capacity)
    - Gradually increases concurrency on sustained success
    """

    def __init__(
        self,
        client: AsyncOpenAI,
        initial_concurrency: int = 20,
        min_concurrency: int = 1,
        max_concurrency: int = 50,
        recovery_interval: float = 30.0,  # seconds between increases
    ):
        self.client = client
        self.concurrency = initial_concurrency
        self.min_concurrency = min_concurrency
        self.max_concurrency = max_concurrency
        self.recovery_interval = recovery_interval
        self._semaphore = asyncio.Semaphore(initial_concurrency)
        self._lock = asyncio.Lock()
        self._last_reduction = 0.0
        self._consecutive_successes = 0

    async def _adjust_concurrency(self, new_limit: int):
        async with self._lock:
            new_limit = max(self.min_concurrency, min(self.max_concurrency, new_limit))
            if new_limit != self.concurrency:
                old = self.concurrency
                self.concurrency = new_limit
                self._semaphore = asyncio.Semaphore(new_limit)
                self._last_reduction = time.monotonic()
                self._consecutive_successes = 0
                print(f"Concurrency adjusted: {old} → {new_limit}")

    async def _maybe_increase(self):
        async with self._lock:
            self._consecutive_successes += 1
            elapsed = time.monotonic() - self._last_reduction
            # Only increase if we've had sustained success after a cooldown
            if (
                elapsed > self.recovery_interval
                and self._consecutive_successes >= 20
                and self.concurrency < self.max_concurrency
            ):
                new_limit = min(self.concurrency + 2, self.max_concurrency)
                self.concurrency = new_limit
                self._semaphore = asyncio.Semaphore(new_limit)
                self._consecutive_successes = 0
                print(f"Concurrency increased to {new_limit}")

    async def chat(self, messages: list, model: str = "openai/gpt-oss-120b", max_retries: int = 5):
        """Make a chat request with adaptive concurrency and retry."""
        delay = 1.0

        for attempt in range(max_retries):
            async with self._semaphore:
                try:
                    response = await self.client.chat.completions.create(
                        model=model,
                        messages=messages,
                    )
                    await self._maybe_increase()
                    return response

                except openai.RateLimitError:
                    # Aggressive reduction — halve concurrency
                    await self._adjust_concurrency(self.concurrency // 2)
                    jitter = random.uniform(0, delay)
                    await asyncio.sleep(jitter)
                    delay = min(delay * 2, 60.0)

                except openai.InternalServerError as e:
                    if e.status_code == 503:
                        # Moderate reduction — cluster is overloaded
                        await self._adjust_concurrency(int(self.concurrency * 0.75))
                    jitter = random.uniform(0, delay)
                    await asyncio.sleep(jitter)
                    delay = min(delay * 2, 60.0)

                except (openai.APIConnectionError, openai.APITimeoutError):
                    jitter = random.uniform(0, delay)
                    await asyncio.sleep(jitter)
                    delay = min(delay * 2, 60.0)

        raise RuntimeError(f"Failed after {max_retries} attempts")


# Usage
async def main():
    async_client = AsyncOpenAI(
        base_url="https://mindrouter.example.com/v1",
        api_key="mr2_your_api_key_here",
    )
    adaptive = AdaptiveClient(async_client, initial_concurrency=20)

    prompts = [f"Write a haiku about topic {i}" for i in range(100)]

    tasks = [
        adaptive.chat([{"role": "user", "content": p}])
        for p in prompts
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]
    print(f"Completed: {len(successes)}, Failed: {len(failures)}")

import httpx
from openai import OpenAI, AsyncOpenAI

# Sync client
client = OpenAI(
    base_url="https://mindrouter.example.com/v1",
    api_key="mr2_your_api_key_here",
    timeout=httpx.Timeout(
        200.0,       # overall timeout — slightly above MindRouter's 180s per-attempt
        connect=5.0, # fail fast on DNS/network issues
        read=200.0,  # allow slow model generation
        write=10.0,  # request upload should be fast
        pool=10.0,   # don't wait too long for an HTTP connection from the pool
    ),
    max_retries=3,
)

# Async client with custom connection pool
http_client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=100,          # total connection pool size
        max_keepalive_connections=50, # idle connections to keep warm
    ),
    timeout=httpx.Timeout(200.0, connect=5.0),
)

async_client = AsyncOpenAI(
    base_url="https://mindrouter.example.com/v1",
    api_key="mr2_your_api_key_here",
    http_client=http_client,
)

import time

class CircuitBreaker:
    """
    Three states:
    - CLOSED: normal operation, requests pass through
    - OPEN: failing fast, no requests sent
    - HALF_OPEN: testing if service recovered (one probe request)
    """

    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = 0.0
        self.state = "closed"

    def before_request(self):
        """Call before making a request. Raises if circuit is open."""
        if self.state == "open":
            if time.monotonic() - self.last_failure_time >= self.reset_timeout:
                self.state = "half_open"
            else:
                raise CircuitOpenError(
                    f"Circuit breaker is OPEN. "
                    f"Retry in {self.reset_timeout - (time.monotonic() - self.last_failure_time):.0f}s"
                )

    def on_success(self):
        """Call after a successful request."""
        self.failures = 0
        self.state = "closed"

    def on_failure(self):
        """Call after a failed request."""
        self.failures += 1
        self.last_failure_time = time.monotonic()
        if self.failures >= self.failure_threshold:
            self.state = "open"


class CircuitOpenError(Exception):
    pass


# Usage with OpenAI client
breaker = CircuitBreaker(failure_threshold=5, reset_timeout=30.0)

def safe_chat(client, messages, model="openai/gpt-oss-120b"):
    breaker.before_request()
    try:
        response = client.chat.completions.create(model=model, messages=messages)
        breaker.on_success()
        return response
    except (openai.APIConnectionError, openai.InternalServerError) as e:
        breaker.on_failure()
        raise
    except openai.RateLimitError:
        # Rate limits are not server failures — don't trip the breaker
        raise

response = client.chat.completions.create(
    model="openai/gpt-oss-120b",
    messages=[
        {"role": "system", "content": "Return a JSON object with fields: name, age, city."},
        {"role": "user", "content": "Tell me about Alice who is 30 and lives in Portland."},
    ],
    response_format={"type": "json_object"},
)

import json
data = json.loads(response.choices[0].message.content)
print(data)  # {"name": "Alice", "age": 30, "city": "Portland"}

response = client.chat.completions.create(
    model="openai/gpt-oss-120b",
    messages=[
        {"role": "user", "content": "Extract: Alice is 30 and lives in Portland."},
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "person",
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"},
                    "city": {"type": "string"},
                },
                "required": ["name", "age", "city"],
            },
        },
    },
)

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City and state, e.g. 'San Francisco, CA'",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                    },
                },
                "required": ["location"],
            },
        },
    }
]

response = client.chat.completions.create(
    model="openai/gpt-oss-120b",
    messages=[{"role": "user", "content": "What's the weather in Moscow, Idaho?"}],
    tools=tools,
    tool_choice="auto",
)

choice = response.choices[0]
if choice.finish_reason == "tool_calls":
    for tool_call in choice.message.tool_calls:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)
        print(f"Call {name} with {args}")

        # Execute the tool, then send results back
        tool_result = get_weather(**args)  # your implementation

        followup = client.chat.completions.create(
            model="openai/gpt-oss-120b",
            messages=[
                {"role": "user", "content": "What's the weather in Moscow, Idaho?"},
                choice.message,  # assistant's tool call
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(tool_result),
                },
            ],
            tools=tools,
        )