Function create_app

uncommon_route/proxy.py:2735–5979

Create the ASGI application wired to the given upstream base URL.

(
    upstream: str | None = DEFAULT_UPSTREAM,
    spend_control: SpendControl | None = None,
    providers_config: ProvidersConfig | None = None,
    route_stats: RouteStats | None = None,
    trace_store: TraceStore | None = None,
    feedback: FeedbackCollector | None = None,
    model_mapper: ModelMapper | None = None,
    artifact_store: ArtifactStore | None = None,
    composition_policy: CompositionPolicy | None = None,
    semantic_compressor: SemanticCompressor | None = None,
    model_experience: ModelExperienceStore | None = None,
    routing_config_store: RoutingConfigStore | None = None,
    connections_store: ConnectionsStore | None = None,
    route_confidence_calibrator: object | None = None,
)

def create_app(
    upstream: str | None = DEFAULT_UPSTREAM,
    spend_control: SpendControl | None = None,
    providers_config: ProvidersConfig | None = None,
    route_stats: RouteStats | None = None,
    trace_store: TraceStore | None = None,
    feedback: FeedbackCollector | None = None,
    model_mapper: ModelMapper | None = None,
    artifact_store: ArtifactStore | None = None,
    composition_policy: CompositionPolicy | None = None,
    semantic_compressor: SemanticCompressor | None = None,
    model_experience: ModelExperienceStore | None = None,
    routing_config_store: RoutingConfigStore | None = None,
    connections_store: ConnectionsStore | None = None,
    route_confidence_calibrator: object | None = None,
) -> Starlette:
    """Create the ASGI application wired to the given upstream base URL."""
    _cli_upstream_override = str(upstream or "").strip() or None
    _connections_store = connections_store or ConnectionsStore()
    _effective_connection = resolve_primary_connection(
        cli_upstream=_cli_upstream_override,
        store=_connections_store,
    )
    upstream = _effective_connection.upstream
    _primary_api_key = _effective_connection.api_key
    _spend = spend_control or SpendControl()
    _spend_reservation = _SpendReservation(_spend)
    _providers = providers_config or load_providers()
    _stats = route_stats or RouteStats()
    _traces = trace_store or TraceStore()
    _route_confidence = route_confidence_calibrator or get_active_route_confidence_calibrator()
    if any(record.feedback_signal for record in _stats.history()):
        _route_confidence.fit_from_route_records(_stats.history())
    _model_experience = model_experience or ModelExperienceStore()
    _feedback = feedback or FeedbackCollector(
        model_experience=_model_experience,
        buffer_path=data_dir() / "feedback_buffer.json",
    )
    if getattr(_feedback, "_model_experience", None) is None:
        _feedback._model_experience = _model_experience
    _mapper = model_mapper or ModelMapper(upstream)
    _artifacts = artifact_store or ArtifactStore()
    _composition_policy = composition_policy or load_composition_policy()
    _semantic = semantic_compressor
    _routing_store = routing_config_store or RoutingConfigStore()
    _routing_config = _routing_store.config()
    _scene_store = SceneStore()
    _responses_history: dict[str, list[dict[str, Any]]] = {}
    forced_messages_mode_raw = str(os.environ.get("UNCOMMON_ROUTE_FORCE_MESSAGES_DEFAULT_MODE", "")).strip()
    forced_messages_upstream_model = str(
        os.environ.get("UNCOMMON_ROUTE_FORCE_MESSAGES_UPSTREAM_MODEL", "")
    ).strip()
    _forced_messages_default_mode: RoutingMode | None = None
    disable_anthropic_cache = str(os.environ.get("UNCOMMON_ROUTE_DISABLE_ANTHROPIC_CACHE", "")).strip().lower() in {
        "1",
        "true",
        "yes",
        "on",