08 · Use cases 6 & 7: the model router and the tool picker¶

These are two of the most valuable decision points in production, and both happen before the LLM runs:

• Use case 6, the model router. Most requests don't need your most expensive model. A Choice over tiers sends each request to the cheapest model that can handle it, and a low-confidence answer falls back to the capable tier.
• Use case 7, the tool picker. Agents connected to many MCP servers can have hundreds of tools. Stuffing all of them into every prompt costs tokens and confuses the model. A Choice over the catalog (up to 255 options) shortlists the few that matter.

In [1]:

# --- Provider config: ONE OpenRouter key drives both brains ------------------------------
#   slow brain (LLM) -> OpenAI SDK   -> https://openrouter.ai/api/v1  (chat completions + tools)
#   fast brain (Jev) -> TypeSafe SDK -> https://openrouter.ai/api     (POST /v1/systemone)
import os, json, time

# Load settings from a .env file if present (falls back to existing env vars).
try:
    from dotenv import load_dotenv, find_dotenv
    load_dotenv(find_dotenv(usecwd=True))
except Exception:
    if os.path.exists(".env"):
        for _line in open(".env"):
            _line = _line.strip()
            if _line and not _line.startswith("#") and "=" in _line:
                _k, _v = _line.split("=", 1)
                os.environ.setdefault(_k.strip(), _v.strip())

BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1")
API_KEY  = os.environ.get("OPENAI_API_KEY", "set-me")
MODEL    = os.environ.get("MODEL", "openai/gpt-6-luna")          # slow brain: any tool-capable chat model

JEV_BACKEND  = os.environ.get("JEV_BACKEND", "typesafe").strip().lower()  # typesafe | adapter
JEV_MODEL    = os.environ.get("JEV_MODEL", "~typesafe/jev-latest")
JEV_BASE_URL = os.environ.get("TYPESAFE_BASE_URL", "https://openrouter.ai/api")
JEV_API_KEY  = os.environ.get("TYPESAFE_API_KEY") or API_KEY          # one key for both brains

# Behind a TLS-intercepting firewall/proxy, set VERIFY_SSL=false in .env (trusted networks only).
import httpx, httpx2
VERIFY_SSL = os.environ.get("VERIFY_SSL", "true").strip().lower() not in ("false", "0", "no")
if not VERIFY_SSL:
    import warnings
    warnings.filterwarnings("ignore")
    print("\u26a0\ufe0f  SSL verification DISABLED (VERIFY_SSL=false) \u2014 use only on a trusted network")

from openai import OpenAI
client = OpenAI(base_url=BASE_URL, api_key=API_KEY, http_client=httpx.Client(verify=VERIFY_SSL))

if JEV_BACKEND == "adapter":
    # No Jev access yet? Same System One API, answered by your LLM (slower, pricier, but it runs).
    from system_one_adapter import SystemOneAdapterClient, Noul, Choice, Score
    from system_one_adapter.providers.openai import OpenAIProvider
    jev = SystemOneAdapterClient(
        structured_outputs=True, llm_answer_mode="probabilities", normalize_probabilities=True,
        n_retry_malformed_structure=2,
        model=OpenAIProvider(MODEL, base_url=BASE_URL, api_key=API_KEY, api="chat_completions"))
else:
    from typesafe_sdk import TypeSafeClient, Noul, Choice, Score
    jev = TypeSafeClient(api_key=JEV_API_KEY, base_url=JEV_BASE_URL, model=JEV_MODEL,
                         http_client=httpx2.Client(verify=VERIFY_SSL))

print(f"slow brain (LLM): {MODEL}  @ {BASE_URL}")
print(f"fast brain (Jev): {JEV_MODEL if JEV_BACKEND != 'adapter' else MODEL + ' via adapter'}  @ {JEV_BASE_URL if JEV_BACKEND != 'adapter' else BASE_URL}")

