ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Feat] Add List Callbacks API Endpoint (#11987)

Browse Source 
* add get_callbacks_by_type

* add list_callbacks

* fix _get_callback_string

* add callback_management_endpoints_router

* fix proxy config.yaml

* fixes list callbacks

* TestCallbackManagementEndpoints

* update docs

* docs Response Fields

* docs header format

* docs Dynamic Callback Management
This commit is contained in:
Ishaan Jaff 2025-06-23 15:34:25 -07:00 committed by GitHub
parent eacb4dfdef
commit b5c48c8c22
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 389 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
docs/my-website/docs/proxy/dynamic_logging.md
View File

@ -7,7 +7,7 @@ import TabItem from '@theme/TabItem';
:::info
This is an enterprise feature.
✨ This is an enterprise feature.
[Get started with LiteLLM Enterprise](https://www.litellm.ai/enterprise)
@ -20,74 +20,61 @@ LiteLLM's dynamic callback management enables teams to control logging behavior
You can disable callbacks by passing the `x-litellm-disable-callbacks` header with your requests, giving teams granular control over where their data is logged.
## Quick Start
## Getting Started: List and Disable Callbacks
<Tabs>
<TabItem value="disable-single" label="Disable a single callback">
Managing callbacks is a two-step process:
1. **First, list your active callbacks** to see what's currently enabled
2. **Then, disable specific callbacks** as needed for your requests
## 1. List Active Callbacks
Start by viewing all currently enabled callbacks on your proxy to see what's available to disable.
#### Request
```bash
curl --location 'http://0.0.0.0:4000/chat/completions' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer sk-1234' \
--header 'x-litellm-disable-callbacks: langfuse' \
--data '{
"model": "claude-sonnet-4-20250514",
"messages": [
{
"role": "user",
"content": "what llm are you"
}
]
}'
curl -X 'GET' \
'http://localhost:4000/callbacks/list' \
-H 'accept: application/json' \
-H 'x-litellm-api-key: sk-1234'
```
</TabItem>
<TabItem value="disable-multiple" label="Disable multiple callbacks">
```bash
curl --location 'http://0.0.0.0:4000/chat/completions' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer sk-1234' \
--header 'x-litellm-disable-callbacks: langfuse,datadog' \
--data '{
"model": "claude-sonnet-4-20250514",
"messages": [
{
"role": "user",
"content": "what llm are you"
}
]
}'
```
</TabItem>
</Tabs>
## 1. View Active Logging Callbacks
Before disabling callbacks, you can view all currently enabled callbacks on your proxy.
### Request
```bash
curl --location 'http://0.0.0.0:4000/callbacks/list' \
--header 'Authorization: Bearer sk-1234'
```
### Response
#### Response
```json
{
"callbacks": [
"langfuse",
"datadog",
"prometheus",
"slack_alerting"
]
"success": [
"deployment_callback_on_success",
"sync_deployment_callback_on_success"
],
"failure": [
"async_deployment_callback_on_failure",
"deployment_callback_on_failure"
],
"success_and_failure": [
"langfuse",
"datadog"
]
}
```
## 2. Disable a Single Callback
#### Response Fields
The response contains three arrays that categorize your active callbacks:
- **`success`** - Callbacks that only execute when requests complete successfully. These callbacks receive data from successful LLM responses.
- **`failure`** - Callbacks that only execute when requests fail or encounter errors. These callbacks receive error information and failed request data.
- **`success_and_failure`** - Callbacks that execute for both successful and failed requests. These are typically logging/observability tools that need to capture all request data regardless of outcome.
---
## 2. Disable Callbacks
Now that you know which callbacks are active, you can selectively disable them using the `x-litellm-disable-callbacks` header. You can reference any callback name from the list response above.
### Disable a Single Callback
Use the `x-litellm-disable-callbacks` header to disable specific callbacks for individual requests.
@ -140,9 +127,9 @@ print(response)
</TabItem>
</Tabs>
## 3. Disable Multiple Callbacks
### Disable Multiple Callbacks
You can disable multiple callbacks by providing a comma-separated list in the header.
You can disable multiple callbacks by providing a comma-separated list in the header. Use any combination of callback names from your `/callbacks/list` response.
<Tabs>
<TabItem value="Curl" label="Curl Request">
@ -192,3 +179,36 @@ print(response)
</TabItem>
</Tabs>
## Header Format and Case Sensitivity
### Expected Header Format
The `x-litellm-disable-callbacks` header accepts callback names in the following formats (use the exact names returned by `/callbacks/list`):
- **Single callback**: `x-litellm-disable-callbacks: langfuse`
- **Multiple callbacks**: `x-litellm-disable-callbacks: langfuse,datadog,prometheus`
When specifying multiple callbacks, use comma-separated values without spaces around the commas.
### Case Sensitivity
**Callback name checks are case insensitive.** This means all of the following are equivalent:
```bash
# These are all equivalent
x-litellm-disable-callbacks: langfuse
x-litellm-disable-callbacks: LANGFUSE
x-litellm-disable-callbacks: LangFuse
x-litellm-disable-callbacks: langFUSE
```
This applies to both single and multiple callback specifications:
```bash
# Case insensitive for multiple callbacks
x-litellm-disable-callbacks: LANGFUSE,datadog,PROMETHEUS
x-litellm-disable-callbacks: langfuse,DATADOG,prometheus
```

63
litellm/litellm_core_utils/logging_callback_manager.py
View File

@ -4,6 +4,7 @@ import litellm
from litellm._logging import verbose_logger
from litellm.integrations.additional_logging_utils import AdditionalLoggingUtils
from litellm.integrations.custom_logger import CustomLogger
from litellm.types.utils import CallbacksByType
class LoggingCallbackManager:
@ -275,3 +276,65 @@ class LoggingCallbackManager:
isinstance(callback, callback_type)
for callback in self._get_all_callbacks()
)
def get_callbacks_by_type(self) -> CallbacksByType:
"""
Get all active callbacks categorized by their type (success, failure, success_and_failure).
Returns:
CallbacksByType: Dict with keys 'success', 'failure', 'success_and_failure' containing lists of callback strings
"""
# Get callback lists
success_callbacks = set(litellm.success_callback + litellm._async_success_callback)
failure_callbacks = set(litellm.failure_callback + litellm._async_failure_callback)
general_callbacks = set(litellm.callbacks)
# Get all unique callbacks
all_callbacks = success_callbacks | failure_callbacks | general_callbacks
result: CallbacksByType = CallbacksByType(
success=[],
failure=[],
success_and_failure=[]
)
for callback in all_callbacks:
callback_str = self._get_callback_string(callback)
is_in_success = callback in success_callbacks
is_in_failure = callback in failure_callbacks
is_in_general = callback in general_callbacks
if is_in_general or (is_in_success and is_in_failure):
result["success_and_failure"].append(callback_str)
elif is_in_success:
result["success"].append(callback_str)
elif is_in_failure:
result["failure"].append(callback_str)
# final de-duplication
result["success"] = list(set(result["success"]))
result["failure"] = list(set(result["failure"]))
result["success_and_failure"] = list(set(result["success_and_failure"]))
return result
def _get_callback_string(
self,
callback: Union[CustomLogger, Callable, str]
) -> str:
from litellm.litellm_core_utils.custom_logger_registry import (
CustomLoggerRegistry,
)
"""Convert a callback to its string representation"""
if isinstance(callback, str):
return callback
elif isinstance(callback, CustomLogger):
# Try to get the string representation from the registry
callback_str = CustomLoggerRegistry.get_callback_str_from_class_type(type(callback))
return callback_str if callback_str is not None else type(callback).__name__
elif callable(callback):
return getattr(callback, '__name__', str(callback))
return str(callback)

8
litellm/model_prices_and_context_window_backup.json
View File

@ -13648,10 +13648,10 @@
"output_cost_per_token": 8e-06,
"output_cost_per_reasoning_token": 3e-06,
"citation_cost_per_token": 2e-06,
"search_queries_cost_per_query": {
"search_queries_size_low": 0.005,
"search_queries_size_medium": 0.005,
"search_queries_size_high": 0.005
"search_context_cost_per_query": {
"search_context_size_low": 0.005,
"search_context_size_medium": 0.005,
"search_context_size_high": 0.005
},
"litellm_provider": "perplexity",
"mode": "chat",

27
litellm/proxy/management_endpoints/callback_management_endpoints.py Normal file
View File

@ -0,0 +1,27 @@
"""
Endpoints for managing callbacks
"""
from fastapi import APIRouter, Depends
from litellm.litellm_core_utils.logging_callback_manager import CallbacksByType
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
router = APIRouter()
@router.get(
"/callbacks/list",
tags=["Logging Callbacks"],
dependencies=[Depends(user_api_key_auth)],
response_model=CallbacksByType,
)
async def list_callbacks():
"""
View List of Active Logging Callbacks
"""
from litellm import logging_callback_manager
# Get callbacks organized by type using the callback manager utility
callbacks_by_type = logging_callback_manager.get_callbacks_by_type()
return callbacks_by_type

5
litellm/proxy/proxy_config.yaml
View File

@ -15,4 +15,7 @@ mcp_servers:
general_settings:
store_model_in_db: true
store_prompts_in_spend_logs: true
store_prompts_in_spend_logs: true
litellm_settings:
callbacks: ["langfuse", "datadog"]

4
litellm/proxy/proxy_server.py
View File

@ -231,6 +231,9 @@ from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
from litellm.proxy.management_endpoints.budget_management_endpoints import (
router as budget_management_router,
)
from litellm.proxy.management_endpoints.callback_management_endpoints import (
router as callback_management_endpoints_router,
)
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
from litellm.proxy.management_endpoints.customer_endpoints import (
router as customer_router,
@ -8595,6 +8598,7 @@ app.include_router(spend_management_router)
app.include_router(caching_router)
app.include_router(analytics_router)
app.include_router(guardrails_router)
app.include_router(callback_management_endpoints_router)
app.include_router(debugging_endpoints_router)
app.include_router(ui_crud_endpoints_router)
app.include_router(openai_files_router)

5
litellm/types/utils.py
View File

@ -2489,3 +2489,8 @@ class DynamicPromptManagementParamLiteral(str, Enum):
@classmethod
def list_all_params(cls):
return [param.value for param in cls]
class CallbacksByType(TypedDict):
success: List[str]
failure: List[str]
success_and_failure: List[str]

202
tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py Normal file
View File

@ -0,0 +1,202 @@
import json
import os
import sys
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi.testclient import TestClient
sys.path.insert(
0, os.path.abspath("../../../..")
) #
from typing import cast
import litellm
from litellm.integrations.datadog.datadog import DataDogLogger
from litellm.integrations.langfuse.langfuse import LangFuseLogger
from litellm.proxy.management_endpoints.callback_management_endpoints import router
from litellm.proxy.proxy_server import app
class TestCallbackManagementEndpoints:
"""Test suite for callback management endpoints"""
@pytest.fixture(autouse=True)
def setup_and_teardown(self):
"""Setup and teardown for each test"""
# Reset callbacks before each test
litellm.success_callback = []
litellm.failure_callback = []
litellm._async_success_callback = []
litellm._async_failure_callback = []
litellm.callbacks = []
yield
# Clean up after each test
litellm.success_callback = []
litellm.failure_callback = []
litellm._async_success_callback = []
litellm._async_failure_callback = []
litellm.callbacks = []
def test_list_callbacks_no_active_callbacks(self):
"""Test /callbacks/list endpoint with no active callbacks"""
# Setup test client
client = TestClient(app)
# Make request to list callbacks endpoint
response = client.get(
"/callbacks/list",
headers={"Authorization": "Bearer sk-1234"}
)
# Verify response
assert response.status_code == 200
response_data = response.json()
assert "success" in response_data
assert "failure" in response_data
assert "success_and_failure" in response_data
# All lists should be empty
assert response_data["success"] == []
assert response_data["failure"] == []
assert response_data["success_and_failure"] == []
@patch.dict(os.environ, {
"LANGFUSE_PUBLIC_KEY": "test_public_key",
"LANGFUSE_SECRET_KEY": "test_secret_key",
"LANGFUSE_HOST": "https://test.langfuse.com"
})
def test_list_callbacks_with_langfuse_logger(self):
"""Test /callbacks/list endpoint with real Langfuse logger initialized"""
# Setup test client
client = TestClient(app)
# Initialize Langfuse logger and add to callbacks
with patch('litellm.integrations.langfuse.langfuse.Langfuse') as mock_langfuse:
# Mock the Langfuse client initialization
mock_langfuse_client = MagicMock()
mock_langfuse.return_value = mock_langfuse_client
# Add string representation to callback lists (this is how the system typically works)
litellm.success_callback.append("langfuse")
litellm._async_success_callback.append("langfuse")
# Make request to list callbacks endpoint
response = client.get(
"/callbacks/list",
headers={"Authorization": "Bearer sk-1234"}
)
# Verify response
assert response.status_code == 200
response_data = response.json()
# Verify langfuse appears in success callbacks
assert "langfuse" in response_data["success"]
assert response_data["failure"] == []
assert response_data["success_and_failure"] == []
# Verify the response structure is correct
assert isinstance(response_data["success"], list)
assert isinstance(response_data["failure"], list)
assert isinstance(response_data["success_and_failure"], list)
def test_list_callbacks_with_datadog_logger(self):
"""Test /callbacks/list endpoint with DataDog logger configuration"""
# Setup test client
client = TestClient(app)
# Test with datadog callbacks added directly (without initializing the logger to avoid async issues)
# Add string representations to different callback types to test comprehensive categorization
litellm.success_callback.append("datadog")
litellm.failure_callback.append("datadog")
litellm.callbacks.append("datadog")
# Make request to list callbacks endpoint
response = client.get(
"/callbacks/list",
headers={"Authorization": "Bearer sk-1234"}
)
# Verify response
assert response.status_code == 200
response_data = response.json()
# Verify datadog appears in the correct categorization
# Since datadog is in both success and failure, it should appear in success_and_failure
assert "datadog" in response_data["success_and_failure"]
# The categorization logic should deduplicate properly
assert len([cb for cb in response_data["success"] if cb == "datadog"]) <= 1
assert len([cb for cb in response_data["failure"] if cb == "datadog"]) <= 1
assert len([cb for cb in response_data["success_and_failure"] if cb == "datadog"]) <= 1
# Verify the response structure is correct
assert isinstance(response_data["success"], list)
assert isinstance(response_data["failure"], list)
assert isinstance(response_data["success_and_failure"], list)
def test_list_callbacks_mixed_callback_types(self):
"""Test /callbacks/list endpoint with mixed callback types (string and logger instances)"""
# Setup test client
client = TestClient(app)
# Setup mixed callbacks
litellm.success_callback.append("langfuse")
litellm.failure_callback.append("datadog")
litellm.callbacks.append("prometheus")
# Make request to list callbacks endpoint
response = client.get(
"/callbacks/list",
headers={"Authorization": "Bearer sk-1234"}
)
# Verify response
assert response.status_code == 200
response_data = response.json()
# Verify callbacks are properly categorized
assert "prometheus" in response_data["success_and_failure"] # callbacks list items go to success_and_failure
assert "langfuse" in response_data["success"]
assert "datadog" in response_data["failure"]
# Verify no duplicates
all_callbacks = (
response_data["success"] +
response_data["failure"] +
response_data["success_and_failure"]
)
assert len(set(all_callbacks)) == len(all_callbacks)
def test_list_callbacks_empty_response_structure(self):
"""Test that response always has correct structure even with no callbacks"""
# Setup test client
client = TestClient(app)
# Make request to list callbacks endpoint
response = client.get(
"/callbacks/list",
headers={"Authorization": "Bearer sk-1234"}
)
# Verify response structure
assert response.status_code == 200
response_data = response.json()
# Verify all required keys are present
required_keys = ["success", "failure", "success_and_failure"]
for key in required_keys:
assert key in response_data
assert isinstance(response_data[key], list)