SDK Guide

Stability & versioning

• SemVer. Everything exported from opensmartroute (top-level) is public API. opensmartroute.enterprise, .security, .learning, .realtime, .math, .eval, .aio, .adapters, .config are public too but may add fields between minor versions.
• Modules prefixed _ and anything not in an __all__ are private.
• The complete list of public modules and names, with one-line summaries, is generated into Python API reference from the source.
• Deprecations are announced one minor version ahead with DeprecationWarning.

Installation matrix

curl -LsSf https://opensmartroute.ai/install.sh | sh (Linux, macOS) and irm https://opensmartroute.ai/install.ps1 | iex (Windows) install the osr CLI with the yaml and server extras in an isolated tool environment (uv, pipx or a private venv). For the library, pick the extras you need:

osr doctor prints which extras are present, the bundled routing model, the config directory and the sign-in, and with --targets / --rules whether a catalogue loads; it exits 1 when something required is missing.

Two ways in: a router in your process, or the platform over HTTP

In your process. Router.starter() is a complete router with nothing configured - the same three targets and two rules osr route uses without --targets (a small fast model, a frontier model, an on-premises one; personal data stays on-prem, hard requests go to the frontier). Decisions take a fraction of a millisecond and nothing leaves the machine:

from opensmartroute import Router

router = Router.starter()
d = router.route("Prove that sqrt(2) is irrational, step by step.")
print(d.target.id, round(d.confidence, 2))  # llm-frontier 0.9x
print(d.trace.explain())

Router.from_files("targets.yaml", "rules.yaml") loads your own catalogue; osr stack init --out stack.json writes the starter one to disk to edit.

Over HTTP. OpenSmartRoute is the client of the hosted platform and of a self-hosted osr serve: the same surface as the TypeScript SDK, standard library only. With no arguments it reads OSR_API_KEY / OSR_API_URL, then the profile osr login saved; without any key it can still quote (estimate()) and read the deployment (info()).

from opensmartroute import OpenSmartRoute, RateLimitError

osr = OpenSmartRoute()  # or OpenSmartRoute(api_key="osr_live_...", base_url="https://osr.example.com")
d = osr.route(
    "Refund order #123, the customer is upset",
    constraints={"data_boundary": "private"},
    objective={"cost": 0.4, "quality": 0.6},
)
print(d.target_id, d.confidence, d.explanation)  # d.raw holds everything the server sent (d["ranked"] ...)

answer = osr.route("Summarise this ticket: ...", execute=True).text  # run the chosen target too
osr.feedback(d, success=True, quality=0.9)  # close the loop: the learners improve on your traffic

print(osr.ask("What is the capital of France?"))  # OpenAI-compatible chat with model="auto"
for chunk in osr.chat("Write a haiku about routers", stream=True):
    print(chunk["choices"][0]["delta"].get("content", ""), end="")

quote = osr.estimate("Translate this contract", monthly_requests=10_000)  # no key needed
print(quote["recommended"]["target_id"], quote["monthly"]["savings_usd"])

What the client does on its own: a timeout on every call (OSR_CLIENT_TIMEOUT_S, 60 s), bounded retries with backoff after a connection failure, a 408, a 429 or a 502-504 - Retry-After is honoured, a 4xx that is final is never retried (OSR_CLIENT_MAX_RETRIES, 2) - and typed errors: AuthenticationError (401/403, says how to sign in), RateLimitError (429, with .retry_after), APIError (.status, .detail, .request_id for a support ticket). No error message and no repr() ever contains the key. An API key is refused over plain http:// to anything but a private host (loopback, RFC 1918, a bare service name, .internal, .svc ...) unless you pass allow_insecure_http=True or set OSR_CLIENT_ALLOW_INSECURE_HTTP=1. An osr_local_... token puts the REST routes at the root for osr serve (api_prefix="" does the same by hand); tenant= and app= are sent on every call; osr.rest-style access is osr.get(path) / osr.post(path, body) for anything not wrapped. Every tunable is a ClientSettings field (OSR_CLIENT_*).

Connecting real providers

adapters.OpenAICompatClient is a stdlib HTTP client for any OpenAI-compatible API (OpenAI, Azure OpenAI, vLLM, Ollama, LiteLLM, OpenRouter, Groq, Mistral). It has explicit timeouts, exponential backoff on 408/429/5xx, and maps failures to SDK errors (ConfigurationError for 401/403, TargetUnavailableError for connectivity/5xx, ExecutionError for the rest). The API key is read from OPENAI_API_KEY (or OPENAI_API_KEY_FILE) and never appears in logs or exceptions.

from opensmartroute.adapters import OpenAICompatClient, chat_handler, embedder, judge_fn

openai = OpenAICompatClient()  # https://api.openai.com/v1
local = OpenAICompatClient("http://localhost:11434/v1", api_key="ollama")  # Ollama

