Routing image inputs through a unified gateway forces you to reconcile three different provider shapes for the same JPEG. A robust image inputs unified gateway normalizes these into one request contract, then maps to each backend without leaking provider quirks to your application code. This guide walks through building that translation layer and wiring fallback so a degraded vision provider doesn’t break your pipeline.
Step 1: Adopt a single internal message schema
Pick one provider’s format as your canonical representation. The OpenAI chat completions schema is the path of least resistance because most gateways already proxy it. A vision message looks like this:
{
"role": "user",
"content": [
{"type": "text", "text": "Describe this image"},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.jpg"}}
]
}Base64 works too: "url": "data:image/jpeg;base64,/9j/4AAQ...". Your gateway should accept both. Don’t invent a new envelope; every downstream mapping becomes simpler when the internal shape matches the most common API.
Step 2: Validate and preprocess inbound images
Never trust client-supplied URLs or blobs. Enforce a max payload size, allowlisted content types, and a timeout for any fetch you perform server-side. If a client sends a URL, decide whether the gateway fetches it or passes the URL through. Passing through is cheaper but breaks if the provider can’t reach the URL or requires signed requests.
from urllib.parse import urlparse
import base64
MAX_BYTES = 20 * 1024 * 1024 # 20 MB
def normalize_image(block: dict) -> dict:
url = block["image_url"]["url"]
if url.startswith("data:"):
# client already base64-encoded
return block
parsed = urlparse(url)
if parsed.scheme not in ("http", "https"):
raise ValueError("Unsupported image scheme")
# Gateway fetches and re-encodes to base64 to shield providers from private URLs
import requests
resp = requests.get(url, timeout=5)
resp.raise_for_status()
ct = resp.headers.get("content-type", "")
if not ct.startswith("image/"):
raise ValueError(f"Not an image: {ct}")
if len(resp.content) > MAX_BYTES:
raise ValueError("Image too large")
b64 = base64.b64encode(resp.content).decode()
return {"type": "image_url", "image_url": {"url": f"data:{ct};base64,{b64}"}}This step converts every image to an inline data URI. That trades bandwidth for a consistent backend contract and avoids provider-side fetch failures.
Step 3: Map the normalized request to provider payloads
Your unified gateway now holds a list of OpenAI-style messages. Translate them per provider.
OpenAI
Pass through unchanged. Set model to a vision-capable deployment. The image_url block is native.
Anthropic Claude
Anthropic expects content as an array of blocks where images use source with media_type and data (base64 without the data URI prefix).
def to_anthropic(messages: list) -> list:
out = []
for m in messages:
if m["role"] == "assistant":
out.append({"role": "assistant", "content": m["content"]})
continue
content = []
for part in m["content"]:
if part["type"] == "text":
content.append({"type": "text", "text": part["text"]})
elif part["type"] == "image_url":
uri = part["image_url"]["url"]
header, b64 = uri.split(",", 1)
media = header.split(";")[0].split(":")[1]
content.append({
"type": "image",
"source": {"type": "base64", "media_type": media, "data": b64}
})
out.append({"role": m["role"], "content": content})
return outGoogle Gemini
Gemini uses contents with parts. Inline images go in inline_data with mime_type and data (base64).
def to_gemini(messages: list) -> list:
contents = []
for m in messages:
parts = []
for part in m["content"]:
if part["type"] == "text":
parts.append({"text": part["text"]})
elif part["type"] == "image_url":
uri = part["image_url"]["url"]
header, b64 = uri.split(",", 1)
mime = header.split(";")[0].split(":")[1]
parts.append({"inline_data": {"mime_type": mime, "data": b64}})
contents.append({"role": "user" if m["role"] == "user" else "model", "parts": parts})
return contentsThese three functions are the core of an image inputs unified gateway. They are pure transforms; no network calls except the optional fetch in Step 2.
Step 4: Route with fallback and cache hints
Provider vision endpoints fail in boring ways: 429s, 503s, or silently dropped images. Build a router that tries the primary model, then falls back to a secondary on transport errors. Honor any x-routing directive from the client and forward cache-control hints so providers can reuse image preprocessing.
If you stand up your own image inputs unified gateway, you’ll need to implement health checks and fallback. A gateway like n4n.ai exposes one OpenAI-compatible endpoint across 240+ models and automatically fails over when a provider is rate-limited, while honoring your routing directives and forwarding cache-control hints—but the mapping logic above still applies if you roll your own.
Implement a simple retry wrapper:
import time
import openai
def complete_with_fallback(messages, models):
last_err = None
for model in models:
try:
return openai.ChatCompletion.create(model=model, messages=messages)
except openai.error.RateLimitError as e:
last_err = e
time.sleep(0.5)
raise last_errReplace openai.ChatCompletion with your provider clients. The point is decoupling model selection from request shape.
Step 5: Meter usage and enforce limits
Vision tokens are not text tokens. OpenAI bills image input per 512px tile; Anthropic and Gemini have their own accounting. Your gateway should parse the usage field from the provider response and normalize to a single image_tokens metric for per-token metering.
def extract_usage(resp: dict) -> dict:
# OpenAI response
if "usage" in resp:
return {
"prompt_tokens": resp["usage"]["prompt_tokens"],
"completion_tokens": resp["usage"]["completion_tokens"],
"image_tokens": resp["usage"].get("prompt_tokens_details", {}).get("image_tokens", 0)
}
# Anthropic / Gemini vary; map their fields similarly
return resp.get("usage", {})Log this per request ID. If you expose the gateway to multiple tenants, attribute cost by client_id before the provider response disappears.
Step 6: Verify end-to-end
Write a smoke test that sends a known image and asserts the model describes it. Use a public domain image URL.
import requests
def test_gateway_vision():
payload = {
"model": "gpt-4-vision-preview",
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": "What animal is in the image?"},
{"type": "image_url", "image_url": {"url": "https://upload.wikimedia.org/wikipedia/commons/thumb/3/3a/Cat03.jpg/120px-Cat03.jpg"}}
]
}]
}
r = requests.post("http://localhost:8080/v1/chat/completions", json=payload)
assert r.status_code == 200
data = r.json()
assert "cat" in data["choices"][0]["message"]["content"].lower()
assert data["usage"]["prompt_tokens"] > 0Run it against your gateway with each backend enabled. To verify fallback, blackhole the primary provider (e.g., iptables drop) and confirm the request still returns a valid completion from the secondary.
Success criteria: the same client payload produces correct answers across OpenAI, Anthropic, and Gemini without modifying the request shape. Your image inputs unified gateway is done when adding a new vision model requires only a new mapping function, not a new client integration.
Operational notes
Watch base64 inflation: it adds ~33% size. If you pass through URLs, ensure your providers can reach them; private S3 buckets need signed URLs generated at the edge. Strip EXIF metadata if privacy matters—do it in Step 2 before base64.
Multimodal prompts often interleave multiple images and text. The schema above already supports that; just append more blocks. Keep ordering intact because some models are order-sensitive.
Finally, cache normalized images keyed by content hash. Repeated requests for the same diagram shouldn’t refetch or re-encode. Forward the provider’s cache-control so you don’t hold blobs longer than needed.
That’s the full loop: normalize, map, route, meter, verify.