# --- Provider config: ONE OpenRouter key drives both brains ------------------------------ # slow brain (LLM) -> OpenAI SDK -> https://openrouter.ai/api/v1 (chat completions + tools) # fast brain (Jev) -> TypeSafe SDK -> https://openrouter.ai/api (POST /v1/systemone) import os, json, time # Load settings from a .env file if present (falls back to existing env vars). try: from dotenv import load_dotenv, find_dotenv load_dotenv(find_dotenv(usecwd=True)) except Exception: if os.path.exists(".env"): for _line in open(".env"): _line = _line.strip() if _line and not _line.startswith("#") and "=" in _line: _k, _v = _line.split("=", 1) os.environ.setdefault(_k.strip(), _v.strip()) BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1") API_KEY = os.environ.get("OPENAI_API_KEY", "set-me") MODEL = os.environ.get("MODEL", "openai/gpt-6-luna") # slow brain: any tool-capable chat model JEV_BACKEND = os.environ.get("JEV_BACKEND", "typesafe").strip().lower() # typesafe | adapter JEV_MODEL = os.environ.get("JEV_MODEL", "~typesafe/jev-latest") JEV_BASE_URL = os.environ.get("TYPESAFE_BASE_URL", "https://openrouter.ai/api") JEV_API_KEY = os.environ.get("TYPESAFE_API_KEY") or API_KEY # one key for both brains # Behind a TLS-intercepting firewall/proxy, set VERIFY_SSL=false in .env (trusted networks only). import httpx, httpx2 VERIFY_SSL = os.environ.get("VERIFY_SSL", "true").strip().lower() not in ("false", "0", "no") if not VERIFY_SSL: import warnings warnings.filterwarnings("ignore") print("\u26a0\ufe0f SSL verification DISABLED (VERIFY_SSL=false) \u2014 use only on a trusted network") from openai import OpenAI client = OpenAI(base_url=BASE_URL, api_key=API_KEY, http_client=httpx.Client(verify=VERIFY_SSL)) if JEV_BACKEND == "adapter": # No Jev access yet? Same System One API, answered by your LLM (slower, pricier, but it runs). from system_one_adapter import SystemOneAdapterClient, Noul, Choice, Score from system_one_adapter.providers.openai import OpenAIProvider jev = SystemOneAdapterClient( structured_outputs=True, llm_answer_mode="probabilities", normalize_probabilities=True, n_retry_malformed_structure=2, model=OpenAIProvider(MODEL, base_url=BASE_URL, api_key=API_KEY, api="chat_completions")) else: from typesafe_sdk import TypeSafeClient, Noul, Choice, Score jev = TypeSafeClient(api_key=JEV_API_KEY, base_url=JEV_BASE_URL, model=JEV_MODEL, http_client=httpx2.Client(verify=VERIFY_SSL)) print(f"slow brain (LLM): {MODEL} @ {BASE_URL}") print(f"fast brain (Jev): {JEV_MODEL if JEV_BACKEND != 'adapter' else MODEL + ' via adapter'} @ {JEV_BASE_URL if JEV_BACKEND != 'adapter' else BASE_URL}")

slow brain (LLM): openai/gpt-6-luna  @ https://openrouter.ai/api/v1
fast brain (Jev): ~typesafe/jev-latest  @ https://openrouter.ai/api

In [2]:

# --- Helpers used in every notebook: ask the fast brain, ask the slow brain, track spend ---
class Spend:
    """Running tally of what OpenRouter charged, per brain (it reports exact USD per call)."""
    def __init__(self):
        self.usd, self.calls = {"jev": 0.0, "llm": 0.0}, {"jev": 0, "llm": 0}
    def add(self, brain, resp):
        self.calls[brain] += 1
        try:    cost = resp.raw_http_response.json()["usage"].get("cost")      # Jev response
        except Exception: cost = getattr(getattr(resp, "usage", None), "cost", None)  # LLM response
        self.usd[brain] += cost or 0.0
    def __repr__(self):
        return (f"Jev: {self.calls['jev']} calls ${self.usd['jev']:.5f}  |  "
                f"LLM: {self.calls['llm']} calls ${self.usd['llm']:.5f}")
SPEND = Spend()

def ask_jev(state, questions):
    """Fast brain. state (text or JSON) + typed questions -> typed, calibrated answers."""
    r = jev.system_one(state, questions)
    SPEND.add("jev", r)
    return r

def chat(prompt, system="Be concise.", model=None):
    """Slow brain, single shot (no tools). Returns the reply text."""
    resp = client.chat.completions.create(model=model or MODEL, messages=[
        {"role": "system", "content": system}, {"role": "user", "content": prompt}])
    SPEND.add("llm", resp)
    return resp.choices[0].message.content

