DEV Community: develpmilk

MiniMax H3 Video API: Async Polling, Input Validation, and Cost per Accepted Clip

develpmilk — Mon, 10 Aug 2026 09:01:21 +0000

MiniMax H3 is listed in the CometAPI catalog as minimax-h3. The provider's direct API uses MiniMax-H3. The distinction matters: a catalog route can have a different request contract and availability policy from the provider endpoint.

This guide uses MiniMax's documented API directly so the integration steps are explicit. Before production, verify the current CometAPI route, account access, region, and price for your project.

What the current contract says

MiniMax documents:

POST https://api.minimax.io/v2/video_generation to create a task
GET https://api.minimax.io/v2/query/video_generation/{task_id} to query it
Text, image, video, and audio inputs
4-15 second output at 24 FPS with 32 kHz stereo audio
Maximum 12 reference files, including up to nine images, three videos, and three audio clips
Audio references cannot be used alone

MiniMax's published output rates are $0.08 per second at 768p and $0.13 per second at 2K. A five-second output is therefore $0.40 or $0.65 for the output component, before reference, Context-IR, regeneration, and retry charges.

Install the small test client

python -m venv .venv
source .venv/bin/activate
python -m pip install requests
export MINIMAX_API_KEY="your-key"

Do not commit the key. The example below reads it from MINIMAX_API_KEY.

Create and poll a task

import os
import time
from decimal import Decimal

import requests


CREATE_URL = "https://api.minimax.io/v2/video_generation"
QUERY_URL = "https://api.minimax.io/v2/query/video_generation"


def create_task(prompt: str, resolution: str = "768P", duration: int = 5) -> dict:
    api_key = os.environ["MINIMAX_API_KEY"]
    if not prompt.strip():
        raise ValueError("prompt cannot be empty")
    if resolution not in {"768P", "2K"}:
        raise ValueError("resolution must be 768P or 2K")
    if duration not in range(4, 16):
        raise ValueError("duration must be between 4 and 15 seconds")

    payload = {
        "model": "MiniMax-H3",
        "content": [{"type": "text", "text": prompt}],
        "resolution": resolution,
        "duration": duration,
        "ratio": "16:9",
    }
    response = requests.post(
        CREATE_URL,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        json=payload,
        timeout=60,
    )
    response.raise_for_status()
    result = response.json()
    task_id = result.get("task_id") or result.get("data", {}).get("task_id")
    if not task_id:
        raise RuntimeError(f"create response did not include task_id: {result}")
    return result


def query_task(task_id: str) -> dict:
    api_key = os.environ["MINIMAX_API_KEY"]
    response = requests.get(
        f"{QUERY_URL}/{task_id}",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


def get_status(result: dict) -> str:
    value = result.get("status")
    if value is None:
        value = result.get("data", {}).get("status")
    return str(value or "UNKNOWN").upper()


def main() -> None:
    resolution = "768P"
    task = create_task(
        "A glass product on a desk in soft morning light; slow camera move, clean commercial style.",
        resolution=resolution,
        duration=5,
    )
    task_id = task.get("task_id") or task.get("data", {}).get("task_id")
    terminal = {"SUCCESS", "SUCCEEDED", "FAILED", "ERROR", "CANCELED", "CANCELLED"}

    while True:
        result = query_task(task_id)
        status = get_status(result)
        print({"task_id": task_id, "status": status})
        if status in terminal:
            print(result)
            break
        time.sleep(5)

    # Planning estimate only; reconcile with the usage object and invoice.
    rate = Decimal("0.08") if resolution == "768P" else Decimal("0.13")
    print("estimated output component USD:", rate * Decimal("5"))


if __name__ == "__main__":
    main()

The response keys can change with API revisions, so keep the raw create and query responses in your task log and compare them with the current API reference. The script intentionally prints the terminal response instead of assuming a particular output-URL field.

Validate multimodal content before sending

The most common first-request failure is an invalid reference combination. Validate these rules in your own request builder:

Input rule	Check
Text	Include a non-empty text item
Audio	Never send audio as the only reference
Total files	At most 12
Images	At most 9
Videos	At most 3; each 2-15 seconds; total 15 seconds
Audio clips	At most 3; each 2-15 seconds; total 15 seconds
Modes	Do not mix first/last-frame mode with reference-to-video mode
Request size	Keep the body within 64 MB; use public media URLs for large files

Failing locally protects a queue slot and makes errors easier to attribute.

Log accepted-output economics

Per task, store:

{
  "model": "MiniMax-H3",
  "route_id": "minimax-h3",
  "task_id": "provider-task-id",
  "resolution": "768P",
  "requested_seconds": 5,
  "input_seconds": 0,
  "output_seconds": 5,
  "input_image_count": 0,
  "latency_seconds": 0,
  "retry_count": 0,
  "charged_usd": 0,
  "review_result": "accepted"
}

The zero values are test-record defaults, not provider results. Keep charged_usd and the provider's usage object as the billing source of truth.

API versus open weights

H3-Base is downloadable under the MiniMax H3 Community License, but the default license territory excludes the United States, European Union, United Kingdom, and South Korea. Commercial products have attribution and safeguard requirements. H3-Context-IR is hosted and is not included in the initial open-weight release.

Use the hosted API for the fastest evaluation. Consider self-hosting only after checking territory, authorization, GPU capacity, storage, queueing, and whether a local H3-Base deployment actually meets your workflow needs.

CometAPI is useful as a unified route to check when a project already manages multiple models, but do not assume that a catalog announcement means every account or region is enabled. Probe minimax-h3, record the returned contract, and keep a fallback route.

Seedance 2.5 API Cost: Model ID, Usage, and a Safe Test Harness

develpmilk — Fri, 07 Aug 2026 09:01:59 +0000

BytePlus lists Seedance 2.5 as dreamina-seedance-2-5-260628. Its published five-second, 16:9 examples cost $0.514 at 480p and $1.156 at 720p without video input. The model is worth controlled testing, but an application should verify account, region, endpoint, parameters, and returned usage before treating it as a production dependency.

The official numbers

Item	Published value
Model ID	`dreamina-seedance-2-5-260628`
Rate without video input	$10.70 / 1M tokens
Rate with video input	$6.40 / 1M tokens
5s 480p, no video input	$0.514
5s 720p, no video input	$1.156

Video-input requests use a lower token rate, but input duration increases token consumption. For a five-second output, BytePlus lists $0.553-$2.152 at 480p and $1.244-$4.838 at 720p as input-video duration grows from the low range to 30 seconds.

These are official examples, not a universal per-second quote. Final usage should come from the response field documented by BytePlus: usage.completion_tokens.

Why I am not including a fake API call

ByteDance's launch material describes up to 30-second audio-video generation, references, extensions, and editing. The exact API contract and account rollout still need to be checked in the target project. Copying a Seedance 2.0 payload and changing only the model ID is not a safe migration strategy.

The following harness is provider-neutral. It lets a team calculate a planning budget and log acceptance economics before wiring an endpoint.

A runnable cost-per-accepted-clip calculator

from dataclasses import dataclass
from decimal import Decimal


@dataclass(frozen=True)
class Scenario:
    resolution: str
    price_per_successful_clip: Decimal
    acceptance_rate: Decimal
    review_cost_per_accepted_clip: Decimal = Decimal("0")

    def validate(self) -> None:
        if not (Decimal("0") < self.acceptance_rate <= Decimal("1")):
            raise ValueError("acceptance_rate must be greater than 0 and at most 1")
        if self.price_per_successful_clip < 0:
            raise ValueError("price_per_successful_clip cannot be negative")
        if self.review_cost_per_accepted_clip < 0:
            raise ValueError("review cost cannot be negative")

    def generation_cost_per_accepted(self) -> Decimal:
        self.validate()
        return self.price_per_successful_clip / self.acceptance_rate

    def total_cost_per_accepted(self) -> Decimal:
        return self.generation_cost_per_accepted() + self.review_cost_per_accepted_clip


def main() -> None:
    examples = [
        Scenario("480p", Decimal("0.514"), Decimal("0.70"), Decimal("0.30")),
        Scenario("720p", Decimal("1.156"), Decimal("0.70"), Decimal("0.30")),
    ]
    for item in examples:
        print(
            item.resolution,
            "generation/accepted=",
            item.generation_cost_per_accepted().quantize(Decimal("0.01")),
            "total/accepted=",
            item.total_cost_per_accepted().quantize(Decimal("0.01")),
        )


if __name__ == "__main__":
    main()

The 70% acceptance rate and $0.30 review cost are examples. Replace them with measured values. The script deliberately does not pretend that a creative rejection is a provider failure: a successfully generated but unusable clip is still part of the production cost.

What to log when the endpoint is enabled

Use one record per generation:

{
  "run_id": "eval-001",
  "model": "dreamina-seedance-2-5-260628",
  "resolution": "720p",
  "input_video_seconds": 0,
  "output_video_seconds": 5,
  "completion_tokens": 0,
  "terminal_state": "succeeded",
  "review_result": "accepted",
  "charged_usd": 0
}

The zero values are placeholders for your provider response. Keep model usage, charge, terminal state, and creative review separate so a dashboard can distinguish moderation failures, infrastructure failures, and quality rejection.

A safe evaluation order

Probe the model ID in the intended account and region.
Reproduce the 480p and 720p five-second examples.
Add a short reference video, then a long reference video.
Test complex motion and multiple interacting subjects.
Measure accepted clips, revision time, queue time, and returned usage.
Canary only after a tested fallback is available.

CometAPI can be useful as one operational layer when a team is comparing several video routes, but its current Seedance 2.5 catalog availability must be checked at publish time. A gateway does not remove provider-specific parameter or billing differences.

Run Gemma 4 26B on an 8GB Mac with TurboFieldfare

develpmilk — Fri, 31 Jul 2026 08:56:12 +0000

Run Gemma 4 26B on an 8GB Mac with TurboFieldfare

TurboFieldfare can run the text-only Gemma 4 26B A4B model on an 8GB Apple Silicon Mac with roughly a 1.9–2.1GB process footprint. It achieves that result by keeping a 4K FP16 KV cache and shared core resident while streaming routed experts from a 14.3GB SSD installation. This guide sets up the runtime and measures the configuration honestly.

The commands and API contract below come from the TurboFieldfare project documentation. This draft was not executed on Apple Silicon in the current authoring environment. Run the validation steps on the target Mac and add the real output before publishing.

What the 2GB number means

Gemma 4 26B A4B has approximately 25.2B total parameters and approximately 3.8B active parameters per token. Because each token uses eight of 128 routed experts plus a shared expert, TurboFieldfare can keep a bounded expert cache and stream missing experts from SSD.

The reference configuration is:

Item	Requirement or reported value
Hardware	Apple Silicon Mac; 8GB M2 MacBook Air is the validated entry point
OS	macOS 26+ with Metal 4
Toolchain	Xcode 26 and Swift 6.2+
Process footprint	Approximately 1.9–2.1GB in published runs
Installed model	Approximately 14.3GB
Initial transfer	Approximately 15GB
Headline KV context	4K FP16
Local modalities	Text only

The project reports 5.10–6.30 decode tok/s on an 8GB M2 MacBook Air and 31–35 tok/s on a 24GB M5 Pro. These are project measurements, not Google benchmarks, and they exclude installation, model loading, and prompt prefill.

Step 1: Check the Mac

Confirm the architecture, macOS version, Swift toolchain, free storage, and memory pressure:

uname -m
sw_vers
swift --version
df -h .
memory_pressure -Q

Expected prerequisites:

uname -m returns arm64.
macOS is version 26 or later.
Swift is 6.2 or later.
At least 16GB of free storage is available for the initial installation; leave additional headroom for builds and normal system operation.
Memory pressure is healthy before loading the model.

Close other local-model runtimes and memory-heavy applications. TurboFieldfare’s documentation recommends running one model-owning process at a time.

Step 2: Build TurboFieldfare

Clone the repository and build the release package:

git clone https://github.com/drumih/turbo-fieldfare.git
cd turbo-fieldfare
swift build -c release

Launch the native Mac app:

.build/release/TurboFieldfareMac

In the app:

Choose Download.
Wait for the roughly 15GB transfer and repack.
Confirm that the completed model occupies about 14.3GB.
Choose Load Model.
Start with the default 4K context and 16 expert-cache slots.
Enter a short prompt and generate a response.

The installer streams remote byte ranges into the .gturbo layout and validates the completed manifest and file hashes. It does not need to materialize a second full checkpoint.

Step 3: Verify an installation from the CLI

If you prefer the command line, install or resume the model with:

swift run -c release TurboFieldfareRepack \
  --output scratch/gemma4.gturbo \
  --overwrite \
  --resume

Verify the completed installation without loading the model:

swift run -c release TurboFieldfareRepack \
  --verify-install \
  --input-gturbo scratch/gemma4.gturbo

Do not treat a partial directory as a model installation. The runtime accepts a completed .gturbo directory with a final manifest.json.

Step 4: Run a deterministic CLI smoke test

Use a short raw completion with greedy decoding:

swift run -c release TurboFieldfareCLI \
  --model scratch/gemma4.gturbo \
  --prompt "The capital of France is" \
  --max-new 64 \
  --temperature 0

Generated text goes to standard output, while timing statistics go to standard error. Record the commit, hardware model, OS, Swift version, prompt length, generated tokens, context setting, expert-cache slots, TTFT, decode rate, and peak memory.

One short completion is a smoke test, not a benchmark.

Step 5: Start the local Chat Completions server

First check that another model-owning process is not running:

pgrep -fl 'TurboFieldfareServer|TurboFieldfareMac|TurboFieldfareDecodeService|TurboFieldfareCLI|mlx_lm|mlx-lm'

If the command prints a live model process, stop it normally before starting the server.

Build and start the loopback server:

swift build -c release --product TurboFieldfareServer
.build/release/TurboFieldfareServer \
  --model scratch/gemma4.gturbo \
  --port 8080 \
  --max-context 4096

Using --max-context 4096 keeps this tutorial aligned with the 4K context behind the approximately 2GB headline. The project’s documented server example uses 16K, but larger FP16 KV caches consume more memory.

Wait for TurboFieldfareServer ready, then use another terminal for the client checks.

Step 6: Check health and the model ID

curl --silent --show-error http://127.0.0.1:8080/health
curl --silent --show-error http://127.0.0.1:8080/v1/models

Send a deterministic request:

curl --silent --show-error http://127.0.0.1:8080/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "gemma-4-26b-a4b-it",
    "messages": [
      {"role": "user", "content": "Reply with exactly READY."}
    ],
    "temperature": 0,
    "max_completion_tokens": 16
  }'

