Build a Cost-Aware Model Router MCP Server for Peak/Off-Peak

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

On August 16, 2026 at 16:00 UTC, DeepSeek replaced flat API pricing with a two-tier clock: peak hours (01:00–04:00 and 06:00–10:00 UTC, seven hours a day) bill at double the off-peak rate, and even the off-peak tier is more expensive than the old flat card. For V4-Pro, a cache-hit input token jumped from roughly $0.004 to $0.022 off-peak and $0.044 at peak; output went from $0.87 to $1.98/$3.96 per million tokens. Teams that route a 24/7 agent fleet on "one model, one price" just watched their LLM bill become a function of the UTC clock. The fix is not to pray for a discount — it is to build routing that treats price as a first-class decision input.

This dispatch builds cost-router-mcp: a production-ready FastMCP (Python) server that routes agent LLM calls to the cheapest eligible model using time-of-day pricing, a refreshable cost table, task difficulty, and hard budget caps. It ships a routing decision engine plus four tools — route_request, estimate_cost, list_models, and budget_status — so any agent (Claude, custom, or your internal orchestrator) can ask "who should answer this, at this price, right now?" We cover the full stack: runnable server code, the cost-table JSON, provider config, token-budget enforcement, fallback and retry rules, and OAuth 2.0 / API-key security for provider credentials. The latest AI news hub covered the pricing shift; this guide turns it into infrastructure.

Why routing became a cost problem overnight

Before August 16, model choice was a quality-vs-latency trade. After it, model choice is also a time-of-day trade. The same call that costs $1.32/M output on V4-Flash at 08:00 UTC costs $0.66 at 11:00 UTC. For a fleet running millions of tokens a day, that spread is real money — and it is pure upside: the quality of the output is identical, only the billing window changed. A cost-aware router captures that upside automatically:

• reads the clock, checks the peak/off-peak windows, and applies the right rate card;
• computes the full token estimate (cache-hit input, cache-miss input, output) for each eligible model;
• filters to models that meet the task's quality floor and latency budget;
• picks the cheapest survivor and reserves the cost against the agent's budget cap;
• fails over to the next cheapest model on errors, within budget.

None of this changes your prompt, your chain, or your output. It changes only the question "which model?" — from a static config to a live decision.

The routing decision engine

The engine is a pure function: choose_model(task, difficulty, budget) -> model_id. It computes is_peak(now_utc), looks up each candidate model's rate card, estimates tokens from the task length and difficulty profile, drops models whose estimated cost exceeds the remaining budget, sorts by cost, and returns the cheapest. estimate_cost exposes the same math as a tool so an agent can quote a price before committing — a genuinely useful trick for cost-sensitive user questions ("how much would this report cost?").

Project setup

mkdir cost-router-mcp && cd cost-router-mcp
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

requirements.txt:

fastmcp>=2.0.0
httpx>=0.27.0
pydantic>=2.7.0
python-dotenv>=1.0.0

.env:

COST_TABLE_PATH=cost_table.json
DEFAULT_MAX_BUDGET_USD=0.50
AGENT_BUDGET_CAP_USD=25.00
FALLBACK_CHAIN=deepseek-v4-flash,deepseek-v4-pro
PROVIDER_KEYS_JSON={"deepseek":"sk-live-deepseek-xxxx"}

cost_table.json — the cost table

{
  "currency": "USD",
  "per_million_tokens": true,
  "window": {
    "tz": "UTC",
    "peak": [["01:00", "04:00"], ["06:00", "10:00"]]
  },
  "models": [
    {
      "id": "deepseek-v4-flash",
      "provider": "deepseek",
      "quality": 0.4,
      "peak":    { "input_hit": 0.014, "input_miss": 0.44, "output": 1.32 },
      "off_peak": { "input_hit": 0.007, "input_miss": 0.22, "output": 0.66 }
    },
    {
      "id": "deepseek-v4-pro",
      "provider": "deepseek",
      "quality": 0.9,
      "peak":    { "input_hit": 0.044, "input_miss": 1.32, "output": 3.96 },
      "off_peak": { "input_hit": 0.022, "input_miss": 0.66, "output": 1.98 }
    }
  ]
}

Keep the rate card in a JSON file (or behind a pricing API) so you can refresh it the day a provider changes prices — the DeepSeek change is your warning that every provider's card will move. A scheduled job that re-fetches cost_table.json and validates it before the router reads it is cheap insurance.

server.py — the complete FastMCP server

# server.py
import json
import os
from datetime import datetime, timezone

from dotenv import load_dotenv
from fastmcp import FastMCP

load_dotenv()