def show(r):
    """Print every answer in a Jev response on one line each."""
    for k, a in r.answers.items():
        if a.type == "noul":
            print(f"  {k:<18} noul   P(yes)={a.noul:.2f}")
        elif a.type == "choice":
            top = sorted(a.probabilities.items(), key=lambda kv: -kv[1])[:3]
            print(f"  {k:<18} choice {a.choice!r:<22} conf={a.confidence:.2f}  top={top}")
        else:
            level = {int(i): v for i, v in (a.legend or {}).items()}.get(round(a.score), "")
            print(f"  {k:<18} score  {a.score:.2f} -> {level!r:<18} conf={a.confidence:.2f}")

# --- Helpers used in every notebook: ask the fast brain, ask the slow brain, track spend --- class Spend: """Running tally of what OpenRouter charged, per brain (it reports exact USD per call).""" def __init__(self): self.usd, self.calls = {"jev": 0.0, "llm": 0.0}, {"jev": 0, "llm": 0} def add(self, brain, resp): self.calls[brain] += 1 try: cost = resp.raw_http_response.json()["usage"].get("cost") # Jev response except Exception: cost = getattr(getattr(resp, "usage", None), "cost", None) # LLM response self.usd[brain] += cost or 0.0 def __repr__(self): return (f"Jev: {self.calls['jev']} calls ${self.usd['jev']:.5f} | " f"LLM: {self.calls['llm']} calls ${self.usd['llm']:.5f}") SPEND = Spend() def ask_jev(state, questions): """Fast brain. state (text or JSON) + typed questions -> typed, calibrated answers.""" r = jev.system_one(state, questions) SPEND.add("jev", r) return r def chat(prompt, system="Be concise.", model=None): """Slow brain, single shot (no tools). Returns the reply text.""" resp = client.chat.completions.create(model=model or MODEL, messages=[ {"role": "system", "content": system}, {"role": "user", "content": prompt}]) SPEND.add("llm", resp) return resp.choices[0].message.content def show(r): """Print every answer in a Jev response on one line each.""" for k, a in r.answers.items(): if a.type == "noul": print(f" {k:<18} noul P(yes)={a.noul:.2f}") elif a.type == "choice": top = sorted(a.probabilities.items(), key=lambda kv: -kv[1])[:3] print(f" {k:<18} choice {a.choice!r:<22} conf={a.confidence:.2f} top={top}") else: level = {int(i): v for i, v in (a.legend or {}).items()}.get(round(a.score), "") print(f" {k:<18} score {a.score:.2f} -> {level!r:<18} conf={a.confidence:.2f}")

In [3]:

from concurrent.futures import ThreadPoolExecutor

def jev_map(fn, items, workers=8):
    """Run fn over items in parallel threads. Jev allows 1,200 requests/min, so 8 workers is safe."""
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(fn, items))

from concurrent.futures import ThreadPoolExecutor def jev_map(fn, items, workers=8): """Run fn over items in parallel threads. Jev allows 1,200 requests/min, so 8 workers is safe.""" with ThreadPoolExecutor(max_workers=workers) as pool: return list(pool.map(fn, items))

1. Real prices, fetched live from OpenRouter¶

In [4]:

SMART_MODEL = os.environ.get("SMART_MODEL", "openai/gpt-6-sol")
_models = {m["id"]: m for m in httpx.get("https://openrouter.ai/api/v1/models", timeout=30).json()["data"]}

def price(model):
    p = _models.get(model, {}).get("pricing", {})
    return float(p.get("prompt", 0)) * 1e6, float(p.get("completion", 0)) * 1e6   # USD per million tokens

for m in (MODEL, SMART_MODEL):
    i, o = price(m)
    print(f"{m:<22} ${i:.2f} in / ${o:.2f} out per 1M tokens")

SMART_MODEL = os.environ.get("SMART_MODEL", "openai/gpt-6-sol") _models = {m["id"]: m for m in httpx.get("https://openrouter.ai/api/v1/models", timeout=30).json()["data"]} def price(model): p = _models.get(model, {}).get("pricing", {}) return float(p.get("prompt", 0)) * 1e6, float(p.get("completion", 0)) * 1e6 # USD per million tokens for m in (MODEL, SMART_MODEL): i, o = price(m) print(f"{m:<22} ${i:.2f} in / ${o:.2f} out per 1M tokens")

openai/gpt-6-luna      $0.10 in / $0.50 out per 1M tokens
openai/gpt-6-sol       $2.00 in / $10.00 out per 1M tokens

