maslul

Smart LLM router — one call, the right model.

Async and fully typed, across Anthropic, Gemini, xAI Grok, and OpenAI — routing each request to the right model tier by difficulty. Stop hardcoding model choices and stop re-writing the tool-use / structured-output / web-search / retry plumbing for every provider.

maslul (Hebrew מסלול, "route / lane") is a small library that does exactly two things: routing (pick a model tier per request, or pin one) and provider normalization (one Request/Response shape for every SDK). No server, no CLI, no heavy ML deps — providers live behind extras, and the core is stdlib-only.

import asyncio
from maslul import Router, Request, Message

router = Router.from_toml("maslul.toml")           # tiers + classifier + providers, from config

async def main() -> None:
    resp = await router.complete(Request(messages=[Message(role="user", content="Hello!")]))
    print(resp.text, "·", resp.level_used, "·", resp.usage.output_tokens, "tokens")

asyncio.run(main())

Install

pip install "maslul[anthropic,gemini,grok]"     # or just the providers you use

Each provider's SDK lives behind an extra, so import maslul pulls in none of them — you only install what you route to. maslul[anthropic] → anthropic; maslul[gemini] → google-genai; maslul[grok] → xai-sdk; maslul[openai] → openai.

How it compares

maslul is a library, not a gateway — you embed the routing brain in your app, you don't run a proxy in front of it.

Choose maslul when you want a typed async library you embed — difficulty routing with your own strategy + hooks, and one Request/Response over several providers (tools, structured output, vision, web search, retries, cost cache) — without standing up a gateway. Reach for LiteLLM when you want a provider proxy across 100+ models, or RouteLLM when you specifically want a trained router.

The routing brain

flowchart LR
    R["complete(req)"] --> M{"model= pin?"}
    M -- yes --> RUN["run that model"]
    M -- no --> L{"level= pin?"}
    L -- yes --> RUN
    L -- no --> B{"bypass_predicate?"}
    B -- "tier" --> RUN
    B -- "None" --> H{"hard_signal?<br/>(media · code · long · intent verbs)"}
    H -- "yes" --> HARD["HARD tier"] --> RUN
    H -- "no" --> S["strategy<br/>route_default · classify ·<br/>classify_and_answer · verify_cascade"] --> RUN
    RUN --> X["tool loop · web search ·<br/>retry / fallback · usage breakdown"]

Routing

Difficulty is not readable from surface features — a short prompt can be very hard, a long paste trivial — so maslul never applies a short ⇒ simple rule. You choose how each request is routed, in this precedence order:

from maslul import Level

await router.complete(req, model="anthropic:claude-opus-4-8")  # 0. pin an exact model
await router.complete(req, level=Level.HARD)                   # 1. pin a difficulty tier
await router.complete(req)                                     # 2-4. let the router decide

When you don't pin, the routing brain runs: a deterministic bypass (your fast-path, e.g. greetings → SIMPLE) → a hard-signal detector (intent verbs, code, attachments, long context → HARD, up-only) → the configured strategy for the ambiguous middle:

All three injection points are yours to supply:

def my_classifier(req):      # your own difficulty call (sync or async); None defers to the strategy
    return Level.SIMPLE if is_trivial(req) else None

def my_verifier(req, resp):  # VERIFY_CASCADE: True keeps the cheap answer, False escalates
    return "I don't know" not in resp.text

router = Router.from_toml("maslul.toml", classifier=my_classifier, verifier=my_verifier)

One shape for every capability

The same Request/Response works across all three providers:

from maslul import Request, Message, ToolDef, ToolCall, MediaPart

# Tools — the router runs a provider-agnostic tool-use loop
async def get_weather(call: ToolCall) -> str:
    return f"18°C in {call.input['city']}"

req = Request(
    messages=[Message(role="user", content="Weather in Paris?")],
    tools=[ToolDef(name="get_weather", description="Current weather for a city.",
                   input_schema={"type": "object", "properties": {"city": {"type": "string"}},
                                 "required": ["city"]})],
    tool_executor=get_weather,
)

# Structured output — response_format → resp.structured (parsed)
req = Request(messages=[Message(role="user", content="Extract name + age")],
              response_format={"type": "object", "properties": {"name": {"type": "string"},
                                                                "age": {"type": "integer"}}})

# Vision — images / PDFs
req = Request(messages=[Message(role="user", content="What's in this image?")],
              media=[MediaPart(mime_type="image/png", data=png_bytes)])

# Web search — one flag, grounded on ANY provider (Anthropic web_search / Gemini Google Search /
# Grok Agent Tools); citations land in resp.sources regardless of which model answers.
req = Request(messages=[Message(role="user", content="Latest news on X?")], web_search=True)

Resilience & observability

