feat: generate provider_endpoints_support.json from the cost map (#48619)

* feat: generate provider endpoint support flags

* test: remove derived endpoint flag assertions

* test: run provider endpoint generator in CI

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-10-27 12:06:55 -07:00 • committed by GitHub
parent a2c23d95ae
commit c39b5a3efd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 6705 additions and 1895 deletions

View file

@ -524,6 +524,12 @@ jobs:
- name: check_provider_folders_documented
run: uv run --no-sync python ./tests/code_coverage_tests/check_provider_folders_documented.py
- name: check_provider_endpoints_support_generated
run: uv run --no-sync python scripts/generate_provider_endpoints_support.py --check
- name: test_generate_provider_endpoints_support
run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_generate_provider_endpoints_support.py
- name: check_prisma_binary_cache
run: uv run --no-sync python ./tests/code_coverage_tests/check_prisma_binary_cache.py

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,248 @@
from __future__ import annotations
import argparse
import json
import sys
from collections.abc import Iterator, Mapping, Sequence
from itertools import chain
from pathlib import Path
from types import MappingProxyType
from typing import Final
from pydantic import JsonValue, TypeAdapter
ROOT: Final = Path(__file__).resolve().parents[1]
COST_MAP_PATH: Final = ROOT / "model_prices_and_context_window.json"
SOURCE_PATH: Final = ROOT / "provider_endpoints_support.json"
OUTPUT_PATHS: Final = (
SOURCE_PATH,
ROOT / "litellm/provider_endpoints_support_backup.json",
)
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
MODE_TO_FLAG: Final[Mapping[str, str]] = MappingProxyType(
{
"chat": "chat_completions",
"responses": "responses",
"completion": "text_completion",
"embedding": "embeddings",
"image_generation": "image_generations",
"image_edit": "image_edits",
"audio_transcription": "audio_transcriptions",
"audio_speech": "audio_speech",
"rerank": "rerank",
"ocr": "ocr",
"search": "search",
"moderation": "moderations",
"realtime": "realtime",
"video_generation": "video_generations",
}
)
SUPPORTED_ENDPOINT_TO_FLAG: Final[Mapping[str, str]] = MappingProxyType(
{
"/v1/chat/completions": "chat_completions",
"/v1/responses": "responses",
"/v1/completions": "text_completion",
"/v1/embeddings": "embeddings",
"/v1/images/generations": "image_generations",
"/v1/images/edits": "image_edits",
"/v1/audio/transcriptions": "audio_transcriptions",
"/v1/audio/speech": "audio_speech",
"/v1/ocr": "ocr",
"/v1/realtime": "realtime",
"/v1/videos": "video_generations",
"/v1/batch": "batches",
"/v1/messages": "messages",
"/v1beta/interactions": "interactions",
"/v1/systemone": "systemone",
}
)
def _json_object(value: JsonValue, name: str) -> Mapping[str, JsonValue]:
if not isinstance(value, dict):
raise ValueError(f"{name} must be a JSON object")
return value
def _provider_slug(provider: str) -> str:
if provider.startswith("vertex_ai-"):
return "vertex_ai"
if provider == "fireworks_ai-embedding-models":
return "fireworks_ai"
if provider == "bedrock_converse":
return "bedrock"
if provider == "text-completion-inception":
return "inception"
return provider
def _model_provider_slug(model: JsonValue) -> str | None:
if not isinstance(model, dict):
return None
provider: Final = model.get("litellm_provider")
if not isinstance(provider, str) or not provider:
return None
return _provider_slug(provider)
def _model_flags(model: Mapping[str, JsonValue]) -> tuple[str, ...]:
mode: Final = model.get("mode")
mode_flag: Final = MODE_TO_FLAG.get(mode) if isinstance(mode, str) else None
supported_endpoints: Final = model.get("supported_endpoints")
endpoint_flags: Final = (
tuple(
SUPPORTED_ENDPOINT_TO_FLAG[endpoint]
for endpoint in supported_endpoints
if isinstance(endpoint, str) and endpoint in SUPPORTED_ENDPOINT_TO_FLAG
)
if isinstance(supported_endpoints, list)
else ()
)
return ((mode_flag,) if mode_flag is not None else ()) + endpoint_flags
def _bridge_field(endpoint: JsonValue) -> str | None:
if not isinstance(endpoint, dict) or endpoint.get("bridges_to_chat_completion") is not True:
return None
provider_json_field: Final = endpoint.get("provider_json_field")
return provider_json_field if isinstance(provider_json_field, str) else None
def _bridged_fields(endpoints: Mapping[str, JsonValue]) -> tuple[str, ...]:
return tuple(
dict.fromkeys(
provider_json_field
for endpoint in endpoints.values()
if (provider_json_field := _bridge_field(endpoint)) is not None
)
)
def _derived_keys(bridged_fields: tuple[str, ...]) -> tuple[str, ...]:
return tuple(dict.fromkeys((*MODE_TO_FLAG.values(), *SUPPORTED_ENDPOINT_TO_FLAG.values(), *bridged_fields)))
def _model_evidence(
model: JsonValue, bridged_fields: tuple[str, ...]
) -> Iterator[tuple[str, str]]:
provider: Final = _model_provider_slug(model)
if provider is None or not isinstance(model, dict):
return
flags: Final = _model_flags(model)
yield from ((provider, flag) for flag in flags)
if "chat_completions" not in flags:
return
yield from ((provider, flag) for flag in bridged_fields)
def _provider_evidence(
cost_map: Mapping[str, JsonValue], bridged_fields: tuple[str, ...]
) -> Iterator[tuple[str, str]]:
return chain.from_iterable(
_model_evidence(model, bridged_fields)
for model_name, model in cost_map.items()
if model_name != "sample_spec"
)
def _provider_slugs(cost_map: Mapping[str, JsonValue]) -> frozenset[str]:
return frozenset(
provider_slug
for model_name, model in cost_map.items()
if model_name != "sample_spec" and (provider_slug := _model_provider_slug(model)) is not None
)
def _generated_provider(
slug: str,
value: JsonValue,
derived_keys: tuple[str, ...],
derived_key_set: frozenset[str],
evidence: frozenset[tuple[str, str]],
) -> dict[str, JsonValue]:
provider: Final = _json_object(value, f"provider {slug}")
current_flags_value: Final = provider.get("endpoints")
current_flags: Final = (
_json_object(current_flags_value, f"provider {slug} endpoints")
if current_flags_value is not None
else {}
)
derived_flags: Final = {flag: (slug, flag) in evidence for flag in derived_keys}
handwritten_flags: Final = {
flag: value for flag, value in current_flags.items() if flag not in derived_key_set
}
return {
**provider,
"endpoints": {**derived_flags, **handwritten_flags},
}
def generate_provider_endpoints_support(
cost_map: Mapping[str, JsonValue],
current_json: Mapping[str, JsonValue],
) -> dict[str, JsonValue]:
current_providers: Final = _json_object(
current_json.get("providers"), "provider_endpoints_support providers"
)
endpoints: Final = _json_object(current_json.get("endpoints"), "provider_endpoints_support endpoints")
bridged_fields: Final = _bridged_fields(endpoints)
derived_keys: Final = _derived_keys(bridged_fields)
derived_key_set: Final = frozenset(derived_keys)
provider_slugs: Final = _provider_slugs(cost_map)
missing_slugs: Final = tuple(sorted(provider_slugs - frozenset(current_providers)))
if missing_slugs:
raise ValueError(f"Cost map providers missing from provider_endpoints_support.json: {', '.join(missing_slugs)}")
evidence: Final = frozenset(_provider_evidence(cost_map, bridged_fields))
providers: Final = {
slug: _generated_provider(slug, value, derived_keys, derived_key_set, evidence)
for slug, value in current_providers.items()
}
return {**current_json, "providers": providers}
def generate_files(
cost_map: Mapping[str, JsonValue],
current_json: Mapping[str, JsonValue],
output_paths: tuple[Path, ...],
check: bool,
) -> int:
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
content: Final = json.dumps(generated, indent=2, ensure_ascii=False) + "\n"
stale_paths: Final = tuple(
path for path in output_paths if not path.exists() or path.read_text(encoding="utf-8") != content
)
if check:
if not stale_paths:
return 0
sys.stderr.write(
"provider endpoint support is stale in "
+ ", ".join(str(path) for path in stale_paths)
+ "\nRun `uv run --no-sync python scripts/generate_provider_endpoints_support.py` to regenerate it\n"
)
return 1
for path in output_paths:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return 0
def main(argv: Sequence[str] | None = None, *, repo_root: Path = ROOT) -> int:
parser: Final = argparse.ArgumentParser(description="Generate provider endpoint support from the model cost map")
parser.add_argument("--check", action="store_true", help="fail if either generated file is stale")
args: Final = parser.parse_args(argv)
cost_map: Final = JSON_OBJECT.validate_json((repo_root / COST_MAP_PATH.name).read_bytes())
current_json: Final = JSON_OBJECT.validate_json((repo_root / SOURCE_PATH.name).read_bytes())
return generate_files(
cost_map,
current_json,
(
repo_root / SOURCE_PATH.name,
repo_root / OUTPUT_PATHS[1].relative_to(ROOT),
),
args.check,
)
if __name__ == "__main__":
sys.exit(main())

View file

@ -0,0 +1,179 @@
import json
from collections.abc import Mapping
from pathlib import Path
from typing import Final
import pytest
from pydantic import JsonValue
from scripts.generate_provider_endpoints_support import (
generate_provider_endpoints_support,
main,
)
ENDPOINTS: Final = {
"a2a": {"provider_json_field": "a2a", "bridges_to_chat_completion": True},
"messages": {"provider_json_field": "messages", "bridges_to_chat_completion": True},
"responses": {"provider_json_field": "responses", "bridges_to_chat_completion": True},
"generateContent": {"provider_json_field": "generateContent", "bridges_to_chat_completion": True},
"text_completion": {"provider_json_field": "text_completion", "bridges_to_chat_completion": True},
"count_tokens": {"provider_json_field": "count_tokens"},
}
def _current_json(providers: Mapping[str, JsonValue]) -> dict[str, JsonValue]:
return {
"_comment": "retained",
"_schema": {"version": 1},
"providers": dict(providers),
"endpoints": ENDPOINTS,
}
def _provider(flags: Mapping[str, JsonValue] | None = None) -> dict[str, JsonValue]:
return {
"display_name": "Example (`example`)",
"url": "https://docs.litellm.ai/docs/providers/example",
"endpoints": dict(flags or {}),
}
def _flags(generated: Mapping[str, JsonValue], slug: str) -> Mapping[str, JsonValue]:
providers: Final = generated["providers"]
assert isinstance(providers, dict)
provider: Final = providers[slug]
assert isinstance(provider, dict)
flags: Final = provider["endpoints"]
assert isinstance(flags, dict)
return flags
def test_mode_derives_chat_flag() -> None:
cost_map: Final = {"example/model": {"litellm_provider": "example", "mode": "chat"}}
current_json: Final = _current_json({"example": _provider()})
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
assert _flags(generated, "example")["chat_completions"] is True
def test_supported_endpoint_derives_embedding_flag() -> None:
cost_map: Final = {
"example/model": {
"litellm_provider": "example",
"mode": "completion",
"supported_endpoints": ["/v1/embeddings"],
}
}
current_json: Final = _current_json({"example": _provider()})
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
assert _flags(generated, "example")["embeddings"] is True
def test_chat_models_enable_bridge_flags_but_other_models_do_not() -> None:
cost_map: Final = {
"example/chat": {"litellm_provider": "has_chat", "mode": "chat"},
"example/endpoint-chat": {
"litellm_provider": "endpoint_chat",
"mode": "image_generation",
"supported_endpoints": ["/v1/chat/completions"],
},
"example/image": {"litellm_provider": "no_chat", "mode": "image_generation"},
}
current_json: Final = _current_json(
{
"has_chat": _provider(),
"endpoint_chat": _provider(),
"no_chat": _provider(),
}
)
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
assert _flags(generated, "has_chat")["a2a"] is True
assert _flags(generated, "has_chat")["generateContent"] is True
assert _flags(generated, "endpoint_chat")["a2a"] is True
assert _flags(generated, "no_chat")["a2a"] is False
assert _flags(generated, "no_chat")["generateContent"] is False
def test_unbacked_derived_flags_reset_and_handwritten_flags_remain() -> None:
cost_map: Final = {"example/model": {"litellm_provider": "example", "mode": "image_generation"}}
current_json: Final = _current_json(
{
"example": _provider({"responses": True, "count_tokens": True, "files": False}),
"without_models": _provider({"responses": True}),
}
)
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
flags: Final = _flags(generated, "example")
assert flags["responses"] is False
assert flags["count_tokens"] is True
assert flags["files"] is False
assert _flags(generated, "without_models")["chat_completions"] is False
def test_provider_aliases_share_their_mapped_slugs() -> None:
cost_map: Final = {
"vertex/model-a": {"litellm_provider": "vertex_ai-chat", "mode": "chat"},
"vertex/model-b": {"litellm_provider": "vertex_ai-embedding", "mode": "embedding"},
"bedrock/model": {"litellm_provider": "bedrock_converse", "mode": "responses"},
}
current_json: Final = _current_json({"vertex_ai": _provider(), "bedrock": _provider()})
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
assert _flags(generated, "vertex_ai")["chat_completions"] is True
assert _flags(generated, "vertex_ai")["embeddings"] is True
assert _flags(generated, "bedrock")["responses"] is True
def test_cost_map_provider_without_file_entry_is_an_error() -> None:
cost_map: Final = {"example/model": {"litellm_provider": "missing_provider", "mode": "chat"}}
current_json: Final = _current_json({"example": _provider()})
with pytest.raises(ValueError, match="missing_provider"):
generate_provider_endpoints_support(cost_map, current_json)
def test_check_reports_stale_output_without_writing(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
cost_map: Final = {"example/model": {"litellm_provider": "example", "mode": "chat"}}
current_json: Final = _current_json({"example": _provider()})
(tmp_path / "model_prices_and_context_window.json").write_text(json.dumps(cost_map), encoding="utf-8")
source_path: Final = tmp_path / "provider_endpoints_support.json"
source_content: Final = json.dumps(current_json)
source_path.write_text(source_content, encoding="utf-8")
result: Final = main(["--check"], repo_root=tmp_path)
assert result == 1
assert source_path.read_text(encoding="utf-8") == source_content
assert not (tmp_path / "litellm/provider_endpoints_support_backup.json").exists()
error_output: Final = capsys.readouterr().err
assert "provider endpoint support is stale" in error_output
assert "scripts/generate_provider_endpoints_support.py" in error_output
def test_check_reports_stale_backup_without_writing(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
cost_map: Final = {"example/model": {"litellm_provider": "example", "mode": "chat"}}
current_json: Final = _current_json({"example": _provider()})
generated: Final = generate_provider_endpoints_support(cost_map, current_json)
generated_content: Final = json.dumps(generated, indent=2, ensure_ascii=False) + "\n"
source_path: Final = tmp_path / "provider_endpoints_support.json"
backup_path: Final = tmp_path / "litellm/provider_endpoints_support_backup.json"
backup_content: Final = "stale\n"
(tmp_path / "model_prices_and_context_window.json").write_text(json.dumps(cost_map), encoding="utf-8")
source_path.write_text(generated_content, encoding="utf-8")
backup_path.parent.mkdir()
backup_path.write_text(backup_content, encoding="utf-8")
result: Final = main(["--check"], repo_root=tmp_path)
assert result == 1
assert source_path.read_text(encoding="utf-8") == generated_content
assert backup_path.read_text(encoding="utf-8") == backup_content
assert str(backup_path) in capsys.readouterr().err

View file

@ -440,14 +440,3 @@ class TestDashboardRegistration:
entry = json.loads(matrix_path.read_text())["providers"]["edenai"]
assert entry["url"] == "https://docs.litellm.ai/docs/providers/edenai"
served = {name for name, flag in entry["endpoints"].items() if flag}
assert served == {
"chat_completions",
"messages",
"responses",
"embeddings",
"image_generations",
"audio_transcriptions",
"audio_speech",
"video_generations",
}

View file

@ -58,26 +58,6 @@ def test_cortecs_is_available_in_add_model_form():
}
def test_cortecs_supported_endpoints():
matrix_path = Path(litellm.__file__).parent / "provider_endpoints_support_backup.json"
providers = json.loads(matrix_path.read_text())["providers"]
assert providers["cortecs"]["endpoints"] == {
"chat_completions": True,
"messages": True,
"responses": True,
"embeddings": False,
"image_generations": False,
"audio_transcriptions": False,
"audio_speech": False,
"moderations": False,
"batches": False,
"rerank": False,
"a2a": False,
"interactions": False,
}
def test_cortecs_chat_completion_request():
with respx.mock() as upstream:
route: Final = upstream.post("https://api.cortecs.ai/v1/chat/completions").respond(

View file

@ -91,10 +91,3 @@ class TestLibertAIProviderConfig:
matrix = json.loads(backup_path.read_text())
assert "libertai" in matrix["providers"]
endpoints = matrix["providers"]["libertai"]["endpoints"]
assert endpoints["chat_completions"] is True
# embeddings is advertised false: the JSON-configured-provider path only
# wires chat routing (the OpenAILike embedding handler is reached solely
# for the literal openai_like/llamafile/lm_studio providers), matching
# the llamagate precedent. bge-m3 stays in the cost map for metadata.
assert endpoints["embeddings"] is False

View file

@ -99,25 +99,6 @@ def test_prism_is_available_in_add_model_form():
}
def test_prism_supported_endpoints():
matrix_path = Path(litellm.__file__).parent / "provider_endpoints_support_backup.json"
providers = json.loads(matrix_path.read_text())["providers"]
assert providers["prism"]["endpoints"] == {
"chat_completions": True,
"messages": True,
"responses": True,
"embeddings": False,
"image_generations": False,
"audio_transcriptions": False,
"audio_speech": False,
"moderations": False,
"batches": False,
"rerank": False,
"a2a": False,
}
def test_prism_responses_request():
with respx.mock() as upstream:
route: Final = upstream.post("https://api.prisminference.com/v1/responses").respond(

View file

@ -115,28 +115,6 @@ def test_reka_is_available_in_add_model_form():
}
def test_reka_supported_endpoints():
expected: Final = {
"chat_completions": True,
"messages": True,
"responses": True,
"embeddings": False,
"image_generations": False,
"audio_transcriptions": False,
"audio_speech": False,
"moderations": False,
"batches": False,
"rerank": False,
"a2a": False,
"interactions": False,
}
backup_path = Path(litellm.__file__).parent / "provider_endpoints_support_backup.json"
root_path = Path(litellm.__file__).parent.parent / "provider_endpoints_support.json"
assert json.loads(backup_path.read_text())["providers"]["reka"]["endpoints"] == expected
assert json.loads(root_path.read_text())["providers"]["reka"]["endpoints"] == expected
def test_reka_chat_completion_request():
with respx.mock() as upstream:
route: Final = upstream.post("https://api.reka.ai/v1/chat/completions").respond(200, json=_CHAT_COMPLETION)