2. Use case 6: route each request to a tier¶

In [5]:

TIERS = {
    "fast":    "lookups, conversions, short factual answers, formatting, simple single-step tasks",
    "capable": "multi-step reasoning, architecture or design trade-offs, debugging, security or legal judgement, "
               "anything where a wrong answer is costly",
}

def route(query, min_conf=0.7):
    r = ask_jev(query, {"tier": Choice(instructions="Which model tier should answer this request?", criteria=TIERS)})
    a = r.choices["tier"]
    tier = a.choice if a.confidence >= min_conf else "capable"          # unsure -> play it safe
    return {"tier": tier, "jev": a.choice, "conf": round(a.confidence, 2)}

QUERIES = [  # (query, the tier a senior engineer would pick)
    ("What's 18% of 2,450?", "fast"),
    ("Convert this date to ISO format: March 5th 2026", "fast"),
    ("What port does PostgreSQL use by default?", "fast"),
    ("Rewrite this sentence to be more polite: 'send me the file now'", "fast"),
    ("Capitalize every word in: the quick brown fox", "fast"),
    ("What's the capital of Australia?", "fast"),
    ("Translate 'good morning' into Spanish", "fast"),
    ("Our p99 latency doubled after moving to Kubernetes, but CPU is idle. What could cause it and how do I prove it?", "capable"),
    ("Design a multi-region active-active architecture for a payments ledger that never double-spends.", "capable"),
    ("Review this contract clause for liability risks: 'Vendor's total liability shall not exceed fees paid.'", "capable"),
    ("We have 40 microservices and deploys keep breaking each other. Propose a strategy and a migration plan.", "capable"),
    ("Is it safe to store session tokens in localStorage? Walk through the threat model.", "capable"),
    ("Why does this recursive function blow the stack only for inputs above 10,000, and what are the fixes?", "capable"),
    ("Compare event sourcing vs CRUD for our order system given we need a full audit trail.", "capable"),
]
ROUTES = jev_map(lambda q: route(q[0]), QUERIES)
correct = sum(r["tier"] == exp for r, (_, exp) in zip(ROUTES, QUERIES))
for r, (q, exp) in zip(ROUTES, QUERIES):
    print(f"{'ok ' if r['tier'] == exp else 'XX '} {r['tier']:<8} (jev={r['jev']}, conf {r['conf']:.2f})  {q[:70]}")
print(f"\nrouting accuracy {correct}/{len(QUERIES)}")

TIERS = { "fast": "lookups, conversions, short factual answers, formatting, simple single-step tasks", "capable": "multi-step reasoning, architecture or design trade-offs, debugging, security or legal judgement, " "anything where a wrong answer is costly", } def route(query, min_conf=0.7): r = ask_jev(query, {"tier": Choice(instructions="Which model tier should answer this request?", criteria=TIERS)}) a = r.choices["tier"] tier = a.choice if a.confidence >= min_conf else "capable" # unsure -> play it safe return {"tier": tier, "jev": a.choice, "conf": round(a.confidence, 2)} QUERIES = [ # (query, the tier a senior engineer would pick) ("What's 18% of 2,450?", "fast"), ("Convert this date to ISO format: March 5th 2026", "fast"), ("What port does PostgreSQL use by default?", "fast"), ("Rewrite this sentence to be more polite: 'send me the file now'", "fast"), ("Capitalize every word in: the quick brown fox", "fast"), ("What's the capital of Australia?", "fast"), ("Translate 'good morning' into Spanish", "fast"), ("Our p99 latency doubled after moving to Kubernetes, but CPU is idle. What could cause it and how do I prove it?", "capable"), ("Design a multi-region active-active architecture for a payments ledger that never double-spends.", "capable"), ("Review this contract clause for liability risks: 'Vendor's total liability shall not exceed fees paid.'", "capable"), ("We have 40 microservices and deploys keep breaking each other. Propose a strategy and a migration plan.", "capable"), ("Is it safe to store session tokens in localStorage? Walk through the threat model.", "capable"), ("Why does this recursive function blow the stack only for inputs above 10,000, and what are the fixes?", "capable"), ("Compare event sourcing vs CRUD for our order system given we need a full audit trail.", "capable"), ] ROUTES = jev_map(lambda q: route(q[0]), QUERIES) correct = sum(r["tier"] == exp for r, (_, exp) in zip(ROUTES, QUERIES)) for r, (q, exp) in zip(ROUTES, QUERIES): print(f"{'ok ' if r['tier'] == exp else 'XX '} {r['tier']:<8} (jev={r['jev']}, conf {r['conf']:.2f}) {q[:70]}") print(f"\nrouting accuracy {correct}/{len(QUERIES)}")