COST_TABLE_PATH = os.getenv("COST_TABLE_PATH", "cost_table.json")
DEFAULT_MAX_BUDGET_USD = float(os.getenv("DEFAULT_MAX_BUDGET_USD", "0.50"))
AGENT_BUDGET_CAP_USD = float(os.getenv("AGENT_BUDGET_CAP_USD", "25.00"))
FALLBACK_CHAIN = [m for m in os.getenv("FALLBACK_CHAIN", "").split(",") if m]

mcp = FastMCP("cost-router-mcp", version="1.0.0")


def load_cost_table():
    with open(COST_TABLE_PATH) as f:
        return json.load(f)


def in_peak(now: datetime, window: dict) -> bool:
    hhmm = now.strftime("%H:%M")
    for start, end in window["peak"]:
        if start <= hhmm < end:
            return True
    return False


def estimate_cost(model: dict, input_tokens: int, output_tokens: int,
                  peak: bool, cache_hit_rate: float = 0.5) -> float:
    card = model["peak"] if peak else model["off_peak"]
    input_cost = (input_tokens * (1 - cache_hit_rate) * card["input_miss"]
                  + input_tokens * cache_hit_rate * card["input_hit"])
    output_cost = output_tokens * card["output"]
    return (input_cost + output_cost) / 1_000_000  # per-million rates


_budgets: dict[str, float] = {}  # agent_id -> remaining USD for the window


def choose_model(difficulty: str, max_budget_usd: float,
                 input_tokens: int, output_tokens: int) -> dict:
    table = load_cost_table()
    now = datetime.now(timezone.utc)
    peak = in_peak(now, table["window"])
    floors = {"simple": 0.3, "standard": 0.5, "complex": 0.8}
    floor = floors.get(difficulty, 0.5)
    candidates = []
    for model in table["models"]:
        if model["quality"] < floor:
            continue
        cost = estimate_cost(model, input_tokens, output_tokens, peak)
        if cost > max_budget_usd:
            continue
        candidates.append({"model": model["id"], "cost_usd": round(cost, 6), "peak": peak})
    candidates.sort(key=lambda c: c["cost_usd"])
    if not candidates:
        raise ValueError("No model fits budget; raise cap or lower difficulty")
    return candidates[0]


@mcp.tool()
def route_request(task: str, difficulty: str, input_tokens: int,
                  output_tokens: int, max_budget_usd: float | None = None,
                  agent_id: str = "default") -> dict:
    """Choose the cheapest eligible model for a task right now, within budget,
    and reserve the cost against the agent's window budget."""
    budget = min(max_budget_usd or DEFAULT_MAX_BUDGET_USD,
                 _budgets.get(agent_id, AGENT_BUDGET_CAP_USD))
    choice = choose_model(difficulty, budget, input_tokens, output_tokens)
    _budgets[agent_id] = round(_budgets.get(agent_id, AGENT_BUDGET_CAP_USD)
                               - choice["cost_usd"], 6)
    return {"chosen": choice["model"], "cost_usd": choice["cost_usd"],
            "peak": choice["peak"], "remaining_budget": _budgets[agent_id],
            "fallback_chain": FALLBACK_CHAIN}


@mcp.tool()
def estimate_cost(model: str, input_tokens: int, output_tokens: int,
                  now_utc: str | None = None) -> dict:
    """Quote the price for a model given token counts and an optional UTC time."""
    table = load_cost_table()
    now = (datetime.fromisoformat(now_utc) if now_utc
           else datetime.now(timezone.utc))
    peak = in_peak(now, table["window"])
    model_def = next(m for m in table["models"] if m["id"] == model)
    return {"model": model, "peak": peak,
            "cost_usd": round(estimate_cost(model_def, input_tokens,
                                            output_tokens, peak), 6)}


@mcp.tool()
def list_models(min_quality: float = 0.0) -> list[dict]:
    """List models in the cost table with quality scores and live rates."""
    table = load_cost_table()
    now = datetime.now(timezone.utc)
    peak = in_peak(now, table["window"])
    return [{"id": m["id"], "provider": m["provider"], "quality": m["quality"],
             "rate": m["peak"] if peak else m["off_peak"]}
            for m in table["models"] if m["quality"] >= min_quality]


@mcp.tool()
def budget_status(agent_id: str = "default") -> dict:
    """Return the agent's remaining window budget."""
    return {"agent_id": agent_id,
            "remaining_usd": _budgets.get(agent_id, AGENT_BUDGET_CAP_USD),
            "window_cap_usd": AGENT_BUDGET_CAP_USD}


if __name__ == "__main__":
    mcp.run(transport="stdio")

FastMCP derives the schema from the type hints. The route_request inputSchema the model sees looks like this:

{
  "name": "route_request",
  "description": "Choose the cheapest eligible model for a task right now, within budget, and reserve the cost against the agent's window budget.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "task": { "type": "string" },
      "difficulty": { "type": "string", "enum": ["simple", "standard", "complex"] },
      "input_tokens": { "type": "integer", "minimum": 0 },
      "output_tokens": { "type": "integer", "minimum": 0 },
      "max_budget_usd": { "type": "number", "minimum": 0 },
      "agent_id": { "type": "string" }
    },
    "required": ["task", "difficulty", "input_tokens", "output_tokens"]
  }
}