The local model ID matches the official hosted model ID, but the interfaces are not feature-equivalent. TurboFieldfare’s server supports Chat Completions, streaming, function-tool declarations, and single-prefix reuse. It does not support the Responses API, embeddings, multimodal input, structured output, batching, log probabilities, or remote model switching.

Step 7: Measure TTFT and decode throughput

Create a separate Python environment and install the OpenAI client:

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip 'openai>=1.60,<3'

Save this as measure_local_gemma.py:

from __future__ import annotations

import json
import time
from dataclasses import asdict, dataclass

from openai import OpenAI


@dataclass
class Result:
    time_to_first_text_seconds: float | None
    total_seconds: float
    completion_tokens: int | None
    measured_decode_tokens_per_second: float | None
    text: str


client = OpenAI(
    base_url="http://127.0.0.1:8080/v1",
    api_key="local",
)

started = time.perf_counter()
first_text_at: float | None = None
parts: list[str] = []
completion_tokens: int | None = None

stream = client.chat.completions.create(
    model="gemma-4-26b-a4b-it",
    messages=[
        {
            "role": "user",
            "content": (
                "Explain mixture-of-experts routing in 150 to 200 words. "
                "Use plain language and no bullet points."
            ),
        }
    ],
    temperature=0,
    max_completion_tokens=256,
    stream=True,
    stream_options={"include_usage": True},
)

for chunk in stream:
    now = time.perf_counter()
    if chunk.choices:
        content = chunk.choices[0].delta.content or ""
        if content:
            if first_text_at is None:
                first_text_at = now
            parts.append(content)

    if chunk.usage is not None:
        completion_tokens = chunk.usage.completion_tokens

finished = time.perf_counter()
text = "".join(parts)
ttft = None if first_text_at is None else first_text_at - started
decode_seconds = None if first_text_at is None else finished - first_text_at
decode_rate = (
    None
    if not completion_tokens or not decode_seconds or decode_seconds <= 0
    else completion_tokens / decode_seconds
)

result = Result(
    time_to_first_text_seconds=ttft,
    total_seconds=finished - started,
    completion_tokens=completion_tokens,
    measured_decode_tokens_per_second=decode_rate,
    text=text,
)

print(json.dumps(asdict(result), indent=2, ensure_ascii=False))

Run it while the local server is active:

python measure_local_gemma.py | tee local-gemma-result.json

The client-side decode calculation is approximate because the final usage chunk and network buffering affect the clock. Use the server’s timing output as the primary decode figure and the script as an end-to-end client observation.

Step 8: Create a reproducible result table

Do not publish one run as a hardware guarantee. Run a warmup, then repeat fixed prompts in fresh processes and record every condition.

Field	Result
Git commit	`[commit]`
Hardware	`[Mac model / RAM]`
macOS / Swift	`[versions]`
Prompt tokens	`[value]`
Generated tokens	`[value]`
Context / expert cache	`4K / 16 slots`
File-cache state	`[cold / warm / uncontrolled]`
TTFT	`[seconds]`
Server decode rate	`[tok/s]`
Peak RSS / footprint	`[MiB]`
Output accepted?	`[yes/no + rule]`

The project’s own benchmark protocol uses fixed prompts and requires coherent, non-repeating output. Follow that methodology before describing a result as comparable.

Keep the server on loopback

The local server binds to 127.0.0.1 and has no authentication or TLS. The project explicitly says not to expose it through a proxy or tunnel.

For local tools, keep the client on the same Mac. For remote or multi-user production access, do not simply change the bind address. Design a separate serving layer with authentication, authorization, TLS, request limits, queueing, process supervision, readiness checks, memory-pressure and SSD metrics, privacy-aware logging, overload handling, and failover.

Model-generated tool calls also require client-side approval. The server returns a proposal; the application decides whether the tool may execute.

Local versus official API

Google’s official hosted model ID is also gemma-4-26b-a4b-it. The Gemini API provides text and image input, managed infrastructure, function calling, and model context up to 256K.

Google currently lists Gemma 4 on its API free tier and does not list a paid Gemma 4 tier. Recheck the pricing page and data-use terms before production use. Free-tier content may be used to improve Google products.

As of July 30, 2026, CometAPI does not list Gemma 4 26B. Do not point this tutorial’s model ID at CometAPI. Use Google’s official API for hosted Gemma 4 access unless another provider explicitly lists the exact model.

What should you compare?

Use the same prompts and acceptance criteria for local and hosted routes. Record:

time to first token;
total response time;
prompt prefill;
decode throughput;
peak memory and SSD reads;
context length;
queue and concurrency behavior;
structured-output validity;
tool-call validity;
accepted-task rate;
retries and human correction;
total operating cost.

Resident RAM is one metric. It is not the deployment decision.

Audit Your Codebase Before Migrating to Stateless MCP

develpmilk — Thu, 30 Jul 2026 07:32:23 +0000

MCP 2026-07-28 removes protocol-level sessions, the required initialization handshake, and Mcp-Session-Id. Before deleting those code paths, find the application state they were hiding. This tutorial builds a dependency scanner, classifies each match by migration risk, and turns the results into a staged rollout checklist.