ok  fast     (jev=fast, conf 1.00)  What's 18% of 2,450?
ok  fast     (jev=fast, conf 1.00)  Convert this date to ISO format: March 5th 2026
ok  fast     (jev=fast, conf 1.00)  What port does PostgreSQL use by default?
ok  fast     (jev=fast, conf 1.00)  Rewrite this sentence to be more polite: 'send me the file now'
ok  fast     (jev=fast, conf 1.00)  Capitalize every word in: the quick brown fox
ok  fast     (jev=fast, conf 1.00)  What's the capital of Australia?
ok  fast     (jev=fast, conf 1.00)  Translate 'good morning' into Spanish
ok  capable  (jev=capable, conf 1.00)  Our p99 latency doubled after moving to Kubernetes, but CPU is idle. W
ok  capable  (jev=capable, conf 1.00)  Design a multi-region active-active architecture for a payments ledger
ok  capable  (jev=capable, conf 1.00)  Review this contract clause for liability risks: 'Vendor's total liabi
ok  capable  (jev=capable, conf 1.00)  We have 40 microservices and deploys keep breaking each other. Propose
ok  capable  (jev=capable, conf 1.00)  Is it safe to store session tokens in localStorage? Walk through the t
ok  capable  (jev=capable, conf 0.99)  Why does this recursive function blow the stack only for inputs above 
ok  capable  (jev=capable, conf 1.00)  Compare event sourcing vs CRUD for our order system given we need a fu

routing accuracy 14/14

What routing saves¶

We assume an average request of 1,000 tokens in and 600 out. The comparison is between "always use the capable model" and "route". The Jev routing calls themselves are included.

In [6]:

def req_cost(model, tin=1000, tout=600):
    i, o = price(model)
    return (tin * i + tout * o) / 1e6

always_smart = len(QUERIES) * req_cost(SMART_MODEL)
routed = sum(req_cost(SMART_MODEL if r["tier"] == "capable" else MODEL) for r in ROUTES) + SPEND.usd["jev"]
print(f"always {SMART_MODEL:<18} ${always_smart:.4f}")
print(f"routed by Jev              ${routed:.4f}   ({1 - routed / always_smart:.0%} cheaper)")
print(f"at 100k requests/day that's ${(always_smart - routed) / len(QUERIES) * 100_000:,.0f}/day saved")

def req_cost(model, tin=1000, tout=600): i, o = price(model) return (tin * i + tout * o) / 1e6 always_smart = len(QUERIES) * req_cost(SMART_MODEL) routed = sum(req_cost(SMART_MODEL if r["tier"] == "capable" else MODEL) for r in ROUTES) + SPEND.usd["jev"] print(f"always {SMART_MODEL:<18} ${always_smart:.4f}") print(f"routed by Jev ${routed:.4f} ({1 - routed / always_smart:.0%} cheaper)") print(f"at 100k requests/day that's ${(always_smart - routed) / len(QUERIES) * 100_000:,.0f}/day saved")

always openai/gpt-6-sol   $0.1120
routed by Jev              $0.0590   (47% cheaper)
at 100k requests/day that's $378/day saved

In [7]:

q = QUERIES[2][0]
r = route(q)
model = SMART_MODEL if r["tier"] == "capable" else MODEL
print(f"{q}\n-> {r['tier']} ({model}): {chat(q, model=model)}")

q = QUERIES[2][0] r = route(q) model = SMART_MODEL if r["tier"] == "capable" else MODEL print(f"{q}\n-> {r['tier']} ({model}): {chat(q, model=model)}")

What port does PostgreSQL use by default?
-> fast (openai/gpt-6-luna): PostgreSQL uses port **5432** by default.

3. Use case 7: pick tools from a big catalog¶

In [8]:

