[Feature] Add /public/endpoints endpoint for provider endpoint support

Add new /public/endpoints endpoint that returns which providers support each LiteLLM
endpoint (e.g., chat_completions, embeddings). The endpoint reads from a local backup
JSON file bundled with the package, caches the result in-process, and transforms the
raw provider-centric data into an endpoint-centric response format.

Changes:
- Add litellm/provider_endpoints_support_backup.json (copy of root source file)
- Add Pydantic response models (EndpointProvider, SupportedEndpoint, SupportedEndpointsResponse)
- Add /public/endpoints route with transformation and caching logic
- Add 16 comprehensive tests covering HTTP layer and transformation functions

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
yuneng-jiang
2026-02-26 18:17:37 -08:00
co-authored by Claude Haiku 4.5
parent 5a95d9beee
commit 86b2efd67a
4 changed files with 3035 additions and 6 deletions
File diff suppressed because it is too large Load Diff
@@ -1,6 +1,8 @@
import json
import os
from typing import List
import re
from importlib.resources import files
from typing import Any, Dict, List, Optional
import litellm
from fastapi import APIRouter, Depends, HTTPException
@@ -23,11 +25,95 @@ from litellm.types.proxy.public_endpoints.public_endpoints import (
AgentCreateInfo,
ProviderCreateInfo,
PublicModelHubInfo,
SupportedEndpointsResponse,
)
from litellm.types.utils import LlmProviders
router = APIRouter()
# ---------------------------------------------------------------------------
# /public/endpoints — helpers
# ---------------------------------------------------------------------------
_ENDPOINT_METADATA: Dict[str, Dict[str, str]] = {
"chat_completions": {"label": "Chat Completions", "endpoint": "/chat/completions"},
"messages": {"label": "Messages", "endpoint": "/messages"},
"responses": {"label": "Responses", "endpoint": "/responses"},
"embeddings": {"label": "Embeddings", "endpoint": "/embeddings"},
"image_generations": {"label": "Image Generations", "endpoint": "/images/generations"},
"audio_transcriptions": {"label": "Audio Transcriptions", "endpoint": "/audio/transcriptions"},
"audio_speech": {"label": "Audio Speech", "endpoint": "/audio/speech"},
"moderations": {"label": "Moderations", "endpoint": "/moderations"},
"batches": {"label": "Batches", "endpoint": "/batches"},
"rerank": {"label": "Rerank", "endpoint": "/rerank"},
"ocr": {"label": "OCR", "endpoint": "/ocr"},
"search": {"label": "Search", "endpoint": "/search"},
"skills": {"label": "Skills", "endpoint": "/skills"},
"interactions": {"label": "Interactions", "endpoint": "/interactions"},
"a2a_(Agent Gateway)": {"label": "A2A (Agent Gateway)", "endpoint": "/a2a/{agent}/message/send"},
"container": {"label": "Containers", "endpoint": "/containers"},
"container_file": {"label": "Container Files", "endpoint": "/containers/{id}/files"},
"compact": {"label": "Compact", "endpoint": "/responses/compact"},
"files": {"label": "Files", "endpoint": "/files"},
"image_edits": {"label": "Image Edits", "endpoint": "/images/edits"},
"vector_stores_create": {"label": "Vector Stores (Create)", "endpoint": "/vector_stores"},
"vector_stores_search": {"label": "Vector Stores (Search)", "endpoint": "/vector_stores/{id}/search"},
"video_generations": {"label": "Video Generations", "endpoint": "/videos/generations"},
}
_SLUG_SUFFIX_RE = re.compile(r"\s*\(`[^`]+`\)\s*$")
# Loaded once on first request; never invalidated (local file, no TTL needed).
_cached_endpoints: Optional[List[Dict[str, Any]]] = None
def _clean_display_name(raw: str) -> str:
return _SLUG_SUFFIX_RE.sub("", raw).strip()
def _build_endpoints(raw: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Transform raw provider_endpoints_support_backup.json into the response shape."""
providers: Dict[str, Any] = raw.get("providers", {})
# Collect endpoint keys in insertion order (union across all providers).
seen: set = set()
all_keys: List[str] = []
for provider_data in providers.values():
for key in provider_data.get("endpoints", {}):
if key not in seen:
seen.add(key)
all_keys.append(key)
result: List[Dict[str, Any]] = []
for key in all_keys:
meta = _ENDPOINT_METADATA.get(key)
label = meta["label"] if meta else key.replace("_", " ").title()
path = meta["endpoint"] if meta else "/" + key.replace("_", "/")
supporting: List[Dict[str, str]] = [
{
"slug": slug,
"display_name": _clean_display_name(pd.get("display_name", slug)),
}
for slug, pd in providers.items()
if pd.get("endpoints", {}).get(key)
]
result.append({"key": key, "label": label, "endpoint": path, "providers": supporting})
return result
def _load_endpoints() -> List[Dict[str, Any]]:
raw = json.loads(
files("litellm")
.joinpath("provider_endpoints_support_backup.json")
.read_text(encoding="utf-8")
)
return _build_endpoints(raw)
# ---------------------------------------------------------------------------
@router.get(
"/public/model_hub",
@@ -225,6 +311,24 @@ async def get_litellm_blog_posts():
return BlogPostsResponse(posts=posts)
@router.get(
"/public/endpoints",
tags=["public"],
response_model=SupportedEndpointsResponse,
)
async def get_supported_endpoints() -> SupportedEndpointsResponse:
"""
Return the list of LiteLLM proxy endpoints and which providers support each one.
Reads from the bundled local backup file. Result is cached in-process for
the lifetime of the server process.
"""
global _cached_endpoints
if _cached_endpoints is None:
_cached_endpoints = _load_endpoints()
return SupportedEndpointsResponse(endpoints=_cached_endpoints)
@router.get(
"/public/agents/fields",
tags=["public", "[beta] Agents"],
@@ -233,7 +337,7 @@ async def get_litellm_blog_posts():
async def get_agent_fields() -> List[AgentCreateInfo]:
"""
Return agent type metadata required by the dashboard create-agent flow.
If an agent has `inherit_credentials_from_provider`, the provider's credential
fields are automatically appended to the agent's credential_fields.
"""
@@ -242,19 +346,19 @@ async def get_agent_fields() -> List[AgentCreateInfo]:
"proxy",
"public_endpoints",
)
agent_create_fields_path = os.path.join(base_path, "agent_create_fields.json")
provider_create_fields_path = os.path.join(base_path, "provider_create_fields.json")
with open(agent_create_fields_path, "r") as f:
agent_create_fields = json.load(f)
with open(provider_create_fields_path, "r") as f:
provider_create_fields = json.load(f)
# Build a lookup map for providers by name
provider_map = {p["provider"]: p for p in provider_create_fields}
# Merge inherited credential fields
for agent in agent_create_fields:
inherit_from = agent.get("inherit_credentials_from_provider")
@@ -52,3 +52,19 @@ class AgentCreateInfo(BaseModel):
credential_fields: List[AgentCredentialField]
litellm_params_template: Optional[Dict[str, str]] = None
model_template: Optional[str] = None
class EndpointProvider(BaseModel):
slug: str
display_name: str
class SupportedEndpoint(BaseModel):
key: str
label: str
endpoint: str
providers: List[EndpointProvider]
class SupportedEndpointsResponse(BaseModel):
endpoints: List[SupportedEndpoint]
@@ -357,3 +357,164 @@ def test_public_model_hub_mixed_health_statuses():
assert claude["health_checked_at"] is None
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# /public/endpoints
# ---------------------------------------------------------------------------
import litellm.proxy.public_endpoints.public_endpoints as _pe_module
from litellm.proxy.public_endpoints.public_endpoints import _build_endpoints, _clean_display_name
@pytest.fixture(autouse=False)
def reset_endpoints_cache():
"""Reset the module-level cache before and after each cache-related test."""
original = _pe_module._cached_endpoints
_pe_module._cached_endpoints = None
yield
_pe_module._cached_endpoints = original
def _make_client():
app = FastAPI()
app.include_router(router)
return TestClient(app)
def test_get_supported_endpoints_returns_200(reset_endpoints_cache):
response = _make_client().get("/public/endpoints")
assert response.status_code == 200
def test_get_supported_endpoints_response_shape(reset_endpoints_cache):
data = _make_client().get("/public/endpoints").json()
assert "endpoints" in data
assert isinstance(data["endpoints"], list)
assert len(data["endpoints"]) > 0
def test_get_supported_endpoints_item_fields(reset_endpoints_cache):
endpoints = _make_client().get("/public/endpoints").json()["endpoints"]
for item in endpoints:
assert "key" in item
assert "label" in item
assert "endpoint" in item
assert "providers" in item
assert isinstance(item["providers"], list)
def test_get_supported_endpoints_provider_fields(reset_endpoints_cache):
endpoints = _make_client().get("/public/endpoints").json()["endpoints"]
for item in endpoints:
for provider in item["providers"]:
assert "slug" in provider
assert "display_name" in provider
def test_get_supported_endpoints_paths_start_with_slash(reset_endpoints_cache):
endpoints = _make_client().get("/public/endpoints").json()["endpoints"]
for item in endpoints:
assert item["endpoint"].startswith("/"), f"Expected path starting with /, got: {item['endpoint']}"
def test_get_supported_endpoints_chat_completions_present(reset_endpoints_cache):
endpoints = _make_client().get("/public/endpoints").json()["endpoints"]
keys = [item["key"] for item in endpoints]
assert "chat_completions" in keys
chat = next(item for item in endpoints if item["key"] == "chat_completions")
assert chat["endpoint"] == "/chat/completions"
assert chat["label"] == "Chat Completions"
assert len(chat["providers"]) > 0
def test_get_supported_endpoints_display_names_have_no_slug_suffix(reset_endpoints_cache):
"""Provider display_names must not contain the raw `` (`slug`) `` suffix."""
import re
suffix_re = re.compile(r"\(`[^`]+`\)")
endpoints = _make_client().get("/public/endpoints").json()["endpoints"]
for item in endpoints:
for provider in item["providers"]:
assert not suffix_re.search(provider["display_name"]), (
f"display_name still contains slug suffix: {provider['display_name']!r}"
)
def test_get_supported_endpoints_is_cached(reset_endpoints_cache):
"""`_load_endpoints` is called only once; subsequent requests use the cache."""
client = _make_client()
with patch(
"litellm.proxy.public_endpoints.public_endpoints._load_endpoints",
wraps=_pe_module._load_endpoints,
) as mock_load:
client.get("/public/endpoints")
client.get("/public/endpoints")
client.get("/public/endpoints")
mock_load.assert_called_once()
# ---------------------------------------------------------------------------
# _build_endpoints unit tests (transformation logic)
# ---------------------------------------------------------------------------
_MINIMAL_RAW = {
"providers": {
"openai": {
"display_name": "OpenAI (`openai`)",
"url": "https://example.com",
"endpoints": {"chat_completions": True, "embeddings": True, "images": False},
},
"anthropic": {
"display_name": "Anthropic (`anthropic`)",
"url": "https://example.com",
"endpoints": {"chat_completions": True, "embeddings": False, "images": False},
},
}
}
def test_build_endpoints_known_key_uses_metadata():
result = _build_endpoints(_MINIMAL_RAW)
chat = next(e for e in result if e["key"] == "chat_completions")
assert chat["label"] == "Chat Completions"
assert chat["endpoint"] == "/chat/completions"
def test_build_endpoints_only_includes_supporting_providers():
result = _build_endpoints(_MINIMAL_RAW)
embeddings = next(e for e in result if e["key"] == "embeddings")
slugs = [p["slug"] for p in embeddings["providers"]]
assert slugs == ["openai"]
def test_build_endpoints_unknown_key_derives_label_and_path():
raw = {
"providers": {
"someprovider": {
"display_name": "Some Provider (`someprovider`)",
"endpoints": {"my_custom_endpoint": True},
}
}
}
result = _build_endpoints(raw)
item = result[0]
assert item["key"] == "my_custom_endpoint"
assert item["label"] == "My Custom Endpoint"
assert item["endpoint"].startswith("/")
def test_build_endpoints_empty_providers_returns_empty():
result = _build_endpoints({"providers": {}})
assert result == []
def test_clean_display_name_strips_suffix():
assert _clean_display_name("OpenAI (`openai`)") == "OpenAI"
assert _clean_display_name("AI/ML API (`aiml`)") == "AI/ML API"
assert _clean_display_name("A2A (Agent-to-Agent) (`a2a`)") == "A2A (Agent-to-Agent)"
def test_clean_display_name_passthrough_when_no_suffix():
assert _clean_display_name("OpenAI") == "OpenAI"
assert _clean_display_name("") == ""