Token-budget enforcement

An unrouted fleet is an unbounded bill; a budget-aware fleet is a capped one. The server keeps a per-agent window budget: AGENT_BUDGET_CAP_USD (default $25/day per agent) is consumed by every route_request reservation, and budget_status lets the agent (or your orchestrator) check remaining headroom before kicking off a batch. When the ledger hits zero, route_request refuses new calls — the agent either defers the work to the next UTC window or escalates to a human for a budget raise. This is the same discipline as a cloud spend alert, but enforced at the tool layer where it cannot be ignored.

Fallback and retry rules

Providers fail in clusters — when one is degraded, the router must not hammer it. Implement a per-provider circuit breaker: after five consecutive failures, stop sending calls to that provider for 30 seconds and let the fallback chain absorb traffic. Apply exponential backoff (0.5s → 1s → 2s) on transient 429s and respect Retry-After. And because routing decisions are pure functions of (time, cost table, budget), a replayed route_request is harmless — no idempotency key is needed on reads, only on the downstream provider call you issue from the chosen model.

Security: OAuth 2.0 and API keys for providers

The router holds credentials for every provider in your fleet, which makes it a high-value target. The blast radius is controlled the same way as any payment system:

• Provider API keys in a secret store. PROVIDER_KEYS_JSON is for local development only. In production, read keys from Vault, AWS Secrets Manager, or the OS keychain at startup, and never expose them through MCP tools, logs, or errors. A model that can call list_models must not be able to read sk-live-*.
• OAuth 2.0 for the control plane. If you manage budgets and rate cards from a dashboard or multi-tenant control plane, protect those endpoints with OAuth 2.0 authorization-code flow and scopes like router:route, budget:read, budget:write, config:read. The router's own provider calls use scoped API keys or client-credentials tokens per provider — never one super-key for every vendor.
• Least-privilege scopes. Grant the router only the scopes it needs per provider (chat/completions), never admin or billing endpoints. Rotate provider keys every 90 days and on any compromise; rotate control-plane client secrets on the same cadence.
• Audit every decision. Log model chosen, cost, peak/off-peak flag, agent_id, and timestamp. Cost routing without an audit trail is just an opinion; with one, it is a finance report.

Wire it into your fleet

Run fastmcp dev server.py, validate each tool in the MCP Inspector, then add the server to your client's mcpServers block:

{
  "mcpServers": {
    "cost-router": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "/opt/cost-router-mcp",
      "env": {
        "COST_TABLE_PATH": "cost_table.json",
        "AGENT_BUDGET_CAP_USD": "25.00",
        "PROVIDER_KEYS_JSON": "${ROUTER_PROVIDER_KEYS}"
      }
    }
  }
}

Start with estimate_cost and list_models in a shadow mode where the router recommends but the fleet still uses the old model — measure the delta for a week. Then flip route_request on for simple tasks, then standard, and finally complex. You will almost certainly find that a share of your fleet's traffic happily rides V4-Flash off-peak, that your peak-hour spend drops by cutting complex calls in the 01:00–04:00 window, and that the budget ledger catches the runaway agent before finance does. The MCP Directory has more servers to pair with this one, and the Workflows hub shows how routers like this slot into full agent pipelines.

FAQ

Does peak/off-peak routing change output quality? No. The rates differ by clock, not capability — deepseek-v4-flash at 09:00 UTC answers the same as at 13:00 UTC. Routing by time-of-day captures price savings with zero quality tradeoff; routing by difficulty is where quality decisions live.

Which hours are peak, exactly? DeepSeek's card, effective 16:00 UTC August 16, 2026, marks 01:00–04:00 and 06:00–10:00 UTC as peak (seven hours/day); everything else is off-peak at half the peak rate. The router reads the window from cost_table.json, so you can extend the same logic to any provider with time-tiered pricing.

Is this just DeepSeek? The pattern is provider-agnostic. The router holds a rate card per model and a per-model peak window; add OpenAI, Anthropic, or Google rates to the same table and the same choose_model math picks the cheapest eligible provider at any hour.

How do I stop one agent from eating the whole budget? Per-agent window budgets in the ledger, enforced by route_request refusing calls once the cap is hit, plus budget_status for visibility. Escalation (defer or raise) is a deliberate agent behavior, not an accident.

Where should provider keys live in production? In a secrets manager (Vault, AWS Secrets Manager, OS keychain), injected at startup. OAuth 2.0 protects the control plane that edits budgets and rate cards; per-provider scoped API keys handle model calls. Never commit keys, never log them, rotate every 90 days.