---
title: "Custom Callbacks"
url: "/docs/observability/custom_callback"
canonical_url: "https://docs.litellm.ai/docs/observability/custom_callback"
type: "docs"
last_updated: "2026-10-20"
related:
  - "/docs/observability/callbacks"
  - "/docs/observability/generic_api"
---
# Custom Callbacks

> Index of all LiteLLM docs: https://docs.litellm.ai/llms.txt


:::info
**For PROXY** [Go Here](../proxy/logging.md#custom-callback-class-async)
::: 

## Callback Class
You can create a custom callback class to precisely log events as they occur in litellm. 

```python
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm import completion, acompletion

class MyCustomHandler(CustomLogger):
    def log_pre_api_call(self, model, messages, kwargs): 
        print(f"Pre-API Call")
    
    def log_post_api_call(self, kwargs, response_obj, start_time, end_time): 
        print(f"Post-API Call")
    

    def log_success_event(self, kwargs, response_obj, start_time, end_time): 
        print(f"On Success")

    def log_failure_event(self, kwargs, response_obj, start_time, end_time): 
        print(f"On Failure")
    
    #### ASYNC #### - for acompletion/aembeddings

    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Success")

    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Failure")

customHandler = MyCustomHandler()

litellm.callbacks = [customHandler]

## sync 
response = completion(model="gpt-6-luna", messages=[{ "role": "user", "content": "Hi - i'm openai"}],
                              stream=True)
for chunk in response: 
    continue

## async
import asyncio 

async def completion():
    response = await acompletion(model="gpt-6-luna", messages=[{ "role": "user", "content": "Hi - i'm openai"}],
                              stream=True)
    async for chunk in response: 
        continue
asyncio.run(completion())
```

## Common Hooks

- `async_log_success_event` - Log successful API calls
- `async_log_failure_event` - Log failed API calls  
- `log_pre_api_call` - Log before API call
- `log_post_api_call` - Log after API call

**Proxy-only hooks** (only work with LiteLLM Proxy):
- `async_post_call_success_hook` - Access user data + modify responses
- `async_pre_call_hook` - Modify requests before sending

### Per-Attempt Deployment Hooks

`async_log_success_event` and `async_log_failure_event` are documented as request-level hooks: each is meant to fire once per logical client request. The per-attempt deployment hooks below are a different contract entirely: each one is guaranteed to fire once for every real deployment call, including the original attempt, every retry, and every fallback step, with no dedup logic involved. Reach for these, not the request-level hooks, whenever you need a signal per deployment attempt rather than per logical request; a Prometheus-style per-deployment counter or a circuit breaker keyed on individual attempt failures is exactly what these are for.

These work in both the SDK and the proxy:

- `async_pre_call_deployment_hook(kwargs, call_type)` - runs before each deployment call, can modify the request
- `async_post_call_success_deployment_hook(request_data, response, call_type)` - runs after each successful deployment call
- `async_post_call_failure_deployment_hook(request_data, exception, call_type, fallback_depth=None)` - runs after each failed deployment call

A fallback chain where the first two deployments fail and the third succeeds calls the failure hook twice and the success hook once, in that order. `request_data` is that attempt's own request kwargs, not shared state carried over from an earlier attempt. `fallback_depth` is a best-effort field on the failure hook: `None` on the original attempt, `1` on the first fallback hop, `2` on the second, and so on.

```python
from litellm.integrations.custom_logger import CustomLogger

class DeploymentFailureCounter(CustomLogger):
    def __init__(self):
        super().__init__()
        self.failures_by_model = {}

    async def async_post_call_failure_deployment_hook(self, request_data, exception, call_type, fallback_depth=None):
        model = request_data.get("model", "unknown")
        self.failures_by_model[model] = self.failures_by_model.get(model, 0) + 1
        print(f"deployment failure: model={model} exception={type(exception).__name__} fallback_depth={fallback_depth} total={self.failures_by_model[model]}")

counter = DeploymentFailureCounter()
litellm.callbacks = [counter]
```

### Example: Modifying the Response in async_post_call_success_hook

You can use `async_post_call_success_hook` to add custom headers or metadata to the response before it is returned to the client. For example:

```python
async def async_post_call_success_hook(data, user_api_key_dict, response):
    # Add a custom header to the response
    additional_headers = getattr(response, "_hidden_params", {}).get("additional_headers", {}) or {}
    additional_headers["x-litellm-custom-header"] = "my-value"
    if not hasattr(response, "_hidden_params"):
        response._hidden_params = {}
    response._hidden_params["additional_headers"] = additional_headers
    return response
```

This allows you to inject custom metadata or headers into the response for downstream consumers. You can use this pattern to pass information to clients, proxies, or observability tools.

## Callback Functions
If you just want to log on a specific event (e.g. on input) - you can use callback functions. 

You can set custom callbacks to trigger for:
- `litellm.input_callback`   - Track inputs/transformed inputs before making the LLM API call
- `litellm.success_callback` - Track inputs/outputs after making LLM API call
- `litellm.failure_callback` - Track inputs/outputs + exceptions for litellm calls

## Defining a Custom Callback Function
Create a custom callback function that takes specific arguments:

```python
def custom_callback(
    kwargs,                 # kwargs to completion
    completion_response,    # response from completion
    start_time, end_time    # start/end time
):
    # Your custom code here
    print("LITELLM: in custom callback function")
    print("kwargs", kwargs)
    print("completion_response", completion_response)
    print("start_time", start_time)
    print("end_time", end_time)
```

### Setting the custom callback function
```python
import litellm
litellm.success_callback = [custom_callback]
```

## Using Your Custom Callback Function

```python
import litellm
from litellm import completion

# Assign the custom callback function
litellm.success_callback = [custom_callback]

response = completion(
    model="gpt-6-luna",
    messages=[
        {
            "role": "user",
            "content": "Hi - i'm openai"
        }
    ]
)

print(response)

```

## Async Callback Functions 

We recommend using the Custom Logger class for async.

```python
from litellm.integrations.custom_logger import CustomLogger
from litellm import acompletion 

class MyCustomHandler(CustomLogger):
    #### ASYNC #### 
    

    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Success")

    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Failure")

import asyncio 
customHandler = MyCustomHandler()

litellm.callbacks = [customHandler]

async def completion():
    response = await acompletion(model="gpt-6-luna", messages=[{ "role": "user", "content": "Hi - i'm openai"}],
                              stream=True)
    async for chunk in response: 
        continue
asyncio.run(completion())
```

**Functions**

If you just want to pass in an async function for logging. 

LiteLLM currently supports just async success callback functions for async completion/embedding calls. 

```python
import asyncio, litellm 

async def async_test_logging_fn(kwargs, completion_obj, start_time, end_time):
    print(f"On Async Success!")

async def test_chat_openai():
    try:
        # litellm.set_verbose = True
        litellm.success_callback = [async_test_logging_fn]
        response = await litellm.acompletion(model="gpt-6-luna",
                              messages=[{
                                  "role": "user",
                                  "content": "Hi - i'm openai"
                              }],
                              stream=True)
        async for chunk in response: 
            continue
    except Exception as e:
        print(e)
        pytest.fail(f"An error occurred - {str(e)}")

asyncio.run(test_chat_openai())
```

## What's Available in kwargs?

The kwargs dictionary contains all the details about your API call.

:::info
For the complete logging payload specification, see the [Standard Logging Payload Spec](https://docs.litellm.ai/docs/proxy/logging_spec).
:::

```python
def custom_callback(kwargs, completion_response, start_time, end_time):
    # Access common data
    model = kwargs.get("model")
    messages = kwargs.get("messages", [])
    cost = kwargs.get("response_cost", 0)
    cache_hit = kwargs.get("cache_hit", False)
    
    # Access metadata you passed in
    metadata = kwargs.get("litellm_params", {}).get("metadata", {})
```

**Key fields in kwargs:**
- `model` - The model name
- `messages` - Input messages  
- `response_cost` - Calculated cost
- `cache_hit` - Whether response was cached
- `litellm_params.metadata` - Your custom metadata

## Practical Examples

### Track API Costs
```python
def track_cost_callback(kwargs, completion_response, start_time, end_time):
    cost = kwargs["response_cost"] # litellm calculates this for you
    print(f"Request cost: ${cost}")

litellm.success_callback = [track_cost_callback]

response = completion(model="gpt-6-luna", messages=[{"role": "user", "content": "Hello"}])
```

### Log Inputs to LLMs
```python
def get_transformed_inputs(kwargs):
    params_to_model = kwargs["additional_args"]["complete_input_dict"]
    print("params to model", params_to_model)

litellm.input_callback = [get_transformed_inputs]

response = completion(model="claude-sonnet-5", messages=[{"role": "user", "content": "Hello"}])
```

### Send to External Service
```python
import requests

def send_to_analytics(kwargs, completion_response, start_time, end_time):
    data = {
        "model": kwargs.get("model"),
        "cost": kwargs.get("response_cost", 0),
        "duration": (end_time - start_time).total_seconds()
    }
    requests.post("https://your-analytics.com/api", json=data)

litellm.success_callback = [send_to_analytics]
```

## Common Issues

### Callback Not Called
Make sure you:
1. Register callbacks correctly: `litellm.callbacks = [MyHandler()]`
2. Use the right hook names (check spelling)
3. Don't use proxy-only hooks in library mode

### Performance Issues  
- Use async hooks for I/O operations
- Don't block in callback functions
- Handle exceptions properly:

```python
class SafeHandler(CustomLogger):
    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        try:
            await external_service(response_obj)
        except Exception as e:
            print(f"Callback error: {e}")  # Log but don't break the flow
```

## Related pages

- [Callbacks](https://docs.litellm.ai/docs/observability/callbacks.md)
- [Generic API](https://docs.litellm.ai/docs/observability/generic_api.md)