registry = TargetRegistry(
    [
        RouteTarget(
            "gpt-4o-mini",
            TargetKind.LLM,
            cost={"usd_per_1k_tokens": 0.00015},
            handler=chat_handler(openai, model="gpt-4o-mini"),
        ),
        RouteTarget(
            "llama-local",
            TargetKind.LLM,
            cost={"usd_per_1k_tokens": 0.0},
            constraints=TargetConstraints(data_boundary="on_prem"),
            handler=chat_handler(local, model="llama3.1"),
        ),
    ]
)

router = Router(
    registry,
    strategies=[CapabilityStrategy(), SimilarityStrategy(embedder=embedder(openai, model="text-embedding-3-small"))],
    llm_judge=LLMJudgeStrategy(judge_fn(openai, model="gpt-4o-mini")),
    escalate_llm_judge_below=0.5,
)

decision = router.route(request)
result = router.execute(decision, request)  # ChatResult(text, tokens, latency_ms); outcome auto-recorded

chat_handler builds the message list from request.history (user, assistant - including tool_calls - and tool turns are kept verbatim), and uses request.context["system"] as the system prompt — which is how a persona slot in a RoutePlan changes model behaviour without a separate deployment. It forwards context["openai"] (tools, tool_choice, response_format, stop, seed, ...) and context["on_delta"] (a streaming callback) to the upstream, and the ChatResult carries tool_calls, finish_reason and refusal beside the text and token counts. embed_handler(client, model) is the counterpart for embeddings-tagged targets (what answers /v1/embeddings), and attach_chat_handlers(targets, client) gives every LLM target the right one from metadata.model - the call behind OSR_LLM_BASE_URL in osr serve (an api.anthropic.com base URL there picks the Anthropic client automatically).

adapters.AnthropicClient speaks Anthropic's Messages API natively behind the same chat / stream_chat surface, so chat_handler, judge_fn and every other adapter route to Claude without an OpenAI-compatible gateway in between: chat-shaped messages, OpenAI function specs, tool_choice, tool results, images and streaming are translated to POST /v1/messages (x-api-key + anthropic-version headers) and back; reasoning_effort (or an explicit thinking dict) turns on extended thinking and the thinking text lands on ChatResult.reasoning. Fields the Messages API has no equivalent for are dropped rather than sent to fail, and embed() raises a clear error (Anthropic serves no embeddings endpoint).

from opensmartroute.adapters import AnthropicClient, chat_handler

claude = AnthropicClient()  # https://api.anthropic.com/v1, key from ANTHROPIC_API_KEY
target = RouteTarget("claude", TargetKind.LLM, handler=chat_handler(claude, model="claude-sonnet-4-5"))

Optional adapters (lazy-imported):

from opensmartroute.adapters import sentence_transformers_embedder, OpenTelemetryTelemetry

SimilarityStrategy(embedder=sentence_transformers_embedder("all-MiniLM-L6-v2"))
RouterBuilder(reg).with_telemetry(OpenTelemetryTelemetry())

TensorFlow / Keras models (pip install 'opensmartroute[tensorflow]', opensmartroute.learning.neural) sharpen two learnable parts of every request without touching the zero-dependency core:

from opensmartroute.estimate import estimate
from opensmartroute.learning import NeuralRoutingHead, RouterSLM, TokenCountModel

# tokenisation: learn the residual between the heuristic and the provider's billed prompt_tokens
tokens = TokenCountModel(family="o200k")
report = tokens.fit(pairs)  # [(text, billed_tokens), ...] from your usage rows
report["mape_heuristic"], report["mape_model"]
tokens.save("models/tokens-o200k")  # .keras + .json sidecar
TokenCountModel.load(
    "models/tokens-o200k"
).install()  # estimate_tokens(), the admission bound and quotes count through it

# the decision: a dense head over the SLM's hashed features, and its representation as the SLM's dense embedder
head = NeuralRoutingHead(hidden=64)
head.fit(rows, targets=router.registry.all())  # the same EvalRows `osr slm train` reads
head.predict_proba("Refactor this function", candidates=["coder", "writer"])
head.predict_proba("why?", history=[{"role": "user", "content": "prove that sqrt 2 is irrational"}])  # context
head.observe("why?", "frontier", reward=1.0, history=...)  # experience replay: every outcome is remembered ...
head.replay_fit()  # ... and replayed between full fits (saved with the model)
slm = RouterSLM(router.registry.all(), embedder=head.embedder())  # or OSR_SLM_EMBEDDER=keras:models/head

The head reads a follow-up with the user turns it continues, exactly as the SLM does (learning.routing_text; EvalRow.history on training rows), and its learning.ReplayMemory (bounded, oldest dropped first) travels in the .json sidecar, so a restart keeps what live traffic taught.

estimate.set_token_counter(fn) takes any str -> int counter (a tokenizer library works too); a counter that raises falls back to the heuristic. OSR_ESTIMATE_TOKENIZER names one without code: tiktoken:o200k_base (or any tiktoken encoding; pip install 'opensmartroute[tiktoken]') counts exactly the way OpenAI bills, keras:<path> loads a saved TokenCountModel; a counter that cannot be built is logged and the heuristic stays. Feature vectors (learning.text_stats, the hashed featurizer) are pure Python, so a model trained on one machine reads the same features on the router that loads it.