This is a static audit, not proof that a deployment is compatible. It finds likely legacy assumptions; you still need runtime tests across clients, servers, gateways, and replicas.

What changed?

The official MCP 2026-07-28 changelog makes several changes that affect production architecture:

Mcp-Session-Id is removed from Streamable HTTP.
initialize and notifications/initialized are no longer required.
Protocol version and client capabilities travel in each request’s _meta.
Servers implement server/discover for version and capability discovery.
Multi Round-Trip Requests replace server-initiated requests.
Mcp-Method and Mcp-Name support HTTP routing and policy.
Cacheable list/read results include ttlMs and cacheScope.
Roots, Sampling, Logging, HTTP+SSE, selected includeContext values, and Dynamic Client Registration are deprecated rather than immediately removed.

The risky misunderstanding is that “stateless protocol” means “the application has no state.” A workspace, durable task, OAuth credential, or pending approval can still span calls. The dependency must simply be explicit and available to any compatible worker that receives the next request.

Step 1: Build a zero-dependency scanner

The following Node.js script searches a repository for older MCP assumptions. It uses only Node’s standard library and produces both a readable report and an optional JSON report for CI.

Create audit-mcp-2026.mjs:

#!/usr/bin/env node

import { promises as fs } from "node:fs";
import path from "node:path";
import process from "node:process";

const rules = [
  {
    id: "protocol-session",
    risk: "high",
    pattern: /Mcp-Session-Id|ctx\.sessionId|extra\.sessionId|sessionId/g,
    action: "Map the state behind the session, then use an explicit handle or shared storage.",
  },
  {
    id: "initialization-gate",
    risk: "medium",
    pattern: /notifications\/initialized|\binitialize\b/g,
    action: "Remove gates for modern requests and add server/discover plus version negotiation.",
  },
  {
    id: "sticky-routing",
    risk: "high",
    pattern: /sticky[_-]?session|sessionAffinity|affinity:\s*(?:ClientIP|cookie)/gi,
    action: "Test consecutive calls on different replicas before removing affinity.",
  },
  {
    id: "server-initiated-request",
    risk: "high",
    pattern: /elicitation\/create|sampling\/createMessage|roots\/list/g,
    action: "Move interaction to MRTR and make retried side effects idempotent.",
  },
  {
    id: "legacy-resource-subscription",
    risk: "medium",
    pattern: /resources\/(?:subscribe|unsubscribe)|Last-Event-ID/g,
    action: "Plan subscriptions/listen and test broken-stream behavior.",
  },
  {
    id: "legacy-task",
    risk: "medium",
    pattern: /tasks\/(?:result|list)/g,
    action: "Adopt the io.modelcontextprotocol/tasks extension contract.",
  },
  {
    id: "legacy-logging",
    risk: "low",
    pattern: /logging\/setLevel|notifications\/message/g,
    action: "Review per-request log metadata and production observability.",
  },
];

const ignoredDirectories = new Set([
  ".git",
  "node_modules",
  "dist",
  "build",
  "coverage",
  "vendor",
]);

const textExtensions = new Set([
  ".cjs", ".conf", ".go", ".java", ".js", ".json", ".jsx",
  ".md", ".mjs", ".py", ".rb", ".rs", ".sh", ".toml",
  ".ts", ".tsx", ".yaml", ".yml",
]);

function parseArguments(argv) {
  const result = { root: ".", json: false };

  for (const argument of argv) {
    if (argument === "--json") result.json = true;
    else if (argument.startsWith("-")) {
      throw new Error(`Unknown option: ${argument}`);
    } else result.root = argument;
  }

  return result;
}

async function* walk(directory) {
  const entries = await fs.readdir(directory, { withFileTypes: true });

  for (const entry of entries) {
    if (ignoredDirectories.has(entry.name)) continue;

    const fullPath = path.join(directory, entry.name);
    if (entry.isDirectory()) yield* walk(fullPath);
    else if (entry.isFile() && textExtensions.has(path.extname(entry.name))) {
      yield fullPath;
    }
  }
}

function scanLine(line, file, lineNumber) {
  const findings = [];

  for (const rule of rules) {
    rule.pattern.lastIndex = 0;
    const matches = [...line.matchAll(rule.pattern)];
    for (const match of matches) {
      findings.push({
        file,
        line: lineNumber,
        column: (match.index ?? 0) + 1,
        match: match[0],
        rule: rule.id,
        risk: rule.risk,
        action: rule.action,
      });
    }
  }

  return findings;
}

async function audit(root) {
  const absoluteRoot = path.resolve(root);
  const findings = [];

  for await (const file of walk(absoluteRoot)) {
    const content = await fs.readFile(file, "utf8");
    const lines = content.split(/\r?\n/);

    lines.forEach((line, index) => {
      findings.push(
        ...scanLine(line, path.relative(absoluteRoot, file), index + 1),
      );
    });
  }

  return findings;
}

function printText(findings) {
  if (findings.length === 0) {
    console.log("No known legacy MCP patterns found. Runtime testing is still required.");
    return;
  }

  const weight = { high: 0, medium: 1, low: 2 };
  findings.sort((a, b) => weight[a.risk] - weight[b.risk]);

  for (const finding of findings) {
    console.log(
      `[${finding.risk.toUpperCase()}] ${finding.file}:${finding.line}:${finding.column} ` +
      `${finding.rule} (${finding.match})`,
    );
    console.log(`  ${finding.action}`);
  }

  const counts = findings.reduce(
    (summary, item) => ({ ...summary, [item.risk]: summary[item.risk] + 1 }),
    { high: 0, medium: 0, low: 0 },
  );
  console.log(`\nSummary: ${counts.high} high, ${counts.medium} medium, ${counts.low} low`);
}

async function main() {
  const options = parseArguments(process.argv.slice(2));
  const findings = await audit(options.root);

  if (options.json) console.log(JSON.stringify({ findings }, null, 2));
  else printText(findings);

  process.exitCode = findings.some((item) => item.risk === "high") ? 2 : 0;
}

main().catch((error) => {
  console.error(error instanceof Error ? error.message : error);
  process.exitCode = 1;
});

Step 2: Run the audit

The script requires Node.js 20 or later. Run it from any directory and pass the repository path:

node audit-mcp-2026.mjs /path/to/your/repository

Use JSON output in CI or to create migration tickets:

node audit-mcp-2026.mjs /path/to/your/repository --json > mcp-audit.json

The exit codes are:

Code	Meaning
`0`	No high-risk static match was found
`1`	The scanner itself failed
`2`	At least one high-risk pattern was found

Do not fail a production release merely because a string appears in documentation or a compatibility test. Treat each finding as a review item. The script intentionally favors visibility over certainty.

Step 3: Classify every match by state ownership

A grep result is not a migration plan. For each high-risk match, record:

What state exists? Workspace, credential, user preference, approval, task, cursor, or cache entry.
Who owns it? User, tenant, request, tool, server, or deployment.
How long must it survive? One retry, several calls, a process restart, or days.
Who may read or change it? The authorization boundary belongs in the design.
Can two workers process it safely? If not, the migration is incomplete.
Can a retry duplicate a side effect? Add staging and idempotency before enabling MRTR.

The common replacement patterns are explicit handles and shared storage. An explicit handle makes the dependency visible in the tool schema. Shared storage makes the state accessible across replicas and restarts. Long-running work belongs in a durable task system rather than the memory of a request worker.

Step 4: Test the new failure model

Static scanning cannot detect whether a retry is safe or whether an authorization boundary is correct. Add integration tests that deliberately exercise the new architecture.

Cross-replica state test

Send a request that creates a workspace or task.
Force the next request to a different worker.
Pass only the documented handle and authenticated identity.
Confirm that the second worker can continue the operation.
Restart the first worker and repeat.

MRTR retry test

Start a destructive or billable operation.
Confirm that the first response is input_required before the side effect.
Tamper with and expire requestState; both cases must fail closed.
Retry twice with the same business idempotency key.
Confirm that exactly one side effect occurs.

Gateway mismatch test

Send a request whose Mcp-Name identifies a low-risk tool while the JSON-RPC body calls a restricted tool. The request should be rejected and the mismatch should appear in security telemetry.

Cache isolation test

Fetch a private tool or resource list as tenant A, then request the same method as tenant B. The second tenant must never receive A’s entry, regardless of the remaining TTL.

Step 5: Canary the protocol, not just the package

A safe rollout keeps modern and legacy behavior observable at the same time. Test these combinations explicitly:

modern client + modern server    -> MCP 2026-07-28
modern client + legacy server    -> tested fallback, if supported
legacy client + dual server      -> legacy path
unsupported combination         -> clear protocol-version error

Track protocol version, discovery success, fallback rate, missing or invalid headers, header/body mismatches, MRTR completion and timeout, request-state verification, duplicate-operation prevention, cache hits by scope, OAuth issuer failures, legacy HTTP+SSE traffic, and deprecated-method traffic.

Retire a compatibility path because the telemetry says it is unused—not because a calendar says the upgrade should be complete.

Where does a model gateway fit?

MCP and a model API solve different problems. MCP connects an agent to tools, resources, approvals, and tasks. A model API connects an application to inference providers, credentials, usage, and billing.

When migrating away from the deprecated Sampling capability, an application can call a model provider directly or use a unified model layer such as CometAPI. That model layer does not replace MCP authorization, tool contracts, application state, or MRTR handling. Keeping the interfaces separate simply allows the protocol and inference providers to evolve independently.

Migration checklist

[ ] Run the static audit across clients, servers, gateways, and deployment config.
[ ] Map every session lookup to its actual business state.
[ ] Implement server/discover and version negotiation.
[ ] Include protocol metadata on each modern request.
[ ] Replace hidden state with handles, shared storage, or durable tasks.
[ ] Secure and expire MRTR requestState.
[ ] Make side effects idempotent across retries.
[ ] Validate MCP routing headers against the JSON-RPC body.
[ ] Partition caches by authorization boundary.
[ ] Test OAuth issuer changes and credential storage.
[ ] Canary modern and legacy traffic with rollback.
[ ] Retire deprecated behavior only after measuring its use.

