"""Render only credential-ready routes; credentials remain environment references.""" import os import json from copy import deepcopy from pathlib import Path import yaml from provider_policy import without_subscription, apply_provider_policy def oauth_ready(path): try: record = json.loads(path.read_text()) return isinstance(record, dict) and bool(record.get('access_token')) except (OSError, ValueError): return False def namespace_config(config): """Add explicit provider names without removing existing client aliases.""" routes = config['model_list'] existing = {r['model_name']: r for r in routes} fallbacks = config['router_settings']['fallbacks'] automatic = {name for group in fallbacks for name in group} renamed = {} def clone(route, alias): if alias in existing: if existing[alias]['litellm_params'] != route['litellm_params']: raise ValueError('Conflicting namespace alias: ' + alias) return new = deepcopy(route) new['model_name'] = alias existing[alias] = new routes.append(new) for route in list(routes): alias = route['model_name'] model = route['litellm_params']['model'] if alias in {'gemini-3.6-flash', 'gemini-3.5-flash-lite', 'google/gemini-embedding-2', 'google/gemini-3.8-flash'}: continue if alias.endswith('-openrouter'): new = 'openrouter/' + alias.removesuffix('-openrouter') elif alias in automatic: new = 'auto/' + alias clone(route, 'zenmux/' + alias) elif model.startswith('anthropic/'): new = ('zenmux-payg/' + alias.removesuffix('-payg') if alias.endswith('-payg') else 'zenmux/' + alias) elif model.startswith('openai/'): name = model.removeprefix('openai/').split('/', 1)[-1] new = 'zenmux/' + (name if alias.startswith('asr-') else alias.replace('-zenmux', '-latest')) elif model.startswith('openrouter/'): new = 'openrouter/' + alias elif model.startswith('gemini/'): new = 'aistudio/' + alias.removesuffix('-studio') elif model.startswith('vertex_ai/'): name = {'google-rerank-default': 'rerank-default-latest', 'google-rerank-fast': 'rerank-fast-latest'}.get(alias, alias.replace('-vertex', '-latest')) new = 'vertex/' + name elif model.startswith('chatgpt/'): name = model.split('/', 1)[1] if alias in {'gpt-work-sol', 'gpt-review-5.5'} else alias new = 'codex/' + name elif model.startswith('xai/'): new = 'xai/' + alias else: raise ValueError('Unrecognized namespace provider: ' + alias) clone(route, new) renamed[alias] = new for group in list(fallbacks): for source, targets in group.items(): fallbacks.append({renamed[source]: [renamed[target] for target in targets]}) return config def add_bailian_routes(config, env, vocabulary_path): if env.get('BAILIAN_CN_API_KEY'): base = env.get('BAILIAN_CN_API_BASE') if not base: raise ValueError('BAILIAN_CN_API_BASE required for Beijing Bailian') for personal in [False, True]: params = {'model': 'dashscope/qwen-audio-3.1-asr-flash', 'api_base': base, 'api_key': 'os.environ/BAILIAN_CN_API_KEY', 'max_retries': 0} if personal: params['bailian_vocabulary_path'] = vocabulary_path config['model_list'].append({'model_name': 'bailian-cn/qwen-audio-3.1-asr-flash' + ('-personal' if personal else ''), 'litellm_params': params}) return config def add_openai_routes(config, catalog, env): """Official API is separate from OAuth and third-party OpenAI-compatible APIs.""" if not env.get('OPENAI_API_KEY'): return config models = dict(catalog.get('openai', {})) for alias, model in catalog.get('chatgpt', {}).items(): name = alias if alias.endswith('-latest') else model if name in models and models[name] != model: raise ValueError('Official API/OAuth model mismatch: ' + name) models[name] = model by_model = {} for alias, model in models.items(): name = 'openai/' + alias config['model_list'].append({'model_name': name, 'litellm_params': { 'model': 'openai/' + model, 'api_key': 'os.environ/OPENAI_API_KEY', 'api_base': 'https://api.openai.com/v1', 'max_retries': 0}}) by_model[model] = name for route in list(config['model_list']): params = route['litellm_params'] if not params['model'].startswith('chatgpt/'): continue backup = by_model.get(params['model'].removeprefix('chatgpt/')) if not backup: raise ValueError('Missing same-model official API fallback') params['max_retries'] = 0 route['model_info'] = {'cooldown_time': 60, 'allowed_fails_policy': { 'RateLimitErrorAllowedFails': 0, 'AuthenticationErrorAllowedFails': 0, 'InternalServerErrorAllowedFails': 0, 'ServiceUnavailableErrorAllowedFails': 0, 'TimeoutErrorAllowedFails': 0}} config['router_settings']['fallbacks'].append({route['model_name']: [backup]}) return config def build_config(catalog, env, exists, namespaced=True): env = without_subscription(catalog, env) routes = [] fallbacks = [] def add(alias, model, key=None, **params): if key: params['api_key'] = 'os.environ/' + key routes.append({'model_name': alias, 'litellm_params': {'model': model, **params}}) for key, suffix in [('ZENMUX_SUBSCRIPTION_API_KEY', ''), ('ZENMUX_PAYG_API_KEY', '-payg')]: if env.get(key): for alias, model in catalog['zenmux'].items(): # Strip only the LiteLLM provider prefix, retaining ZenMux's vendor prefix. add(alias + suffix, 'anthropic/' + model, key, api_base='https://zenmux.ai/api/anthropic') if env.get('ZENMUX_SUBSCRIPTION_API_KEY'): for alias, model in catalog.get('zenmux_chat', {}).items(): add(alias, 'openai/' + model, 'ZENMUX_SUBSCRIPTION_API_KEY', api_base='https://zenmux.ai/api/v1') if env.get('OPENROUTER_API_KEY'): for alias, model in catalog['openrouter'].items(): primary = catalog.get('zenmux_open_models', {}).get(alias) if primary and env.get('ZENMUX_SUBSCRIPTION_API_KEY'): if primary != model: raise ValueError('Cross-provider model mismatch: ' + alias) add(alias, 'openai/' + primary, 'ZENMUX_SUBSCRIPTION_API_KEY', api_base='https://zenmux.ai/api/v1', max_retries=0) routes[-1]['model_info'] = {'cooldown_time': 60, 'allowed_fails_policy': {'RateLimitErrorAllowedFails': 0, 'AuthenticationErrorAllowedFails': 0, 'InternalServerErrorAllowedFails': 0, 'ServiceUnavailableErrorAllowedFails': 0, 'TimeoutErrorAllowedFails': 0}} add(alias + '-openrouter', 'openrouter/' + model, 'OPENROUTER_API_KEY') fallbacks.append({alias: [alias + '-openrouter']}) else: add(alias, 'openrouter/' + model, 'OPENROUTER_API_KEY') elif env.get('ZENMUX_SUBSCRIPTION_API_KEY'): for alias, model in catalog.get('zenmux_open_models', {}).items(): add(alias, 'openai/' + model, 'ZENMUX_SUBSCRIPTION_API_KEY', api_base='https://zenmux.ai/api/v1', max_retries=0) if env.get('GOOGLE_AI_STUDIO_API_KEY'): add('gemini-transcribe-personal', 'gemini/gemini-3.5-transcribe', 'GOOGLE_AI_STUDIO_API_KEY', personal_vocabulary=True) for alias, model in catalog['google_ai_studio'].items(): add(alias, 'gemini/' + model, 'GOOGLE_AI_STUDIO_API_KEY') if env.get('VERTEXAI_ENABLED', '').lower() == 'true' and env.get('VERTEXAI_PROJECT'): for alias, model in catalog.get('google_vertex', {}).items(): add(alias, 'vertex_ai/' + model, vertex_project='os.environ/VERTEXAI_PROJECT', vertex_location='os.environ/VERTEXAI_LOCATION') for provider, token_var, default_dir in [ ('chatgpt', 'CHATGPT_TOKEN_DIR', '/tokens/chatgpt'), ('xai', 'XAI_OAUTH_TOKEN_DIR', '/tokens/xai'), ]: if exists(Path(env.get(token_var, default_dir)) / 'auth.json'): for alias, model in catalog[provider].items(): extra = {'use_xai_oauth': True} if provider == 'xai' else {} add(alias, provider + '/' + model, **extra) config = { 'model_list': routes, 'general_settings': { 'master_key': 'os.environ/LITELLM_MASTER_KEY', 'database_url': 'os.environ/DATABASE_URL', 'store_model_in_db': False, 'disable_env_credential_login': True, }, 'litellm_settings': {'drop_params': False, 'set_verbose': False}, # Official OpenAI fallbacks are appended after namespace expansion. 'router_settings': {'num_retries': 1, 'timeout': 180, 'fallbacks': fallbacks}, } if namespaced: config = add_bailian_routes(namespace_config(config), env, '/etc/litellm/personal-vocabulary.json') config = add_openai_routes(config, catalog, env) return apply_provider_policy(config, catalog, env) if namespaced else config if __name__ == '__main__': catalog = yaml.safe_load(Path('/etc/litellm/models.yaml').read_text()) config = build_config(catalog, os.environ, oauth_ready) output = Path('/tmp/litellm.yaml') output.write_text(yaml.safe_dump(config, allow_unicode=True, sort_keys=False)) output.chmod(0o600) print('Configured routes:', ', '.join(route['model_name'] for route in config['model_list']))