Knowledge graph and context engineering

opensmartroute.knowledge is the SDK's knowledge management layer: a KnowledgeGraph of typed, sourced, weighted, expiring facts (JSONL on disk; JSON-LD, N-Triples, GraphML, Cypher and NetworkX out, JSON-LD in), rule-based ingestion of what a person states, of the catalogue, of decisions, outcomes and live web hits, and the ContextEngineer that chooses, ranks and trims what a model should know for one request into a token budget with full accounting (User guide). RouterBuilder.with_knowledge() wires the ContextMiddleware and the KnowledgeStrategy; the graph is router.knowledge.

from opensmartroute.knowledge import ContextEngineer, KnowledgeGraph, ingest_targets, ingest_turn
from opensmartroute.learning import KnowledgeEmbedding  # TensorFlow to train, numpy to use

g = KnowledgeGraph("state/knowledge.jsonl")
ingest_targets(g, registry.all())
ingest_turn(g, "person:u1", "I work at Acme and we use Python and dbt")
ctx = ContextEngineer(g, budget_tokens=400).build(RouteRequest(text="test a dbt model"), subject="person:u1")
ctx.text, ctx.tokens, ctx.sections(), [ln.reason for ln in ctx.dropped]

emb = KnowledgeEmbedding(dim=32)
report = emb.fit(g)  # TransE over the shared facts: {mrr, hits_at_10, ...}
g.embedding = emb  # KnowledgeGraph.suggest() asks it first
emb.predict_tail("target:llm-small", "has_domain")  # [("domain:finance", 0.81), ...]
emb.save("models/knowledge")
KnowledgeEmbedding.load("models/knowledge")  # .npz + .json, numpy only

Loading configuration

from opensmartroute.config import load_targets, load_rules

registry = load_targets("targets.yaml")  # or .json
rules = load_rules("rules.yaml")

Both raise ConfigurationError with the file, the offending entry and the reason. Only yaml.safe_load is used.

Executing end to end

Router.route() returns a decision; Router.run() (also on AsyncRouter and EnterpriseRouter) routes and executes the resulting plan, returning an ExecutionResult. execute(decision, request) is still available and returns just the primary handler's raw response.

res = router.run("Write SQL for monthly active users by region", task_id="job-17")
res.text  # best-effort text of the primary response
res.response  # the raw handler response (ChatResult, HarnessResult, dict, str, ...)
res.target_id  # primary target
res.steps  # [ExecutionStep(role, target_id, latency_ms, ok, note), ...]
res.system_prompt  # what was composed into context["system"] (None if nothing was added)
res.effective_request  # the RouteRequest the primary handler actually received
res.outcomes  # one Outcome per participant, already passed to router.learn()

What run() does, in order:

1. route(request, plan=True) builds the plan: primary + optional persona and skill slots. A slot is only filled when its best candidate's quality estimate is at least Router(slot_quality_floor=0.5) and (for run) its confidence is at least min_slot_confidence.
2. Skill pre-processing. If the skill slot's target has a handler, it is called with the request and may return
   • a RouteRequest — replaces the request the primary sees (rewrite / enrich);
   • a str — attached as context["skill_output"] and appended to the system prompt;
   • None — the skill contributes instructions only. Exceptions are recorded as a failed step (and a failed Outcome) but do not abort the run.
3. Prompt composition. context["system"] is set to: any existing system + # Persona: <name>
   • RouteTarget.instructions of the persona + # Skill: <name> + skill instructions + the primary's own instructions (for llm/agent targets) + # Skill output. chat_handler and the harness adapters read this key.
4. Primary call. target.handler(request, **kw). Missing handler → ExecutionError. In the sync path a coroutine result raises ExecutionError("use aexecute()"); AsyncRouter.run() awaits both skill and primary handlers.
5. Outcomes. Success is read from response.success / response["success"] (default True); cost from response.cost_usd or total_tokens × unit_cost; quality from response.quality. One Outcome is created for the primary (role=None) and one per slot (role="persona"|"skill"), all carrying task_id (argument or request.context["task_id"]) so delayed feedback can be joined later. Each is passed to learn; on a handler exception, failure outcomes are recorded and the exception re-raised.

Targets carry instructions

- id: persona-legal-counsel
  kind: persona
  primary: false
  instructions: |
    You are cautious in-house legal counsel. Cite the governing clause...
- id: skill-sql
  kind: skill
  instructions: "Always qualify table names with the schema; never SELECT *."

instructions is the persona's system prompt, a skill's body (SKILL.md style) or an agent's standing instructions. It is disclosed to the model only when the target is part of the plan.

Agent harnesses

An agent harness is any runtime that takes a task and drives its own loop (coding agent, research agent, OpenHands, SWE-agent, Claude Code, LangGraph graph, ...). Adapters live in opensmartroute.adapters and all produce a HarnessResult(text, success, steps, prompt_tokens, completion_tokens, cost_usd, latency_ms, quality, artifacts, raw):