Final takeaway

MCP 2026-07-28 removes sessions from the protocol contract, not from every application requirement. The migration succeeds when any compatible worker can process the next request from explicit inputs, every retry is authorized and idempotent, and the legacy path can be retired from evidence.

Kimi K3 Self-Hosting Cost Calculator: API vs 8 GPUs

develpmilk — Tue, 28 Jul 2026 08:13:25 +0000

Kimi K3 Self-Hosting Cost Calculator: API vs 8 GPUs

Kimi K3 can be self-hosted, but its public checkpoint is about 1.56 TB and the current vLLM recipe starts at eight high-end GPUs. For most teams, the useful comparison is not token price versus GPU rent. It is API cost per accepted task versus the complete cost of a reliable distributed deployment.

Why a calculator is more useful than a yes-or-no answer

Kimi K3 is an open-weight, 2.8-trillion-parameter Mixture-of-Experts model. It activates 104 billion parameters per token, selecting 16 of 896 routed experts, and supports a context window of up to 1,048,576 tokens.

Those numbers make self-hosting possible, but not simple.

The current vLLM recipe lists a starting topology of eight NVIDIA GB300 GPUs or eight AMD MI355X/MI350X-class GPUs. It recommends multi-node infrastructure for real production traffic. Moonshot's launch material recommends 64 or more accelerators for higher inference efficiency.

This leaves most teams with two realistic choices:

Start with a hosted API and pay for measured usage.
Operate a distributed GPU cluster and pay for capacity whether it is busy or idle.

The correct choice changes with traffic, cache reuse, output length, accepted-task rate, engineering cost, and utilization. The Python script below keeps those assumptions explicit.

What the cost model includes

The hosted side includes:

cache-hit input tokens;
cache-miss input tokens;
output tokens;
requests that are billed but fail application acceptance;
the number of requests per month.

The self-hosted side includes:

cluster cost per hour;
730 provisioned hours per month;
platform and inference engineering;
networking and storage;
observability and security;
redundancy and idle headroom.

The output reports both total monthly cost and cost per accepted task.

Complete Python calculator

This example requires Python 3.11 or later and no third-party packages.

from dataclasses import dataclass


HOURS_PER_MONTH = 730
MILLION = 1_000_000


@dataclass(frozen=True)
class Workload:
    requests_per_month: int
    input_tokens_per_request: int
    output_tokens_per_request: int
    cache_hit_ratio: float
    accepted_task_ratio: float

    def validate(self) -> None:
        if self.requests_per_month <= 0:
            raise ValueError("requests_per_month must be positive")
        if self.input_tokens_per_request < 0:
            raise ValueError("input_tokens_per_request cannot be negative")
        if self.output_tokens_per_request < 0:
            raise ValueError("output_tokens_per_request cannot be negative")
        if not 0 <= self.cache_hit_ratio <= 1:
            raise ValueError("cache_hit_ratio must be between 0 and 1")
        if not 0 < self.accepted_task_ratio <= 1:
            raise ValueError("accepted_task_ratio must be above 0 and at most 1")


@dataclass(frozen=True)
class ApiPricing:
    name: str
    cache_hit_input_per_million: float
    cache_miss_input_per_million: float
    output_per_million: float


@dataclass(frozen=True)
class SelfHostedCost:
    cluster_hourly_rate: float
    engineering_per_month: float
    networking_and_storage_per_month: float
    observability_and_security_per_month: float
    redundancy_and_headroom_per_month: float

    def monthly_total(self) -> float:
        return (
            self.cluster_hourly_rate * HOURS_PER_MONTH
            + self.engineering_per_month
            + self.networking_and_storage_per_month
            + self.observability_and_security_per_month
            + self.redundancy_and_headroom_per_month
        )


def monthly_api_cost(workload: Workload, pricing: ApiPricing) -> float:
    workload.validate()

    total_input = (
        workload.requests_per_month * workload.input_tokens_per_request
    )
    total_output = (
        workload.requests_per_month * workload.output_tokens_per_request
    )

    cache_hit_input = total_input * workload.cache_hit_ratio
    cache_miss_input = total_input - cache_hit_input

    return (
        cache_hit_input / MILLION * pricing.cache_hit_input_per_million
        + cache_miss_input / MILLION * pricing.cache_miss_input_per_million
        + total_output / MILLION * pricing.output_per_million
    )


def accepted_tasks(workload: Workload) -> float:
    return workload.requests_per_month * workload.accepted_task_ratio


def print_api_report(workload: Workload, pricing: ApiPricing) -> float:
    total = monthly_api_cost(workload, pricing)
    accepted = accepted_tasks(workload)
    cost_per_accepted = total / accepted

    print(f"{pricing.name}:")
    print(f"  monthly API cost:       ${total:,.2f}")
    print(f"  accepted tasks:         {accepted:,.0f}")
    print(f"  cost / accepted task:   ${cost_per_accepted:,.4f}")
    return cost_per_accepted


def main() -> None:
    workload = Workload(
        requests_per_month=10_000,
        input_tokens_per_request=300_000,
        output_tokens_per_request=30_000,
        cache_hit_ratio=0.50,
        accepted_task_ratio=0.95,
    )

    moonshot = ApiPricing(
        name="Moonshot official API",
        cache_hit_input_per_million=0.30,
        cache_miss_input_per_million=3.00,
        output_per_million=15.00,
    )

    # The July 28, 2026 public CometAPI rate in the source package has
    # one input rate rather than separate hit/miss rates.
    cometapi = ApiPricing(
        name="CometAPI",
        cache_hit_input_per_million=2.40,
        cache_miss_input_per_million=2.40,
        output_per_million=12.00,
    )

    self_hosted = SelfHostedCost(
        cluster_hourly_rate=80.00,
        engineering_per_month=20_000.00,
        networking_and_storage_per_month=5_000.00,
        observability_and_security_per_month=3_000.00,
        redundancy_and_headroom_per_month=7_000.00,
    )

    print("Kimi K3 monthly cost model\n")
    moonshot_cost_per_task = print_api_report(workload, moonshot)
    print()
    cometapi_cost_per_task = print_api_report(workload, cometapi)
    print()

    self_hosted_total = self_hosted.monthly_total()
    print("Self-hosted scenario:")
    print(f"  monthly total:          ${self_hosted_total:,.2f}")
    print(
        "  break-even tasks vs Moonshot: "
        f"{self_hosted_total / moonshot_cost_per_task:,.0f}"
    )
    print(
        "  break-even tasks vs CometAPI: "
        f"{self_hosted_total / cometapi_cost_per_task:,.0f}"
    )


if __name__ == "__main__":
    main()

Run it with:

python3 kimi_k3_tco.py

For the supplied scenario, the script produces approximately:

Kimi K3 monthly cost model

Moonshot official API:
  monthly API cost:       $9,450.00
  accepted tasks:         9,500
  cost / accepted task:   $0.9947

CometAPI:
  monthly API cost:       $10,800.00
  accepted tasks:         9,500
  cost / accepted task:   $1.1368

Self-hosted scenario:
  monthly total:          $93,400.00
  break-even tasks vs Moonshot: 93,894
  break-even tasks vs CometAPI: 82,157

Why the cheaper API changes with cache behavior

Moonshot's published Kimi K3 pricing in July 2026 is:

Usage	Price per 1M tokens
Cache-hit input	$0.30
Cache-miss input	$3.00
Output	$15.00

The source package records CometAPI at $2.40 per million input tokens and $12.00 per million output tokens on July 28, 2026.

CometAPI is below the official cache-miss input and output rates in this snapshot. It is not automatically cheaper when a workload has strong prefix-cache reuse, because Moonshot's official cache-hit input rate is much lower.

This is why the calculator accepts a cache-hit ratio. Change cache_hit_ratio from 0.50 to the value measured in your own application rather than treating one price row as the answer.

Before using the CometAPI price in a decision, check the live Kimi K3 model page. Pricing and availability are time-sensitive.

What the eight-GPU scenario still leaves out

An eight-GPU estimate is a planning floor, not proof of production readiness.

Kimi K3 routes tokens across experts, which creates all-to-all communication. The same nominal GPU count can produce different throughput depending on NVLink, RDMA, network topology, expert parallelism, prefill/decode separation, and concurrency.

A serious self-hosting test should also include:

model download and startup time;
storage replication and recovery;
KV-cache memory for the context lengths actually used;
deployment rollback and weight upgrades;
node failure and degraded-cluster behavior;
tool-call validation and retries;
capacity reserved for traffic bursts;
engineering on-call coverage.

Do not use the theoretical maximum context by default. Long prompts increase prefill work, KV-cache demand, latency, and the amount of capacity unavailable to concurrent requests.

When self-hosting becomes credible

Self-hosting is worth a load test when at least one of these conditions is true:

Demand is sustained and predictable enough to keep the cluster productively busy.
The organization already operates distributed MoE inference and high-bandwidth GPU networking.
Data-path control or a dedicated environment is a hard requirement.
The team needs direct control over model versions, runtime configuration, or fine-tuned weights.
Measured hosted spend approaches the complete cost of an equivalent reliable internal service.

For new or variable traffic, an API is usually the safer baseline. It gives the team real token, cache, latency, retry, and acceptance data before a hardware commitment is made.

Key takeaways

Kimi K3's open weights create deployment freedom, not a low-cost local setup.
The current serving floor starts at eight high-end GPUs; efficient production may require a much larger topology.
Prefix-cache reuse can change which hosted route is less expensive.
Compare cost per accepted task, not raw token price or GPU rent.
Buy infrastructure only after a production-shaped load test supports the complete TCO case.

How to Benchmark Speech-to-Text APIs Without Fooling Yourself

develpmilk — Mon, 27 Jul 2026 09:51:48 +0000