def on_usage(resp):                         # per-model token breakdown for monitoring
    for rec in resp.usage_records:
        metrics.incr(f"{rec.provider}:{rec.model}", rec.usage.output_tokens)

router = Router.from_toml("maslul.toml", on_complete=on_usage)

Transient errors (RateLimited, Timeout) retry with exponential backoff; on persistent failure the request falls back to the next-higher tier — which may be a different provider, giving you cross-provider failover for free. AuthError fails fast. Hooks: on_route (the RoutingDecision), on_complete (the final Response with usage_records), on_error (each failed attempt).

Build a router with missing_provider="degrade" and any tier whose provider isn't configured (e.g. a Grok tier with no XAI_API_KEY) falls back to the nearest available tier instead of erroring — so one config runs across deploys that have different keys.

Cost cache

A [maslul.cache] config returns a prior Response instead of calling a model — exact (identical request) or semantic (nearest request above a cosine threshold, using an embedder you inject, since maslul ships no embeddings). A hit comes back with cached=True and zeroed usage, so monitoring sees the saving. Tool-using requests are never cached.

[maslul.cache]
mode = "semantic"          # off | exact | semantic
max_entries = 1000
ttl_seconds = 86400
similarity_threshold = 0.95

router = Router.from_toml("maslul.toml", embed=my_async_embed)   # embed only needed for semantic

Prompt caching

A different lever from the cost cache above, and they compose: the cost cache doesn't call the model; prompt caching calls it, but pays ~0.1× for the part of the prompt it has already seen. Ask a second question about a 100k-token PDF and you re-send the whole PDF — Anthropic will serve it from cache for a tenth of the price, but only if you tell it what's stable.

You declare what is stable, never a mechanism. Anthropic gets explicit cache_control breakpoints; Gemini, OpenAI and Grok cache a matching prefix automatically, so what they need is layout — and that's the one lever that works on all four:

from maslul import ContextCache, MediaPart, Message, Request

req = Request(
    messages=[Message(role="user", content="What does clause 4 say?")],
    system=[PERSONA],
    media=[MediaPart(mime_type="application/pdf", data=pdf_bytes)],
    context_cache=ContextCache(media=True, ttl_seconds=3600, key=f"doc-{doc_id}"),
)
resp = await router.complete(req, model="anthropic:claude-sonnet-4-6")   # pin: caches are model-scoped
print(resp.usage.cache_read_input_tokens)   # the only proof that any of it worked

The biggest win costs nothing at runtime. Media used to be attached to the last user message on every provider, after that message's text — the most volatile slot in the prompt. So the document sat behind the question, every new question produced a different prefix, and a cache could never reach it. media=True moves the document to the first user message and ahead of the question, into the prefix every provider keys its cache off. Measured live on a 79k-token PDF: a follow-up question went $0.237 → $0.024.

Two things to know. Caches are model-scoped and the router picks the model — a cache written on simple is cold on hard, so pair media=True with a pinned model; maslul emits exactly what you ask for and won't silently drop it. And Usage's input fields are disjoint — input_tokens is what you paid full price for; the prompt's true size is input_tokens + cache_read + cache_creation.

Configuration

A TOML file (or a plain dict — Router(config={...})):

[maslul]
strategy = "route_default"        # route_default | classify | classify_and_answer | verify_cascade
default_level = "hard"            # default-to-capable for the ambiguous middle
min_tokens_to_classify = 40       # CLASSIFY budget guard
request_timeout = 60              # per-call seconds (optional)
max_retries = 2
fallback = true                   # escalate to a higher tier on persistent failure

[maslul.tiers.simple]
provider = "gemini"
model = "gemini-2.5-flash-lite"
[maslul.tiers.medium]
model = "anthropic:claude-haiku-4-5"   # or the provider:model shorthand
[maslul.tiers.hard]
model = "anthropic:claude-sonnet-4-6"

[maslul.classifier]               # required for the classify strategies
model = "anthropic:claude-haiku-4-5"

[maslul.providers.anthropic]
api_key_env = "ANTHROPIC_API_KEY"      # secrets by env-var name, never inlined
[maslul.providers.gemini]
vertex_project = "my-gcp-project"      # Vertex AI + Application Default Credentials (no key)
vertex_location = "global"
[maslul.providers.grok]
api_key_env = "XAI_API_KEY"

Pointing a capability at a different model or provider is a one-line config change — no code deploy. Providers can also be injected directly (Router(config, providers={...})) for tests or custom wiring.

Providers

Status

Beta (0.2.x), fully typed (py.typed), async-first. Routing, tool use, structured output, vision, web search across all three providers (web_search=True), the four strategies, and retry/fallback resilience are implemented and exercised against live APIs.

License

MIT © Ilia Tankelevich