from opensmartroute.adapters import CallableHarness, HTTPHarness, SubprocessHarness, harness_handler

# in-process: fn(task, context, history) -> str | dict | HarnessResult
registry.get("research-agent").handler = harness_handler(CallableHarness(my_graph.invoke))

# HTTP: POST {"task", "context", "history"} -> JSON {"text"|"output", "success", "usage", "cost_usd"}
registry.get("support-agent").handler = harness_handler(
    HTTPHarness("https://agents.internal/support/run", api_key_env="AGENT_TOKEN", timeout_s=300)
)

# CLI: task on stdin, answer on stdout (or JSON with json_output=True); system prompt via a flag
registry.get("coding-agent").handler = harness_handler(
    SubprocessHarness(["my-agent", "--repo", "."], system_flag="--system", timeout_s=900)
)

Non-zero exit codes and "success": false become failed outcomes; HTTP 408/429/5xx, timeouts and unreachable hosts raise TargetUnavailableError (so HealthRegistry breakers trip); other HTTP errors raise ExecutionError; a missing binary or non-http URL raises ConfigurationError. Implement the AgentHarness protocol (run(task, *, context, history) -> HarnessResult) for anything else.

Skills from SKILL.md

Folders following the Agent Skills layout load directly:

from opensmartroute.adapters import load_skill, load_skills, skill_from_markdown

for skill in load_skills("skills/"):  # skills/<name>/SKILL.md, name must match the folder
    registry.add(skill)

Frontmatter name/description become id/examples; the Markdown body becomes instructions (progressive disclosure: only the description is used for routing, the body is shown to the model once selected). Routing metadata can be added under metadata: — osr-domains, osr-actions, osr-languages, osr-tags, osr-quality-prior, osr-primary ("false" for slot-only skills), osr-latency-ms. license, compatibility, allowed-tools and other metadata are preserved in RouteTarget.metadata. Give the returned target a handler to make it an executable pre-processor or primary skill.

Learning from outcomes

Every learner (bandits, IRT, preference, LinUCB, Markov, task table, judge calibration) consumes the same Outcome:

from opensmartroute import Outcome

d = app.route(req)
resp = call_provider(d.target, req)
app.learn(
    Outcome(
        request_id=req.request_id,
        target_id=d.target.id,
        success=True,
        quality=grade(resp),
        cost_usd=resp.cost,
        latency_ms=resp.ms,
    )
)

• Plans: Router.run() records one Outcome per participant with role="skill" | "persona" | "primary"; learners keep a target's record as a skill separate from its record as the primary answerer (signals.extra["role"]).
• Cascades: Cascade.run(request, d.trace.ranked) returns a CascadeResult; feed every step to the learners with app.learn_cascade(request, result) (rejected steps are failures with the gate score as quality, the accepted step a success). result.total_cost_usd is the spend across steps.
• Delayed / task-level rewards: set Outcome.task_id (and step) so a reward observed at the end of a multi-step task credits every call in the trajectory (Router(task_credit=...)).
• Offline warm start: learning.warm_start_from_matrix(rows, strategies, weight=0.5) replays a full-information reward matrix (prompt, {target_id: reward}) from an offline evaluation into every learner before serving traffic; learning.warm_start() seeds a brand-new target from its neighbours.
• Replicas: learning.merge_learners(local, remote) folds another replica's learners into yours (sufficient statistics only, no prompts); AutoLearner.refresh() re-loads a writer's snapshot from a shared StateStore. Corrupt state is quarantined on load (AutoLearner.quarantined) instead of raising.
• Objective: Objective(quality, cost, latency, quality_floor, energy, carbon); energy and carbon weigh RouteTarget.unit_energy (cost["wh_per_1k_tokens"]) and unit_carbon (cost["gco2_per_1k_tokens"] or energy x cost["gco2_per_wh"]).
• Cold-start exploration: a target declared with only id, kind and cost is never ranked first and so never earns an outcome. Router(explore_rate=0.1, explore_min_samples=200) (or RoutingSettings.explore_rate / explore_min_samples) serves the least-seen viable candidate with probability explore_rate until it has that many outcomes; decision.propensities includes the exploration mass (so off-policy estimates stay unbiased) and signals.extra["explored"] marks the request. Off by default.
• Measured exit criteria: opensmartroute.eval.criteria turns each offline-measurable routing guarantee into a simulation of the real router (cold_start_ratio(), effort_token_savings(), multi_round_vs_best_single(), match_at_1_at_scale(), conformal_coverage(), knapsack_never_exceeds_cap(), ope_within_live_ci()) and opensmartroute.eval.agentic adds the tau-bench-style one (AgentTask, load_agentic_tasks(), synthetic_agentic_tasks(), task_routing_frontier(): per-step routing through ProgressRouter against every always-use-agent-X policy on the accuracy-latency frontier). On your own data, osr eval DATASET --multi-round runs the multi-round comparison on scored rows, osr eval DATASET --effort the effort-routing token comparison on rows that carry per-target scores and tokens (EvalRow.tokens), and osr eval TASKS.jsonl --agentic the task-routing frontier on tasks with per-agent per-step success and latency_ms tables.
• Contrastive and policy-gradient learners: learning.ContrastiveRouter().fit(rows, targets) trains on scored EvalRows against each row's acceptable set (objective="distilled" uses reward-softened labels); wrap it in learning.ContrastiveStrategy(model) to keep learning online. learning.PolicyGradientStrategy() learns a routing policy directly from decision_reward() (quality minus weighted cost and latency) with no quality predictor in between; learning.decision_regret(rows, targets, choose, predict) compares decision regret with prediction error on the same rows.
• Headroom before you learn: eval.headroom.routing_headroom(rows, targets) tells you how much of the oracle gap is measurable on your data; target_diversity, min_catalogue, scaling_curve and learnability_by_difficulty answer whether, and where, a learned router can beat the best single target.
• Other target kinds: strategies.SemanticCache.target() turns a semantic cache into a DESTINATION target scored by SemanticCacheStrategy; strategies.AnnotatorPool + HumanRoutingStrategy route among HUMAN annotators with Dawid-Skene skill estimates and select_quorum(); strategies.MemoryRouter routes what to remember across memory tiers under a token budget; strategies.ModalityStrategy / ModalityEscalation handle text-first escalation to multimodal targets; strategies.SpeculativeCascade runs draft and strong models in parallel when the learned acceptance rate makes that cheaper.