The best speech-to-text API is the route that meets your workload's accuracy, latency, reliability, data, and cost requirements. A useful benchmark must use the same audio and scoring rules across providers and calculate cost per accepted transcript—not merely compare advertised price per minute.

This tutorial builds the local scoring layer for that benchmark. It deliberately avoids fake provider wrappers: each speech API has its own current authentication, upload, streaming, and asynchronous-job contract. Export every provider result into one CSV, then score them consistently.

What the benchmark measures

The evaluation tracks six production metrics:

Word error rate (WER) under one normalization policy.
p95 latency, because averages hide slow requests.
Failure rate, including empty output and transport errors.
Acceptance rate based on the product's actual criteria.
Human review time per audio hour.
Cost per accepted transcript, including review labor.

Public list prices are useful for shortlisting. They do not replace this benchmark. In the July 27, 2026 snapshot, recorded-audio routes range from AssemblyAI Universal-2 at $0.15/hour to Google standard recognition at $0.96/hour for the first 500,000 monthly minutes. Those routes differ in urgency, channel billing, features, and operational fit.

Environment

Python 3.11 or later
jiwer for word error rate
CSV exports from your provider integrations

Install the dependency:

python -m pip install jiwer

Define one result schema

Create stt_results.csv with this header:

provider,clip_id,audio_seconds,channels,api_cost_usd,latency_ms,review_minutes,accepted,status,reference,hypothesis

Each row represents one provider attempt on one audio clip.

api_cost_usd must include channel and feature charges for that request.
latency_ms means upload-to-final for batch jobs or utterance-end-to-final for live jobs.
accepted is true only when the transcript passes the product contract.
status should distinguish ok, timeout, empty, rate_limited, and other failures.
reference and hypothesis contain the approved transcript and provider output.

Do not put batch and live routes in the same report. Their latency and product behavior are not comparable.

Use synthetic rows only to validate the script

The following fixture is not a provider benchmark. Replace it with real results before drawing any conclusion.

provider,clip_id,audio_seconds,channels,api_cost_usd,latency_ms,review_minutes,accepted,status,reference,hypothesis
route_a,fixture_01,30,1,0.010,900,1.0,true,ok,"the quick brown fox","the quick brown fox"
route_b,fixture_01,30,1,0.020,600,0.5,true,ok,"the quick brown fox","the quick brown box"
route_a,fixture_02,45,2,0.025,1500,3.0,false,ok,"call account number five","call a count number five"
route_b,fixture_02,45,2,0.040,800,1.0,true,ok,"call account number five","call account number five"

Add the scoring script

Save this as score_stt.py:

from __future__ import annotations

import argparse
import csv
import math
from collections import defaultdict
from dataclasses import dataclass

from jiwer import wer


@dataclass
class Result:
    provider: str
    clip_id: str
    audio_seconds: float
    channels: int
    api_cost_usd: float
    latency_ms: float
    review_minutes: float
    accepted: bool
    status: str
    reference: str
    hypothesis: str


def parse_bool(value: str) -> bool:
    return value.strip().lower() in {"1", "true", "yes", "y"}


def load_results(path: str) -> list[Result]:
    rows: list[Result] = []
    with open(path, newline="", encoding="utf-8") as handle:
        for row in csv.DictReader(handle):
            rows.append(
                Result(
                    provider=row["provider"],
                    clip_id=row["clip_id"],
                    audio_seconds=float(row["audio_seconds"]),
                    channels=int(row["channels"]),
                    api_cost_usd=float(row["api_cost_usd"]),
                    latency_ms=float(row["latency_ms"]),
                    review_minutes=float(row["review_minutes"]),
                    accepted=parse_bool(row["accepted"]),
                    status=row["status"],
                    reference=row["reference"],
                    hypothesis=row["hypothesis"],
                )
            )
    return rows


def percentile(values: list[float], p: float) -> float:
    if not values:
        return math.nan
    ordered = sorted(values)
    index = (len(ordered) - 1) * p
    lower = math.floor(index)
    upper = math.ceil(index)
    if lower == upper:
        return ordered[lower]
    fraction = index - lower
    return ordered[lower] + (ordered[upper] - ordered[lower]) * fraction


def summarize(rows: list[Result], reviewer_hourly_usd: float) -> None:
    grouped: dict[str, list[Result]] = defaultdict(list)
    for row in rows:
        grouped[row.provider].append(row)

    header = (
        "provider",
        "runs",
        "wer",
        "p95_ms",
        "fail_%",
        "accept_%",
        "review_min/audio_hr",
        "cost/accepted_usd",
    )
    print("\t".join(header))

    for provider, provider_rows in sorted(grouped.items()):
        completed = [row for row in provider_rows if row.status == "ok"]
        accepted = [row for row in provider_rows if row.accepted]
        failures = [row for row in provider_rows if row.status != "ok"]

        references = [row.reference for row in completed]
        hypotheses = [row.hypothesis for row in completed]
        provider_wer = wer(references, hypotheses) if completed else math.nan

        total_audio_hours = sum(row.audio_seconds for row in provider_rows) / 3600
        total_review_minutes = sum(row.review_minutes for row in provider_rows)
        review_per_audio_hour = (
            total_review_minutes / total_audio_hours if total_audio_hours else math.nan
        )

        api_cost = sum(row.api_cost_usd for row in provider_rows)
        review_cost = total_review_minutes / 60 * reviewer_hourly_usd
        cost_per_accepted = (
            (api_cost + review_cost) / len(accepted) if accepted else math.inf
        )

        values = (
            provider,
            str(len(provider_rows)),
            f"{provider_wer:.4f}",
            f"{percentile([row.latency_ms for row in completed], 0.95):.0f}",
            f"{100 * len(failures) / len(provider_rows):.1f}",
            f"{100 * len(accepted) / len(provider_rows):.1f}",
            f"{review_per_audio_hour:.2f}",
            f"{cost_per_accepted:.4f}",
        )
        print("\t".join(values))


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("csv_path")
    parser.add_argument("--reviewer-hourly-usd", type=float, default=30.0)
    args = parser.parse_args()

    summarize(load_results(args.csv_path), args.reviewer_hourly_usd)

Run it:

python score_stt.py stt_results.csv --reviewer-hourly-usd 30

The script prints tab-separated results that can be pasted into a spreadsheet. It does not hide failed requests or divide cost only by successful API responses; it uses the stricter accepted field.

Build a production-like audio set

A benchmark should resemble the traffic the product will receive. Start with 60 to 80 licensed clips:

20 clean meetings with names and domain terms
20 noisy calls with interruptions, crosstalk, and hold music
20 accented or multilingual clips with real code switching
10 long-form files
10 short live utterances with silence, false starts, and barge-in

Use one reference-transcript and normalization policy for every route. Otherwise WER becomes a comparison of annotation styles rather than providers.

The 2026 GigaSpeechBench paper is a useful reminder: its 680-hour real-world evaluation reports substantial degradation for leading models and commercial APIs in difficult languages, dialects, accents, domains, and age groups. Your own traffic can be just as different from a clean public dataset.

Track more than WER

WER is necessary but incomplete.

Metric	Why it matters
Named-term accuracy	Product names and numbers can create disproportionate review work
Speaker attribution	Required for interviews, compliance, and contact-center analytics
Formatting quality	Punctuation and number formatting affect downstream usability
p95 finalization	Captures the delays users actually feel
Failure and retry rate	Converts reliability problems into cost
Review minutes/audio hour	Connects transcript quality with labor
Cost per accepted transcript	Combines processing, features, retries, and correction

For live speech, measure first partial, stable partial, and finalization separately. Time to first text can look good while the stable result remains slow.

Normalize the integration conditions

Before running the comparison, align:

audio codec and sample rate
selected region
mono, stereo, or multichannel handling
vocabulary hints and keyterms
diarization and redaction flags
timeouts and retries
model versions
timestamp and formatting options

Log every feature flag with each result. Deepgram, AssemblyAI, and ElevenLabs publish separate prices for some add-ons, while Google and AssemblyAI channel billing can multiply the base duration.

Use list prices only for the shortlist

The source research normalized public prices to one audio hour as of July 27, 2026:

Provider route	Public price snapshot	Important condition
AssemblyAI Universal-2	$0.15/hr	Multichannel billed per channel
OpenAI mini transcribe	$0.18/hr	25 MB uploaded-file limit
Google Dynamic Batch	$0.18/hr	Lower processing urgency
ElevenLabs Scribe v2	$0.22/hr	Entity and keyterm features cost extra
Deepgram Nova-3 pre-recorded	$0.462/hr	Diarization/redaction priced separately
Google standard recognition	$0.96/hr	First 500,000 monthly minutes; per-channel billing

These rows are not an accuracy ranking and do not represent equal service levels. Verify the official pricing pages immediately before using them in a procurement decision.

Turn the benchmark into a routing policy

Do not force one winner across unrelated workloads.

A practical production policy might use:

one low-cost asynchronous route for archives
one streaming specialist for live English calls
one multilingual route for meetings and media
one cloud-native route when data residency and procurement dominate

Teams evaluating supported OpenAI-compatible speech models can keep part of the client setup consistent with the CometAPI Quickstart. Verify the current model ID, input limits, feature support, and data terms before adding any route to the benchmark.

FLUX 3 Is Not a Public Production API Yet—Here Is What Developers Should Prepare

develpmilk — Fri, 24 Jul 2026 07:28:10 +0000

Black Forest Labs has announced FLUX 3, and the headline capabilities are impressive:

video generation with native audio;
clips up to 20 seconds per generation;
text, image, video, and keyframe references;
multilingual dialogue;
video and audio continuation;
a future open-weight multimodal model.

But there is one detail developers should not miss:

FLUX 3 has entered Early Access. It is not yet a generally available production API.

That distinction matters when you are deciding whether to build an integration, commit infrastructure, or promise a feature to users.

What can developers access today?

FLUX 3 Video is being offered through an application-based Early Access program.