CATALOG = {  # 48 tools across domains - imagine an MCP hub. Descriptions are all the router sees.
    "get_weather": "current weather and forecast for a city",
    "convert_currency": "convert an amount between currencies at today's rate",
    "get_stock_quote": "latest price and daily change for a stock ticker",
    "search_web": "general web search for recent information",
    "send_email": "send an email to a recipient",
    "draft_email": "write an email draft without sending",
    "create_calendar_event": "schedule a meeting or event on the calendar",
    "list_calendar_events": "show upcoming meetings and events",
    "find_free_slot": "find a time when all attendees are free",
    "create_jira_ticket": "open a new issue in the issue tracker",
    "search_jira": "search existing issues in the issue tracker",
    "post_slack_message": "post a message to a Slack channel",
    "search_slack": "search Slack message history",
    "get_pr_status": "check CI status and reviews of a pull request",
    "list_open_prs": "list open pull requests in a repository",
    "run_sql_query": "run a read-only SQL SELECT against the analytics warehouse",
    "describe_table": "show the columns and types of a database table",
    "get_k8s_pods": "list Kubernetes pods and their status in a namespace",
    "get_pod_logs": "fetch recent logs from a Kubernetes pod",
    "restart_deployment": "restart a Kubernetes deployment",
    "scale_deployment": "change the replica count of a Kubernetes deployment",
    "get_cloud_costs": "cloud spend broken down by service for a date range",
    "get_uptime": "uptime percentage of a service over a period",
    "query_metrics": "query time-series metrics like latency or error rate",
    "silence_alert": "silence a monitoring alert for a duration",
    "page_oncall": "page the on-call engineer for a service",
    "get_oncall": "who is on call right now for a team",
    "lookup_customer": "find a customer account by email or id",
    "get_invoice": "fetch an invoice by number",
    "issue_refund": "refund a payment to a customer",
    "get_order_status": "shipping and fulfillment status of an order",
    "update_shipping_address": "change the shipping address on an open order",
    "translate_text": "translate text between languages",
    "summarize_document": "summarize a long document",
    "extract_pdf_tables": "pull tables out of a PDF into rows",
    "ocr_image": "read text from an image",
    "generate_chart": "draw a chart from a small table of numbers",
    "calculate_subnet": "network, broadcast and host count for a CIDR",
    "dns_lookup": "resolve a hostname to IP addresses and records",
    "check_ssl_cert": "expiry date and issuer of a site's TLS certificate",
    "ping_host": "check if a host responds to ping and its latency",
    "get_interface_status": "operational status and errors of a network interface",
    "search_kb": "search the internal knowledge base and runbooks",
    "get_employee": "look up a colleague's title, team and manager",
    "request_pto": "submit a time-off request",
    "book_meeting_room": "reserve a meeting room",
    "get_expense_policy": "rules for what expenses can be reimbursed",
    "submit_expense": "file an expense report with a receipt",
}
print(len(CATALOG), "tools")