Error handling

All exceptions derive from OpenSmartRouteError and carry .code and .details:

from opensmartroute import OpenSmartRouteError, NoRouteError, SecurityError

try:
    d = app.route(req)
except NoRouteError as e:
    # e.details["rejections"] -> {target_id: reason}
    fallback()
except SecurityError as e:
    reject_request(e.to_dict())
except OpenSmartRouteError:
    log.exception("routing failed")

The SDK never raises bare Exception/RuntimeError; strategy failures (e.g. a broken LLM judge) degrade to neutral scores rather than failing the route.

Threading & async

• Router and EnterpriseRouter are safe to share across threads. Router.lock (a re-entrant lock) guards strategy scoring, every learner update (learn, learn_correction, credit_task), task pins, the request memory, retrieval narrowing and the exploration RNG, so a strategy's score never observes a half-applied update. The optional LLM judge is scored outside the lock (network I/O) - put it behind escalate_llm_judge_below so it runs rarely, and set TimeoutMiddleware.
• AutoLearner takes the same lock (EnterpriseRouter / RouterBuilder pass router.lock; pass it yourself when wiring by hand) so learn, reset_target, save, load and refresh serialise against routing. save() deep-copies a sequence-numbered snapshot under the lock and writes it outside; a separate I/O lock orders writers and readers so a slow save never overwrites a newer snapshot and two saves never race on the state files.
• Per-module state elsewhere has its own lock: SemanticCache, MemoryRouter, AnnotatorPool, ModalityEscalation (posterior) and SpeculativeCascade (acceptance posteriors, history, failure counters). Embedding / hashing runs outside those locks.
• SpeculativeCascade runs concurrency speculative requests at once (pool of 2 * concurrency workers); draft_timeout_s abandons a slow draft and serves the strong answer; a draft that raises escalates instead of failing (details["draft_error"]); a strong future that is no longer needed is cancelled or its late failure swallowed. stats() reports runs, modes, failure counts and per-bucket acceptance.
• AsyncRouter wraps any router for await-based apps and awaits coroutine handlers; route / learn run in the default executor under the same lock.
• Routing has no I/O of its own.

Multi-round execution

ProgressRouter.run_step routes through router.route (middleware and telemetry apply) and executes with execution.execute; a handler that raises is recorded as a failed step for its target before the exception propagates, so the next step already denies it after failures_before_switch. Step cost falls back to target.unit_cost x estimated tokens when the handler reports none, so budget_usd is tracked even for plain-string handlers. arun_step is the coroutine twin.

MultiRoundExecutor (route -> execute -> judge -> refine) learns once per round with the judged quality (no provisional learn + correction double count); a round whose handler raised becomes a Round with error set, learns a failure for that target and the loop continues on a different target. stop_reason is one of accepted, budget, deadline (deadline_ms) or max_rounds; return_best=True (default) returns the highest-quality round when nothing reached the threshold; credit_task=True delivers the final quality as the task-level reward through router.credit_task, which re-credits every step (learners then see one judged update per round plus one credited update). arun awaits coroutine handlers and an async judge.

Type checking

The package ships py.typed; mypy --strict-friendly signatures on the public surface.

Configuration

# targets.yaml
targets:
  - id: llm-frontier
    kind: llm
    capabilities: { domains: [math, coding], min_complexity: 0.5, supports_tools: true }
    constraints:  { data_boundary: public, pii_allowed: false, regions: [us, eu] }
    cost: { usd_per_1k_tokens: 0.015 }
    latency_ms: 2500
    quality_prior: 0.93
    examples: ["Prove that sqrt(2) is irrational."]