Black Forest Labs has announced support for text-to-video, image-to-video, video-to-video, audiovisual continuation, keyframe transitions, animated typography, and multi-shot chaining.

The company has also said that every FLUX 3 Video output includes native audio generation.

What has not been published for general access is equally important:

stable production model IDs;
complete API specifications;
public pricing;
rate and concurrency limits;
expected queue time;
latency targets;
supported output codecs;
detailed moderation controls;
production data-handling terms.

Without these details, you can evaluate the model’s direction, but you cannot accurately estimate production cost or reliability.

Do not start with a provider-specific integration

The best preparation is not writing FLUX 3-specific code before the endpoint exists.

Instead, define a provider-neutral interface.

For example:

type VideoGenerationRequest = {
prompt: string;
durationSeconds: number;
aspectRatio: string;
startImageUrl?: string;
referenceVideoUrl?: string;
generateAudio?: boolean;
};

type VideoGenerationResult = {
jobId: string;
status: "queued" | "processing" | "completed" | "failed";
outputUrl?: string;
latencyMs?: number;
errorCode?: string;
};

Your application can then map this internal request format to whichever provider you are testing.

This makes it much easier to compare FLUX 3 with existing video models later without rewriting the rest of your product.

Build the benchmark before the API arrives

A useful evaluation set should represent your real product, not a collection of cinematic demo prompts.

Here are six groups worth testing.

Prompt adherence

Does the output include every required object and action?

Does it respect exclusions?

Does it follow camera and scene instructions?

Character consistency

Does the same person remain recognizable across generations?

Are clothing, facial features, body proportions, and voice preserved?

Motion quality

Do objects maintain weight and momentum?

Do hands make believable contact with tools and surfaces?

Do objects disappear or change shape between frames?

Native audio

Is dialogue accurate?

Does speech match lip movement?

Are sound effects synchronized with visible events?

Does multilingual dialogue remain understandable?

Reference control

How well does the model use starting frames, reference images, keyframes, and source videos?

Can it preserve identity while changing the scene?

Reliability

How many requests fail technically?

How long do queued jobs take?

How often must the same prompt be regenerated before an acceptable output appears?

Store more than the final video

For each benchmark request, save:

{
"prompt": "...",
"model": "...",
"duration_seconds": 10,
"resolution": "720p",
"queue_latency_ms": 0,
"generation_latency_ms": 0,
"technical_failure": false,
"safety_rejection": false,
"human_score": 0,
"notes": []
}

The exact schema can change, but storing structured results is critical.

Otherwise, evaluations quickly become collections of hand-picked videos with no reliable way to compare models.

Run every important prompt more than once

Generative video is stochastic.

A model may produce one excellent result and four unusable ones from the same prompt.

For production systems, consistency can matter more than maximum quality.

Consider tracking:

usable_output_rate =
accepted_generations / total_generations

You can also estimate an effective cost per accepted output:

effective_cost =
total_generation_cost / accepted_generations

A cheaper model can become expensive if users must regenerate repeatedly.

What about FLUX 3 Dev?

Black Forest Labs plans to release FLUX 3 Dev as an open-weight multimodal backbone.

That could be important for private inference, fine-tuning, research, and custom infrastructure.

However, the weights, license, parameter size, hardware requirements, and release timeline have not been published yet.

Do not purchase hardware based on assumptions from previous FLUX releases.

The practical conclusion

FLUX 3 is worth monitoring, especially for native-audio video and reference-driven workflows.

But there is no reason to pause development while waiting.

Build with available models, keep the integration provider-neutral, and create a reusable benchmark now. When FLUX 3 access expands, you will be able to test it against the same prompts and measurements instead of relying on launch demos.

Which metric matters most in your video pipeline: quality, latency, consistency, audio, or cost?

Llama 4 Maverick API Hands-On: Is Meta's Open Model Ready for Production?

develpmilk — Thu, 23 Jul 2026 07:22:24 +0000

Liquid syntax error: Unknown tag 'endraw'

DeepSeek V4 Flash: The Free Alternative to GPT-4o — Speed & Price Test

develpmilk — Thu, 23 Jul 2026 07:20:22 +0000

DeepSeek V4 Flash should only be called a free GPT-4o alternative if a current free tier applies to your account and it passes the same real tasks at an acceptable latency and error rate. This guide provides a reproducible test for end-to-end latency, token usage, estimated cost, and output quality without inventing benchmark numbers.

Publication note: “Free alternative” is the hypothesis tested by this article, not a verified conclusion. Before publishing, confirm the current model ID, free-tier terms, quotas, and prices, run the script below, and replace every [RUN REQUIRED] field with measured data.

Why this comparison needs a real protocol

“Faster” and “cheaper” are incomplete claims. Results change with prompt length, output length, region, provider route, load, retries, and the type of work being tested. A useful comparison must answer four separate questions:

Does each model complete the task correctly?
How long does the full response take?
How many input and output tokens does it consume?
What does one accepted result cost?

The benchmark below tests three workload types:

structured extraction;
code repair;
constrained reasoning.

It deliberately does not use one trivia prompt as a proxy for production quality.

The benchmark contract

Setting	Value
Models	DeepSeek V4 Flash and GPT-4o
Runs per prompt	5 by default
Prompt order	Randomized on every round
Temperature	0
Latency	Client-observed end-to-end time
Speed	Output tokens divided by end-to-end seconds
Cost	Input and output tokens × verified per-token prices
Quality	Human scoring against task-specific criteria

End-to-end latency is not time to first token. Measuring time to first token requires a streaming test and should be reported separately.

Setup

Install the client:

python -m pip install --upgrade openai

Copy the exact current model IDs and prices from the provider pages available to your account. Prices below are intentionally required environment variables so stale numbers cannot silently enter the article.

export COMETAPI_API_KEY="your_api_key_here"
export DEEPSEEK_MODEL_ID="current_deepseek_v4_flash_model_id"
export GPT4O_MODEL_ID="current_gpt_4o_model_id"

export DEEPSEEK_INPUT_USD_PER_M="verified_input_price"
export DEEPSEEK_OUTPUT_USD_PER_M="verified_output_price"
export GPT4O_INPUT_USD_PER_M="verified_input_price"
export GPT4O_OUTPUT_USD_PER_M="verified_output_price"

If a genuine free tier applies, keep the public list price in the cost calculation and report the free-tier allowance separately. This makes the result useful to readers who do not share the same promotion or account status.

Complete benchmark script

Save this as benchmark.py:

import json
import math
import os
import random
import statistics
import time
from pathlib import Path

from openai import OpenAI


def required(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"Set {name} before running the benchmark.")
    return value


def required_float(name: str) -> float:
    raw = required(name)
    try:
        return float(raw)
    except ValueError as exc:
        raise RuntimeError(f"{name} must be a number, got {raw!r}.") from exc


def percentile(values: list[float], fraction: float) -> float:
    ordered = sorted(values)
    index = max(0, math.ceil(fraction * len(ordered)) - 1)
    return ordered[index]


api_key = required("COMETAPI_API_KEY")
runs = int(os.environ.get("BENCHMARK_RUNS", "5"))
if runs < 3:
    raise RuntimeError("Use at least 3 runs per prompt; 5 or more is better.")

models = [
    {
        "label": "DeepSeek V4 Flash",
        "id": required("DEEPSEEK_MODEL_ID"),
        "input_usd_per_m": required_float("DEEPSEEK_INPUT_USD_PER_M"),
        "output_usd_per_m": required_float("DEEPSEEK_OUTPUT_USD_PER_M"),
    },
    {
        "label": "GPT-4o",
        "id": required("GPT4O_MODEL_ID"),
        "input_usd_per_m": required_float("GPT4O_INPUT_USD_PER_M"),
        "output_usd_per_m": required_float("GPT4O_OUTPUT_USD_PER_M"),
    },
]

tasks = [
    {
        "id": "extract",
        "prompt": (
            "Return JSON only with keys customer, plan, renewal_date, and risk. "
            "Input: Northwind Labs is on Pro. Renewal is 2026-09-30. "
            "The buyer says two failed exports may prevent renewal."
        ),
        "quality_check": (
            "Valid JSON; exact customer, plan, and date; risk mentions failed exports."
        ),
    },
    {
        "id": "code_repair",
        "prompt": (
            "Fix this Python function so a mutable default is not shared. Return the "
            "corrected function and a two-sentence explanation.\n\n"
            "def add_item(item, bucket=[]):\n"
            "    bucket.append(item)\n"
            "    return bucket"
        ),
        "quality_check": (
            "Uses None as the default, creates a list inside, and explains shared state."
        ),
    },
    {
        "id": "reasoning",
        "prompt": (
            "A batch job handles 120 records per minute. After every 600 records it "
            "pauses for exactly 30 seconds. Starting with no pause, how long does it "
            "take to finish 1,800 records? Show the assumptions and answer in minutes."
        ),
        "quality_check": (
            "States whether a pause after the final batch counts and calculates consistently."
        ),
    },
]

client = OpenAI(
    api_key=api_key,
    base_url="https://api.cometapi.com/v1",
    timeout=120.0,
    max_retries=0,  # Count failures instead of hiding them behind client retries.
)

results = []

for round_number in range(1, runs + 1):
    work = [(model, task) for task in tasks for model in models]
    random.shuffle(work)

    for model, task in work:
        started = time.perf_counter()
        row = {
            "round": round_number,
            "model": model["label"],
            "model_id": model["id"],
            "task": task["id"],
            "quality_check": task["quality_check"],
        }

        try:
            response = client.chat.completions.create(
                model=model["id"],
                messages=[{"role": "user", "content": task["prompt"]}],
                temperature=0,
            )
            elapsed = time.perf_counter() - started
            usage = response.usage
            input_tokens = usage.prompt_tokens if usage else 0
            output_tokens = usage.completion_tokens if usage else 0
            estimated_cost = (
                input_tokens * model["input_usd_per_m"] / 1_000_000
                + output_tokens * model["output_usd_per_m"] / 1_000_000
            )

            row.update(
                {
                    "ok": True,
                    "elapsed_seconds": elapsed,
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "output_tokens_per_second": (
                        output_tokens / elapsed if elapsed else None
                    ),
                    "estimated_list_cost_usd": estimated_cost,
                    "output": response.choices[0].message.content or "",
                }
            )
        except Exception as exc:
            row.update(
                {
                    "ok": False,
                    "elapsed_seconds": time.perf_counter() - started,
                    "error_type": type(exc).__name__,
                }
            )

        results.append(row)
        print(
            f"round={round_number} model={model['label']} "
            f"task={task['id']} ok={row['ok']} "
            f"seconds={row['elapsed_seconds']:.2f}"
        )