CATALOG = { # 48 tools across domains - imagine an MCP hub. Descriptions are all the router sees. "get_weather": "current weather and forecast for a city", "convert_currency": "convert an amount between currencies at today's rate", "get_stock_quote": "latest price and daily change for a stock ticker", "search_web": "general web search for recent information", "send_email": "send an email to a recipient", "draft_email": "write an email draft without sending", "create_calendar_event": "schedule a meeting or event on the calendar", "list_calendar_events": "show upcoming meetings and events", "find_free_slot": "find a time when all attendees are free", "create_jira_ticket": "open a new issue in the issue tracker", "search_jira": "search existing issues in the issue tracker", "post_slack_message": "post a message to a Slack channel", "search_slack": "search Slack message history", "get_pr_status": "check CI status and reviews of a pull request", "list_open_prs": "list open pull requests in a repository", "run_sql_query": "run a read-only SQL SELECT against the analytics warehouse", "describe_table": "show the columns and types of a database table", "get_k8s_pods": "list Kubernetes pods and their status in a namespace", "get_pod_logs": "fetch recent logs from a Kubernetes pod", "restart_deployment": "restart a Kubernetes deployment", "scale_deployment": "change the replica count of a Kubernetes deployment", "get_cloud_costs": "cloud spend broken down by service for a date range", "get_uptime": "uptime percentage of a service over a period", "query_metrics": "query time-series metrics like latency or error rate", "silence_alert": "silence a monitoring alert for a duration", "page_oncall": "page the on-call engineer for a service", "get_oncall": "who is on call right now for a team", "lookup_customer": "find a customer account by email or id", "get_invoice": "fetch an invoice by number", "issue_refund": "refund a payment to a customer", "get_order_status": "shipping and fulfillment status of an order", "update_shipping_address": "change the shipping address on an open order", "translate_text": "translate text between languages", "summarize_document": "summarize a long document", "extract_pdf_tables": "pull tables out of a PDF into rows", "ocr_image": "read text from an image", "generate_chart": "draw a chart from a small table of numbers", "calculate_subnet": "network, broadcast and host count for a CIDR", "dns_lookup": "resolve a hostname to IP addresses and records", "check_ssl_cert": "expiry date and issuer of a site's TLS certificate", "ping_host": "check if a host responds to ping and its latency", "get_interface_status": "operational status and errors of a network interface", "search_kb": "search the internal knowledge base and runbooks", "get_employee": "look up a colleague's title, team and manager", "request_pto": "submit a time-off request", "book_meeting_room": "reserve a meeting room", "get_expense_policy": "rules for what expenses can be reimbursed", "submit_expense": "file an expense report with a receipt", } print(len(CATALOG), "tools")

48 tools

In [9]:

def pick_tools(request, k=5):
    r = ask_jev(request, {"tool": Choice(
        instructions="Which single tool is most useful for handling this request?", criteria=CATALOG)})
    probs = r.choices["tool"].probabilities
    return sorted(probs, key=lambda t: -probs[t])[:k], probs

REQUESTS = [  # (request, the tool that must be in the shortlist)
    ("Is the TLS cert on api.acme.example about to expire?", "check_ssl_cert"),
    ("How much did we spend on the cloud last month, by service?", "get_cloud_costs"),
    ("Why is the checkout pod crash-looping? Show me what it printed.", "get_pod_logs"),
    ("Find a time next week when Priya, Tom and I are all free.", "find_free_slot"),
    ("Who's on call for payments right now?", "get_oncall"),
    ("Customer jane@shop.example says she was charged twice - give her money back.", "issue_refund"),
    ("What columns does the orders table have?", "describe_table"),
    ("Can I expense a co-working day pass?", "get_expense_policy"),
    ("Where's order #55120? It hasn't arrived.", "get_order_status"),
    ("How many usable hosts are in 10.20.0.0/22?", "calculate_subnet"),
]
PICKS = jev_map(lambda r: pick_tools(r[0]), REQUESTS)
top1 = sum(p[0][0] == want for p, (_, want) in zip(PICKS, REQUESTS))
top5 = sum(want in p[0] for p, (_, want) in zip(PICKS, REQUESTS))
for (short, _), (req, want) in zip(PICKS, REQUESTS):
    print(f"{'ok ' if want in short else 'XX '} {req[:55]:<57} -> {short[:3]}")
print(f"\nrecall@1 {top1}/{len(REQUESTS)}   recall@5 {top5}/{len(REQUESTS)}")
assert top5 >= len(REQUESTS) - 1

def pick_tools(request, k=5): r = ask_jev(request, {"tool": Choice( instructions="Which single tool is most useful for handling this request?", criteria=CATALOG)}) probs = r.choices["tool"].probabilities return sorted(probs, key=lambda t: -probs[t])[:k], probs REQUESTS = [ # (request, the tool that must be in the shortlist) ("Is the TLS cert on api.acme.example about to expire?", "check_ssl_cert"), ("How much did we spend on the cloud last month, by service?", "get_cloud_costs"), ("Why is the checkout pod crash-looping? Show me what it printed.", "get_pod_logs"), ("Find a time next week when Priya, Tom and I are all free.", "find_free_slot"), ("Who's on call for payments right now?", "get_oncall"), ("Customer jane@shop.example says she was charged twice - give her money back.", "issue_refund"), ("What columns does the orders table have?", "describe_table"), ("Can I expense a co-working day pass?", "get_expense_policy"), ("Where's order #55120? It hasn't arrived.", "get_order_status"), ("How many usable hosts are in 10.20.0.0/22?", "calculate_subnet"), ] PICKS = jev_map(lambda r: pick_tools(r[0]), REQUESTS) top1 = sum(p[0][0] == want for p, (_, want) in zip(PICKS, REQUESTS)) top5 = sum(want in p[0] for p, (_, want) in zip(PICKS, REQUESTS)) for (short, _), (req, want) in zip(PICKS, REQUESTS): print(f"{'ok ' if want in short else 'XX '} {req[:55]:<57} -> {short[:3]}") print(f"\nrecall@1 {top1}/{len(REQUESTS)} recall@5 {top5}/{len(REQUESTS)}") assert top5 >= len(REQUESTS) - 1

ok  Is the TLS cert on api.acme.example about to expire?      -> ['check_ssl_cert', 'create_jira_ticket', 'submit_expense']
ok  How much did we spend on the cloud last month, by servi   -> ['get_cloud_costs', 'draft_email', 'restart_deployment']
ok  Why is the checkout pod crash-looping? Show me what it    -> ['get_pod_logs', 'create_jira_ticket', 'get_uptime']
ok  Find a time next week when Priya, Tom and I are all fre   -> ['find_free_slot', 'get_order_status', 'submit_expense']
ok  Who's on call for payments right now?                     -> ['get_oncall', 'summarize_document', 'list_open_prs']
ok  Customer jane@shop.example says she was charged twice -   -> ['issue_refund', 'lookup_customer', 'get_oncall']
ok  What columns does the orders table have?                  -> ['describe_table', 'ping_host', 'create_jira_ticket']
ok  Can I expense a co-working day pass?                      -> ['get_expense_policy', 'get_employee', 'check_ssl_cert']
ok  Where's order #55120? It hasn't arrived.                  -> ['get_order_status', 'get_expense_policy', 'search_web']
ok  How many usable hosts are in 10.20.0.0/22?                -> ['calculate_subnet', 'get_weather', 'page_oncall']

recall@1 10/10   recall@5 10/10

Hand the LLM only the shortlist¶

We compare prompt size for all 48 tool schemas against the shortlist of 5, then let the LLM make the actual call with the smaller menu. The tools are stubs here; in production they would be your MCP servers.

In [10]:

def schema(name):
    return {"type": "function", "function": {"name": name, "description": CATALOG[name],
            "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}}

all_tokens = len(json.dumps([schema(t) for t in CATALOG])) // 4
short, _ = pick_tools(REQUESTS[0][0])
short_tokens = len(json.dumps([schema(t) for t in short])) // 4
print(f"tool-schema tokens per call: all {len(CATALOG)} tools ~{all_tokens}  vs  shortlist ~{short_tokens}  "
      f"({1 - short_tokens / all_tokens:.0%} fewer, on EVERY step of the loop)")

resp = client.chat.completions.create(model=MODEL, tools=[schema(t) for t in short],
                                      messages=[{"role": "user", "content": REQUESTS[0][0]}])
SPEND.add("llm", resp)
tc = resp.choices[0].message.tool_calls
print("LLM called:", [(c.function.name, c.function.arguments) for c in tc] if tc else resp.choices[0].message.content)

def schema(name): return {"type": "function", "function": {"name": name, "description": CATALOG[name], "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}} all_tokens = len(json.dumps([schema(t) for t in CATALOG])) // 4 short, _ = pick_tools(REQUESTS[0][0]) short_tokens = len(json.dumps([schema(t) for t in short])) // 4 print(f"tool-schema tokens per call: all {len(CATALOG)} tools ~{all_tokens} vs shortlist ~{short_tokens} " f"({1 - short_tokens / all_tokens:.0%} fewer, on EVERY step of the loop)") resp = client.chat.completions.create(model=MODEL, tools=[schema(t) for t in short], messages=[{"role": "user", "content": REQUESTS[0][0]}]) SPEND.add("llm", resp) tc = resp.choices[0].message.tool_calls print("LLM called:", [(c.function.name, c.function.arguments) for c in tc] if tc else resp.choices[0].message.content)

tool-schema tokens per call: all 48 tools ~2674  vs  shortlist ~277  (90% fewer, on EVERY step of the loop)

LLM called: [('check_ssl_cert', '{"query":"api.acme.example"}')]

In [11]:

print(SPEND)

print(SPEND)

Jev: 26 calls $0.00082  |  LLM: 2 calls $0.00004

Recap¶

• Router: a Choice over tiers plus a confidence fallback. You pay for the capable model only where it earns its price.
• Tool picker: a Choice over the catalog (up to 255 options), then keep the top k by probability. The prompt is smaller, the model gets less confused, and every step of the loop gets cheaper.
• Both decisions take ~300 ms and cost a fraction of a cent. The LLM never has to make them.

Next: 09_oncall_log_triage.ipynb. It's 3 a.m., the pager fires and there are thousands of log lines: Jev ranks them, the LLM finds the root cause.