# rules.yaml  (Arch-Router-style preferences)
rules:
  - name: escalation-intent
    when: { actions: [escalate] }
    prefer: [human-escalation]
    pin: true          # restrict candidates, not just boost

Secrets are never in YAML. Handlers read them at call time:

from opensmartroute.security import load_secret

api_key = load_secret("OPENAI_API_KEY")  # env var or OPENAI_API_KEY_FILE

Extending

Decorator SDK

Every routing component - targets, strategies, signal extractors, policy rules, middleware and telemetry sinks - can be declared with a decorator and assembled by a ComponentRegistry. The decorators are transparent (they return the decorated object unchanged) and the registry stores blueprints (factories), so one declaration can back many independent routers.

import opensmartroute as osr


@osr.tool("weather", domains=["travel"], actions=["lookup"], cost=0.0, examples=["weather in paris"])
def weather(request, **kw):
    """Current weather for a city."""  # first paragraph -> target.description
    return lookup(request.text)


@osr.agent("planner", domains=["travel"], min_complexity=0.3, quality_prior=0.9)
def planner(request, **kw):
    """Plan a multi-day trip."""
    return plan(request.text)


@osr.strategy(weight=0.8)  # fn(request, signals, candidates) -> {id: score}
def geo_affinity(request, signals, candidates):
    region = request.constraints.region
    return {t.id: 1.0 if region in t.constraints.regions else 0.4 for t in candidates}


@osr.strategy  # or a Strategy subclass; name inferred -> "recency"
class RecencyStrategy(osr.Strategy):
    def score(self, request, signals, candidates): ...


@osr.signal(order=-1)  # runs before the built-in extractors
def urgency(request, signals):
    return {"urgent": "asap" in request.text.lower()}  # unknown keys land in signals.extra


@osr.policy_rule  # fn(target, request, signals) -> reason | None
def business_hours_only(target, request, signals):
    if target.kind is osr.TargetKind.HUMAN and request.context.get("after_hours"):
        return "human queues are closed"
    return None


@osr.middleware  # fn(request, next_) -> RouteDecision
def tag_tenant(request, next_):
    request.context.setdefault("tenant", "public")
    return next_(request)


router = osr.components.router()  # Router: targets + strategies + signals + rules
app = osr.components.builder().with_auto_learning().build()  # EnterpriseRouter variant

Most applications need none of the plumbing above. The one-liners below cover the common cases - a rule without a rules.yaml, a hook without a Telemetry subclass, a guard without a middleware chain, code behind a catalogue's targets - and osr.route() / osr.run() route through the registry without building a router by hand:

import opensmartroute as osr

osr.components.load("targets.yaml", "rules.yaml")  # optional: a catalogue and its rules as components


@osr.handler("gpt-4o", "gpt-4o-mini")  # code for targets the catalogue declares
def chat(request, **kw):
    return provider.complete(request.text, model=kw.get("model"))


@osr.persona("legal-counsel")  # first paragraph routes it, the rest is its system prompt
def legal(request, **kw):
    """Contract and compliance questions.

    You are cautious in-house legal counsel. Cite the governing clause before answering."""


osr.rule("pii-stays-home", prefer=["onprem"], pin=True, contains_pii=True)  # what rules.yaml holds, in code


@osr.rule(prefer=["human-desk"], weight=0.8, domains=["support"])  # a predicate when `when` cannot say it
def angry_customer(request, signals):
    return "unacceptable" in request.text.lower()


@osr.guard  # fn(request) -> None (pass) | str / False (refuse: SecurityError) | RouteRequest (rewrite)
def no_empty(request):
    return "empty request" if not request.text.strip() else None


@osr.on_decision  # one hook each; on_outcome(outcome), on_error(request, error) likewise
def log_decision(request, decision):
    log.info("%s -> %s (%.2f)", request.request_id, decision.target.id, decision.confidence)


@osr.signal(name="urgent")  # a scalar lands under the extractor's name in signals.extra
def urgent(request, signals):
    return "asap" in request.text.lower()


d = osr.route("Refund order #123, the customer is upset")  # the registry builds and caches its router
print(d.target.id, d.trace.explain())
osr.feedback(d, success=True, quality=0.9)  # Outcome built from the decision and learned
res = osr.run("Summarise this ticket: ...")  # route + execute the handler + learn
osr.feedback(res, quality=grade(res.text))  # a later judgement of a run is recorded as a correction

d = await osr.aroute("...")  # the same loop for async applications; `async def` handlers are awaited
res = await osr.arun("...")

osr.components.live() is the router behind route / run / learn / feedback (alive() wraps it in an AsyncRouter for aroute / arun / alearn): a plain Router, or an EnterpriseRouter as soon as a guard, middleware or hook is registered so they apply. It is built on first use and rebuilt when something is registered afterwards (learned state restarts) - declare first, route later. A rule condition that is not one of RULE_CONDITIONS (domains, actions, all_actions, language, min_complexity, max_complexity, contains_pii, needs_tools, event, workflow, context) raises ConfigurationError instead of matching silently; a @handler for an id no catalogue declares raises when the targets are materialised, naming the ids it knows.