Path("benchmark_results.json").write_text(
    json.dumps(results, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

print("\nSummary (successful requests only)")
print("| Model | Success | Median latency | P95 latency | Median output tok/s | Total list cost |")
print("|---|---:|---:|---:|---:|---:|")

for model in models:
    rows = [row for row in results if row["model"] == model["label"]]
    successful = [row for row in rows if row["ok"]]
    if not successful:
        print(f"| {model['label']} | 0/{len(rows)} | n/a | n/a | n/a | n/a |")
        continue

    latencies = [row["elapsed_seconds"] for row in successful]
    speeds = [row["output_tokens_per_second"] for row in successful]
    total_cost = sum(row["estimated_list_cost_usd"] for row in successful)
    print(
        f"| {model['label']} | {len(successful)}/{len(rows)} "
        f"| {statistics.median(latencies):.2f}s "
        f"| {percentile(latencies, 0.95):.2f}s "
        f"| {statistics.median(speeds):.1f} "
        f"| ${total_cost:.6f} |"
    )

print("\nRaw responses are in benchmark_results.json. Score them blind.")

Run it from a quiet test machine and record the region and timestamp:

python benchmark.py

For stronger evidence, use at least 20 runs per prompt:

BENCHMARK_RUNS=20 python benchmark.py

How to score quality

Remove the model names from benchmark_results.json before review. Give each response a binary pass/fail score against the quality_check field, then calculate:

quality pass rate = accepted responses / successful responses
cost per accepted result = total measured cost / accepted responses

Cost per accepted result is usually more useful than price per million tokens. A cheap response that must be retried or rewritten may be the more expensive result.

Results

Replace this table after running the test. Do not publish the placeholders.

Metric	DeepSeek V4 Flash	GPT-4o
Successful requests	[RUN REQUIRED]	[RUN REQUIRED]
Quality pass rate	[RUN REQUIRED]	[RUN REQUIRED]
Median end-to-end latency	[RUN REQUIRED]	[RUN REQUIRED]
P95 end-to-end latency	[RUN REQUIRED]	[RUN REQUIRED]
Median output tokens/second	[RUN REQUIRED]	[RUN REQUIRED]
Verified input price / 1M tokens	[VERIFY]	[VERIFY]
Verified output price / 1M tokens	[VERIFY]	[VERIFY]
Cost per accepted result	[RUN REQUIRED]	[RUN REQUIRED]
Free-tier allowance and expiry	[VERIFY]	[VERIFY]

Also publish the test timestamp, client region, model IDs, number of runs, prompts, and failures. Without that context, readers cannot reproduce the comparison.

When “free” is a useful label

A free tier can be meaningful for prototypes, learning, or low-volume internal tools. It is not the same as a permanent zero-cost production service. Check:

whether the offer is a trial, recurring allowance, or temporary promotion;
request and token quotas;
rate limits and concurrency;
eligible accounts and regions;
overage behavior;
whether input and output tokens are treated differently;
the date the terms were last verified.

If any of these are unclear, change the headline before publication to “DeepSeek V4 Flash vs GPT-4o: A Reproducible Speed & Price Test.”

Verdict template

Use this only after the run:

In this test, DeepSeek V4 Flash achieved [X]% accepted outputs versus GPT-4o's [Y]%. Its median latency was [A]s and cost per accepted result was $[B], measured on [date] from [region]. The current free tier [did/did not] cover this workload. These results apply to the three published tasks, not to every use case.

Key takeaways

Treat “free,” “faster,” and “cheaper” as claims that require dated evidence.
Randomize run order and report median, P95, and failures.
Compare cost per accepted result, not price alone.
Publish the prompts, exact model IDs, prices, account conditions, and test timestamp.
Re-run the benchmark when providers change routes, models, or pricing.

How to Access 500+ AI Models with One API Key — Complete 2026 Guide

develpmilk — Thu, 23 Jul 2026 07:16:38 +0000

You can access 500+ AI models through one API key by using an AI API gateway with an OpenAI-compatible endpoint. Your application keeps one client, one authentication flow, and one request format; changing models becomes a configuration change. You still need to test each model's capabilities, limits, latency, and output behavior before production use.

Why I wanted one integration instead of 12 SDKs

Most AI applications do not stay on one model forever. A team may use one model for code, another for low-cost classification, and a third for image or video generation. Direct integrations can quickly turn into a maintenance problem:

credentials live in different systems;
request and response formats vary;
model names get scattered through the codebase;
retry, timeout, and logging logic is duplicated;
switching providers requires application changes.

An API gateway does not make every model identical. It gives the application a stable entry point and a shared authentication layer. Model-specific features still need model-specific testing.

For this walkthrough, I use CometAPI, which currently advertises access to 500+ models through one API key and an OpenAI-compatible API.

What actually stays the same?

Integration concern	Without a shared gateway	With a shared gateway
API key	One or more per provider	One application secret
Base URL	Different for each provider	One base URL
Client setup	Multiple SDKs or adapters	One OpenAI-compatible client
Model selection	Often tied to provider code	A configuration value
Observability	Separate logs and dashboards	A shared request path
Model behavior	Varies by model	Still varies by model

The last row matters: a common API schema reduces integration work, but it does not erase differences in context windows, tool calling, structured output, safety behavior, or multimodal support.

Prerequisites

Python 3.10 or later
A CometAPI API key
A current model ID copied from the CometAPI model catalog or documentation

Install the official OpenAI Python client:

python -m pip install --upgrade openai

Store credentials in environment variables. Never paste a real API key into source code or a Git repository.

export COMETAPI_API_KEY="your_api_key_here"
export COMETAPI_MODEL_ID="copy_a_current_model_id_from_the_catalog"

Make the first request

Create quickstart.py:

import os
from openai import OpenAI

api_key = os.environ.get("COMETAPI_API_KEY")
model_id = os.environ.get("COMETAPI_MODEL_ID")

if not api_key:
    raise RuntimeError("Set COMETAPI_API_KEY before running this script.")
if not model_id:
    raise RuntimeError("Set COMETAPI_MODEL_ID to a current catalog model ID.")

client = OpenAI(
    api_key=api_key,
    base_url="https://api.cometapi.com/v1",
    timeout=60.0,
    max_retries=2,
)

response = client.chat.completions.create(
    model=model_id,
    messages=[
        {
            "role": "system",
            "content": "You are a concise senior Python reviewer.",
        },
        {
            "role": "user",
            "content": (
                "Review this function and return three concrete improvements:\n"
                "def total(xs):\n"
                "    value = 0\n"
                "    for x in xs:\n"
                "        value += x\n"
                "    return value"
            ),
        },
    ],
    temperature=0.2,
)

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

if response.usage:
    print(
        f"tokens: input={response.usage.prompt_tokens}, "
        f"output={response.usage.completion_tokens}"
    )

Run it:

python quickstart.py

The important part is not the prompt. It is that the base URL and client stay fixed while COMETAPI_MODEL_ID is configuration.

Switch models without changing application code

Suppose a staging test needs to compare three current catalog models. Put their exact IDs in a comma-separated environment variable:

export COMETAPI_MODEL_IDS="model_id_a,model_id_b,model_id_c"

Then run the same task against each model:

import os
import time
from openai import OpenAI

api_key = os.environ.get("COMETAPI_API_KEY")
raw_model_ids = os.environ.get("COMETAPI_MODEL_IDS", "")
model_ids = [item.strip() for item in raw_model_ids.split(",") if item.strip()]

if not api_key:
    raise RuntimeError("Set COMETAPI_API_KEY.")
if not model_ids:
    raise RuntimeError("Set COMETAPI_MODEL_IDS to one or more current model IDs.")

client = OpenAI(
    api_key=api_key,
    base_url="https://api.cometapi.com/v1",
    timeout=60.0,
    max_retries=2,
)

prompt = "Return JSON with keys summary and risk for a password-reset email."

for model_id in model_ids:
    started = time.perf_counter()
    try:
        response = client.chat.completions.create(
            model=model_id,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )
        elapsed = time.perf_counter() - started
        output = response.choices[0].message.content or ""
        print(f"\n[{model_id}] {elapsed:.2f}s")
        print(output)
    except Exception as exc:
        elapsed = time.perf_counter() - started
        print(f"\n[{model_id}] failed after {elapsed:.2f}s: {type(exc).__name__}")

This is a smoke test, not a benchmark. Run multiple randomized trials before drawing conclusions about latency or quality.

A safer production pattern

Do not let route handlers choose arbitrary model IDs. Put approved models in a small policy layer:

MODEL_POLICY = {
    "fast": os.environ["FAST_MODEL_ID"],
    "quality": os.environ["QUALITY_MODEL_ID"],
    "fallback": os.environ["FALLBACK_MODEL_ID"],
}

def choose_model(task_class: str) -> str:
    try:
        return MODEL_POLICY[task_class]
    except KeyError as exc:
        raise ValueError(f"Unsupported task class: {task_class}") from exc

This keeps product logic independent from vendor naming and gives your team one place to change routing decisions.

What to test before adding a model

Use the same acceptance suite for every candidate:

Capability: Does the model support the endpoint and features you use?
Quality: What is its pass rate on 30–100 representative tasks?
Latency: What are median and P95 response times under realistic load?
Reliability: How often do requests time out, return invalid data, or require a retry?
Cost: What is the cost per accepted result, not merely the list price per token?
Limits: Are context, output, rate, and regional limits compatible with the application?
Data handling: Does the route meet your privacy and retention requirements?

Common mistakes

Hard-coding the API key or model ID in a public example
Assuming all models support identical request parameters
Moving to a new model after a single impressive prompt
Comparing prices without separating input and output tokens
Treating HTTP success as task success
Omitting timeouts, retry limits, logging, and a fallback path

Key takeaways

One OpenAI-compatible client can remove a large amount of provider-specific integration code.
The model ID should be configuration, not application logic.
One schema does not mean identical capabilities or behavior.
A model should pass your own quality, latency, reliability, and cost gates before production.
Secrets belong in environment variables or a secrets manager.

How to Migrate to Gemini 3.6 Flash Without Breaking Production

develpmilk — Wed, 22 Jul 2026 09:56:57 +0000

Gemini 3.6 Flash is generally available at $1.50 per million input tokens and $7.50 per million output tokens. Migrating requires more than changing the model ID: remove deprecated sampling parameters, replace thinking_budget, remove candidate_count, eliminate prefilled model turns, and retest function calling before production rollout.

Why This Migration Can Break

If your Gemini settings have been shared across several model generations, the model-name change is the easy part. The dangerous part is old configuration that is ignored today and may return HTTP 400 later.

Google's current guide identifies five areas to check:

temperature, top_p, and top_k are deprecated;
thinking_budget should become thinking_level;
candidate_count is unsupported in Gemini 3.x;
a request cannot end with a prefilled non-empty model turn;
function-call and multi-turn state must be retested.

Pricing Snapshot

Model	Input / 1M	Output / 1M	Default thinking
Gemini 3.6 Flash	$1.50	$7.50	`medium`
Gemini 3.5 Flash-Lite	$0.30	$2.50	`minimal`

Gemini 3.6 Flash has the same input price as Gemini 3.5 Flash, but output falls from $9.00 to $7.50 per million tokens, a 16.7% reduction.

Do not route every task to 3.6 Flash just because it is newer. Flash-Lite remains a better first route for high-volume classification, extraction, and predictable transformation when it meets your acceptance criteria.

Setup

Use Python 3.11+ and update Google's GenAI SDK:

python -m venv .venv
source .venv/bin/activate
pip install -U google-genai

export GEMINI_API_KEY="your-key"

Never commit the API key.

The New Request

The current Interactions API example is intentionally small:

from google import genai


client = genai.Client()

interaction = client.interactions.create(
    model="gemini-3.6-flash",
    input=(
        "Review this migration plan and return the three "
        "highest-risk compatibility issues."
    ),
)

print(interaction.output_text)

Gemini 3.6 Flash defaults to medium thinking. Configure the level only when your evaluation shows a reason to change it:

from google import genai


client = genai.Client()

interaction = client.interactions.create(
    model="gemini-3.6-flash",
    input="Analyze this multi-step debugging problem.",
    generation_config={
        "thinking_level": "medium",
    },
)

print(interaction.output_text)

Remove the Old Generation Config

This shared configuration should not move to the new models:

# Remove this configuration for Gemini 3.6 Flash.
old_config = {
    "temperature": 0.2,
    "top_p": 0.9,
    "top_k": 40,
    "candidate_count": 1,
    "thinking_budget": 4096,
}

Use the smaller configuration instead:

new_config = {
    "thinking_level": "medium",
}

Explicit system instructions and structured outputs are more reliable than deprecated sampling controls when you need a fixed response shape.

Remove Prefilled Model Turns

Legacy payloads sometimes end with a model turn to force a prefix:

{
  "contents": [
    {
      "role": "user",
      "parts": [{"text": "Translate 'Hello world' to Spanish."}]
    },
    {
      "role": "model",
      "parts": [{"text": "Translation:"}]
    }
  ]
}

Gemini 3.6 Flash rejects that pattern with HTTP 400. Move the format requirement into a system instruction:

interaction = client.interactions.create(
    model="gemini-3.6-flash",
    input="Translate 'Hello world' to Spanish.",
    system_instruction=(
        "Output only the translation without introductory text."
    ),
)

Use structured outputs when your application needs machine-validated JSON.

Audit Function Calling

For multi-step agents, a successful text request does not prove the migration is complete. Test:

tool selection;
tool argument validity;
propagation of the function name and call ID;
tool-result handling;
multi-turn state through previous_interaction_id;
retry behavior after malformed calls or timeouts.

If you still use generateContent, verify that every FunctionResponse carries both call_id and name.

Add a Cost Calculator to Your Eval

from dataclasses import dataclass


@dataclass(frozen=True)
class Price:
    input_per_million: float
    output_per_million: float


GEMINI_36_FLASH = Price(1.50, 7.50)
GEMINI_35_FLASH_LITE = Price(0.30, 2.50)


def request_cost(
    input_tokens: int,
    output_tokens: int,
    price: Price,
) -> float:
    return (
        input_tokens * price.input_per_million
        + output_tokens * price.output_per_million
    ) / 1_000_000


print(request_cost(15_000, 8_000, GEMINI_36_FLASH))
# 0.0825

Thinking tokens are billed as output, so feed the billed token counts from the API into your analysis instead of estimating from visible response length.

Production Evaluation Checklist

Run 30 to 50 recent tasks through your current production model, Gemini 3.6 Flash, and Gemini 3.5 Flash-Lite. Keep prompts, tools, validators, and acceptance rules unchanged.

Record:

input, output, and thinking tokens;
P50 and P95 latency;
validator pass rate;
number of tool calls and retries;
function-call errors;
human correction time;
cost per accepted task.

Google reports that 3.6 Flash uses 17% fewer output tokens than 3.5 Flash on the Artificial Analysis Index and up to 65% fewer on DeepSWE. Treat those as reasons to run an eval, not assumptions about your traffic.

Key Takeaways

Update the model ID to gemini-3.6-flash.
Delete deprecated sampling parameters now.
Replace thinking_budget with thinking_level.
Remove candidate_count and prefilled model turns.
Retest function calls and multi-turn state separately.
Route predictable high-volume work to Flash-Lite when it passes validation.
Compare cost per accepted task, not only price per token.

Primary references: Google's launch announcement and latest-model migration guide.

You can also test Gemini alongside other models through CometAPI while keeping one evaluation harness.

P.S. I work with CometAPI. Google's prices, model IDs, and migration requirements are linked above; benchmark gains should be reproduced on your own workload.

How Can You Access 500+ AI Models Without Rewriting Your OpenAI SDK Integration?

develpmilk — Wed, 22 Jul 2026 08:58:24 +0000

How Can You Access 500+ AI Models Without Rewriting Your OpenAI SDK Integration?

If your application already uses the OpenAI Python SDK, you can connect it to an OpenAI-compatible gateway by changing the API key, base URL, and model ID. CometAPI's official site currently lists access to 500+ models through one API key, so the same client structure can be used while you test different model families.

Why I Built This

The first model integration is usually easy. The second one is where duplicated authentication, provider-specific clients, billing, retries, and error handling start to spread through the codebase.

For an early prototype, that duplication may be acceptable. For a product that needs model evaluation or fallback, I prefer to keep the application-facing interface stable and move provider selection behind one boundary.

Setup

Requirements:

Python 3.11+
The current openai Python package
A CometAPI API key stored outside the source code
A model ID confirmed in the current model catalog

python -m venv .venv
source .venv/bin/activate
pip install openai

export COMETAPI_API_KEY="your-key"
export COMETAPI_MODEL="your-current-model-id"

Minimal Working Example

import os

from openai import OpenAI


client = OpenAI(
    api_key=os.environ["COMETAPI_API_KEY"],
    base_url="https://api.cometapi.com/v1",
)

response = client.chat.completions.create(
    model=os.environ["COMETAPI_MODEL"],
    messages=[
        {
            "role": "user",
            "content": "Return one practical reason to test more than one model.",
        }
    ],
)

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

The important part is not the number of changed lines. It is the boundary: application code talks to one OpenAI-compatible client, while the model ID remains configuration.

Add a Small Model Abstraction

Avoid scattering model IDs throughout the application. Put the choice in one function so experiments, rollbacks, and environment-specific routing stay manageable.

import os

from openai import OpenAI


client = OpenAI(
    api_key=os.environ["COMETAPI_API_KEY"],
    base_url="https://api.cometapi.com/v1",
)


def complete(prompt: str, model: str | None = None) -> str:
    selected_model = model or os.environ["COMETAPI_MODEL"]
    response = client.chat.completions.create(
        model=selected_model,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content or ""


if __name__ == "__main__":
    print(complete("Summarize the value of model portability in two sentences."))

What Changes in Production?

An OpenAI-compatible request format reduces integration work, but it does not make models interchangeable. Before routing production traffic, test the dimensions that affect your workload.

Check	Why it matters
Model ID and availability	Catalogs change; validate before deployment
Tool-calling behavior	Schemas and edge cases can vary by model
Structured output adherence	JSON validity should be measured, not assumed
Timeout and retry policy	Retries can duplicate non-idempotent work
Latency distribution	P95/P99 matters more than one fast demo
Cost per solved task	Token price alone ignores retries and review time

I would run a fixed evaluation set before switching a production default: at least 20 representative prompts, expected-output checks, latency capture, error classification, and a manual review of the failures.

Key Takeaways

Keep the API key and model ID in environment variables.
Treat OpenAI compatibility as an integration interface, not a quality guarantee.
Centralize the model choice instead of hard-coding it across features.
Compare cost per accepted result, not only price per token.
Add fallback only after defining which errors are safe to retry.

The official OpenAI SDK guide covers the current setup, while the CometAPI homepage lists the current catalog scope.

P.S. I used CometAPI for this example. Full disclosure: I work with CometAPI, but the code path and verification steps above are reproducible.