The same declarations ride along with the CLI and the served router. osr --components my_components route "..." (repeatable; my_components:setup also calls setup(registry)) imports the module so its decorators register, then adds what they declared to the router built from --targets / --rules: targets and @handler bindings, strategies, rules, signals and policy rules - @guard, middleware and the hooks make osr serve wrap it in an EnterpriseRouter. OSR_SERVER_COMPONENTS=pkg.module,pkg.other:setup does the same for the container and the Helm chart. In code the step is osr.components.apply(router) for a Router that already exists (one from Router.from_files), plus enterprise.harden(router, middleware=osr.components.middlewares(), telemetry=osr.components.telemetry_sinks()) when middleware or hooks are declared.

• osr.components is the process-wide registry the top-level decorators bind to. Create your own ComponentRegistry() for isolation (tests, multi-tenant apps) - its methods are the same decorators, and route / run / learn / feedback / live (and their a-prefixed async forms) work on it the same way.
• Duplicate names raise ConfigurationError (register(..., replace=True) to override); unregister, clear, merge, add(*instances) (targets, strategies, rules - a RulesStrategy folds into the declared rules - extractors, middleware, telemetry, policy rules) and load(targets, rules) manage the catalogue.
• RouterBuilder.with_components(registry, targets=, strategies=, rules=, signals=, policy=, middleware=, telemetry=) wires a registry into an existing builder selectively; declared rules merge into the builder's with_rules(...) strategy when there is one.
• Plugins: registry.include("pkg.module") imports a module so its decorators run ("pkg.module:setup" calls setup(registry) afterwards); registry.discover() loads every installed distribution advertising an opensmartroute.plugins entry point.

Subclassing

from opensmartroute import Strategy, StrategyScore


class GeoAffinity(Strategy):
    name = "geo"

    def score(self, request, signals, candidates):
        region = request.context.get("region")
        return {
            t.id: StrategyScore(1.0 if region in t.constraints.regions else 0.5, f"region={region}") for t in candidates
        }


app = RouterBuilder(reg).with_defaults().with_strategy(GeoAffinity(), weight=0.8).build()

Custom Policy, SignalExtractor, Middleware, Telemetry, StateStore, AuditSink follow the same pattern — subclass, implement, pass to the builder (or register with a decorator).

Settings

Every tunable lives in one immutable Settings object, grouped by consumer (routing, policy, rules, capability, bandit, weights, slm, implicit, learning, server, observability, marketplace, compression, estimate, inference, knowledge, client). Components resolve the process-wide settings unless given an explicit instance.

import opensmartroute as osr
from opensmartroute.settings import RoutingSettings, WeightSettings

osr.configure(routing=RoutingSettings(narrow_above=200), weights=WeightSettings(rules=3.0))
router = osr.Router(reg)  # uses the configured settings
isolated = osr.Router(reg, settings=osr.Settings())  # library defaults, unaffected
print(osr.get_settings().env_keys())  # every honoured OSR_* variable

Environment overlay: OSR_<GROUP>_<FIELD> (e.g. OSR_ROUTING_SOFTMAX_TEMPERATURE=0.2, OSR_WEIGHTS_CAPABILITY=1.5, OSR_POLICY_DATA_BOUNDARIES=public,internal,confidential). Values are parsed against the default's type; a bad value raises ConfigurationError with details={"env": key}. osr.configure() with no arguments re-reads the environment. osr settings [--json] prints the effective values, their OSR_* keys and which ones the environment overrides.

Naming conventions

opensmartroute.branding is the single source of truth for every brand-derived name:

Strategy names are snake_case nouns (capability, llm_judge); decorator-registered classes drop the Strategy/Signal/Middleware/Telemetry suffix automatically (GeoAffinityStrategy -> geo_affinity).

HTTP API (osr serve)

The proxy is a drop-in base_url for the official OpenAI SDK and the MCP endpoint for the official MCP clients; the test suite runs both against a real osr serve process. Section "The OpenAI-compatible proxy" in User guide documents the streaming, tool-calling, embeddings and fallback semantics; the building blocks are public (execution.execute_with_fallback, execution.ChatStream, server.proxy_request / proxy_response / embeddings_response).

create_app(router) takes a plain Router or an EnterpriseRouter (middleware, health, telemetry and audit then apply to every request). Errors map to HTTP statuses: SecurityError / ValidationError → 400, NoRouteError → 422, TargetUnavailableError → 503, ExecutionError → 502, unknown pinned model → 404. Handlers may be async def; the endpoints run routing and execution in worker threads and drive coroutine handlers to completion.

Hardening the served router. osr serve builds a plain Router from the catalogue; the production controls are one flag (or one OSR_SERVER_* variable, for the image and the Helm chart) each and wrap it in an EnterpriseRouter through enterprise.harden():

The same wiring in code: enterprise.harden(router, guard=True, redact_pii=True, health=True, audit="audit.jsonl", origin_policy=True, limits=True, web_grounding=True, state_store="redis://..."), or on the builder RouterBuilder(...).with_guard(redact=True).with_origin_policy().with_limits(max_steps=20).with_web_grounding().

create_app(router, auth_tokens=[...]) (CLI: serve --token, --generate-token, or OSR_SERVER_AUTH_TOKENS=a,b) turns on access control: every path except /healthz, /readyz, /metrics, /whoami and the OpenAPI documents needs Authorization: Bearer <token> or X-API-Key: <token>, compared in constant time; anything else is 401 with WWW-Authenticate: Bearer. A token entry may carry scopes: <token>:ro is read-only (every GET, /route without execution and /mcp list / read calls, but 403 on /feedback, /autopilot/cycle, /marketplace/sync and on everything that runs a target - /v1/* POSTs and /mcp tools/call; OSR_SERVER_READONLY_EXECUTES=1 restores execution for read-only tokens) and :exp=<unix seconds> expires it (server.AccessToken). ServerSettings.require_auth (OSR_SERVER_REQUIRE_AUTH=1, serve --require-auth) refuses to start without a token. Tokens are opaque strings; osr token generate mints osr_local_... ones and osr login --url http://host:8000 --token ... stores one for the CLI.

CLI

osr login [--url URL] [--token T | --with-token] [--no-browser] [--profile P] [--json]
osr whoami [--json] | osr logout [--all] | osr token generate|create|list|revoke
osr trace REQUEST_ID [--json] | osr events [--name guard.*] [--level warning] [--request-id ID] [--limit N] [--json]
osr -t targets.yaml -r rules.yaml route "text" [--plan] [--json] [--cost-weight 0.3]
osr -t targets.yaml -r rules.yaml eval dataset.jsonl [--frontier] [--multi-round --threshold 0.8 --max-rounds 3]
osr -t targets.yaml eval rows.jsonl --effort              # rows with per-target scores + tokens vs always-think
osr -t targets.yaml eval tasks.jsonl --agentic [--retries 1]   # per-step task routing vs best single agent
osr -t targets.yaml targets | stats | serve [--host --port] [--token T ...] [--generate-token] [--require-auth]
osr -t targets.yaml serve --guard --redact-pii --health --audit audit.jsonl --origin-policy --limits --state-store redis://...
osr -t targets.yaml serve --marketplace-refresh 86400 [--marketplace-url URL]   # follow the marketplace in-process
osr -t targets.yaml mcp [--list-tools] [--marketplace-url URL | --no-marketplace]   # MCP server over stdio for an IDE
osr mcp --url https://host/mcp --api-key K                        # stdio bridge to a remote /mcp endpoint
osr -t targets.yaml improve --slm slm.json --cache-dir data [--feed URL ...]   # self-improvement loop
osr -t targets.yaml --slm bundled route "text"                  # the routing SLM shipped inside the package
osr slm info bundled | osr slm predict bundled "text" --explain
osr -t targets.yaml --skills <folder> route "text" --plan   # add SKILL.md packages as slot targets
osr skills [ROOT] [--json]                                       # validate + list SKILL.md packages
osr settings [--json]

python -m opensmartroute ... is the same command for environments where the osr script is not on PATH (IDE MCP server configurations, containers, uv run).

osr login signs in to the hosted platform with the OAuth 2.0 device authorization grant (the browser opens /platform/cli/authorize, you approve the code, the platform mints a workspace API key) or stores a pasted token; opensmartroute.credentials (CredentialStore, device_login, whoami, PlatformClient) is the library behind it and works with any injected transport.

Works with your stack

Every LLM provider and agent framework connects through one of four surfaces, all of which ship today. There is no per-framework plugin to wait for: if it can speak OpenAI, Anthropic, MCP or call a function, it can use the router.

Providers (the router calls them). Any OpenAI-compatible endpoint works through adapters.openai_compat.OpenAICompatClient; Anthropic is spoken natively over the Messages API by adapters.AnthropicClient (tools, images, streaming and extended thinking are translated both ways, and the platform picks the dialect from the provider's kind or an anthropic.com base URL). The hosted platform's provider presets cover OpenAI, Azure OpenAI / AI Foundry, Amazon Bedrock, Google Gemini and Vertex AI, Anthropic, Mistral, Groq, Together, Fireworks, DeepSeek, xAI, Cohere, OpenRouter, Ollama, vLLM, a LiteLLM gateway and any custom OpenAI-compatible URL - each preset is the provider's real endpoint and auth header, probed by Check before it serves traffic.

Clients and frameworks (they call the router). The served router (osr serve and the hosted /v1) answers four dialects on one metered chat path, so a framework only needs a base URL:

Deep adapters (the router inside their graph). adapters.frameworks.langgraph_node / langgraph_condition make the router a LangGraph node and conditional edge; maf_router_executor is a Microsoft Agent Framework (AutoGen's successor) executor; adapters.a2a imports A2A agent cards as targets; adapters.mcp imports other MCP servers' tools (with signed manifests) as targets; adapters.harness wraps any callable, HTTP endpoint or subprocess agent as a target. All of the above are exercised against real in-process servers in the test suite.