tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-06 20:25:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #21464 from BerriAI/litellm_sanitise_anthropic_mesages_2

Browse Source 
Litellm sanitise anthropic mesages 2
This commit is contained in:
Sameer Kankute
2026-02-18 18:39:11 +05:30
committed by GitHub
parent a01dcc7155 827444cc2e
commit ee3ad3aec4
5 changed files with 1085 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/my-website/docs/completion/message_sanitization.md
+465
View File
@@ -0,0 +1,465 @@
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
# Message Sanitization for Tool Calling for anthropic models
**Automatically fix common message formatting issues when using tool calling with `modify_params=True`**
LiteLLM can automatically sanitize messages to handle common issues that occur during tool calling workflows, especially when using OpenAI-compatible clients with providers that have strict message format requirements (like Anthropic Claude).
## Overview
When `litellm.modify_params = True` is enabled, LiteLLM automatically sanitizes messages to fix three common issues:
1. **Orphaned Tool Calls** - Assistant messages with tool_calls but missing tool results
2. **Orphaned Tool Results** - Tool messages that reference non-existent tool_call_ids
3. **Empty Message Content** - Messages with empty or whitespace-only text content
This ensures your tool calling workflows work seamlessly across different LLM providers without manual message validation.
## Why Message Sanitization?
Different LLM providers have varying requirements for message formats, especially during tool calling:
- **Anthropic Claude** requires every tool_call to have a corresponding tool result
- Some providers reject messages with empty content
- OpenAI-compatible clients may not always maintain perfect message consistency
Without sanitization, these issues cause API errors that interrupt your workflows. With `modify_params=True`, LiteLLM handles these edge cases automatically.
## Quick Start
<Tabs>
<TabItem value="sdk" label="SDK">
```python
import litellm
# Enable automatic message sanitization
litellm.modify_params = True
# This will work even if messages have formatting issues
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=[
{"role": "user", "content": "What's the weather in Boston?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_123",
"type": "function",
"function": {"name": "get_weather", "arguments": '{"city": "Boston"}'}
}
]
# Missing tool result - LiteLLM will add a dummy result automatically
},
{"role": "user", "content": "Thanks!"}
],
tools=[{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"]
}
}
}]
)
```
</TabItem>
<TabItem value="proxy" label="PROXY">
```yaml
litellm_settings:
modify_params: true # Enable automatic message sanitization
model_list:
- model_name: claude-3-5-sonnet
litellm_params:
model: anthropic/claude-3-5-sonnet-20241022
```
</TabItem>
</Tabs>
## Sanitization Cases
### Case A: Orphaned Tool Calls (Missing Tool Results)
**Problem:** An assistant message contains `tool_calls`, but no corresponding tool result messages follow.
**Solution:** LiteLLM automatically adds dummy tool result messages for any missing tool results.
**Example:**
```python
import litellm
litellm.modify_params = True
# Messages with orphaned tool calls
messages = [
{"role": "user", "content": "Search for Python tutorials"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {"name": "web_search", "arguments": '{"query": "Python tutorials"}'}
}
]
},
# Missing tool result here!
{"role": "user", "content": "What about JavaScript?"}
]
# LiteLLM automatically adds:
# {
# "role": "tool",
# "tool_call_id": "call_abc123",
# "content": "[System: Tool execution skipped/interrupted by user. No result provided for tool 'web_search'.]"
# }
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=messages,
tools=[...]
)
```
**When this happens:**
- User interrupts tool execution
- Client loses tool results due to network issues
- Conversation flow changes before tool completes
- Multi-turn conversations where tools are optional
### Case B: Orphaned Tool Results (Invalid tool_call_id)
**Problem:** A tool message references a `tool_call_id` that doesn't exist in any previous assistant message.
**Solution:** LiteLLM automatically removes these orphaned tool result messages.
**Example:**
```python
import litellm
litellm.modify_params = True
# Messages with orphaned tool result
messages = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi! How can I help?"},
{
"role": "tool",
"tool_call_id": "call_nonexistent", # This tool_call_id doesn't exist!
"content": "Some result"
}
]
# LiteLLM automatically removes the orphaned tool message
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=messages
)
```
**When this happens:**
- Message history is manually edited
- Tool results are duplicated or mismatched
- Conversation state is restored incorrectly
- Messages are merged from different conversations
### Case C: Empty Message Content
**Problem:** User or assistant messages have empty or whitespace-only content.
**Solution:** LiteLLM replaces empty content with a system placeholder message.
**Example:**
```python
import litellm
litellm.modify_params = True
# Messages with empty content
messages = [
{"role": "user", "content": ""}, # Empty content
{"role": "assistant", "content": " "}, # Whitespace only
]
# LiteLLM automatically replaces with:
# {"role": "user", "content": "[System: Empty message content sanitised to satisfy protocol]"}
# {"role": "assistant", "content": "[System: Empty message content sanitised to satisfy protocol]"}
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=messages
)
```
**When this happens:**
- UI sends empty messages
- Content is stripped during preprocessing
- Placeholder messages in conversation history
- Edge cases in message construction
## Configuration
### Enable Globally
<Tabs>
<TabItem value="sdk" label="SDK">
```python
import litellm
# Enable for all completion calls
litellm.modify_params = True
```
</TabItem>
<TabItem value="proxy" label="PROXY">
```yaml
litellm_settings:
modify_params: true
```
</TabItem>
<TabItem value="env" label="Environment Variable">
```bash
export LITELLM_MODIFY_PARAMS=True
```
</TabItem>
</Tabs>
### Enable Per-Request
```python
import litellm
# Enable only for specific requests
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=messages,
modify_params=True # Override global setting
)
```
## Supported Providers
Message sanitization currently works with:
- ✅ Anthropic (Claude)
**Note:** While the sanitization logic is provider-agnostic, it is currently only applied in the Anthropic message transformation pipeline. Support for additional providers may be added in future releases.
## Implementation Details
### How It Works
The message sanitization process runs **before** messages are converted to provider-specific formats:
1. **Input:** OpenAI-format messages with potential issues
2. **Sanitization:** Three helper functions process the messages:
- `_sanitize_empty_text_content()` - Fixes empty content
- `_add_missing_tool_results()` - Adds dummy tool results
- `_is_orphaned_tool_result()` - Identifies orphaned results
3. **Output:** Clean, provider-compatible messages
### Code Reference
The sanitization logic is implemented in:
- `litellm/litellm_core_utils/prompt_templates/factory.py`
- Function: `sanitize_messages_for_tool_calling()`
### Logging
When sanitization occurs, LiteLLM logs debug messages:
```python
import litellm
litellm.set_verbose = True # Enable debug logging
# You'll see logs like:
# "_add_missing_tool_results: Found 1 orphaned tool calls. Adding dummy tool results."
# "_is_orphaned_tool_result: Found orphaned tool result with tool_call_id=call_123"
# "_sanitize_empty_text_content: Replaced empty text content in user message"
```
## Best Practices
### 1. Enable for Production Workflows
```python
# Recommended for production
litellm.modify_params = True
# Ensures robust handling of edge cases
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=messages,
tools=tools
)
```
### 2. Preserve Tool Results When Possible
While sanitization handles missing tool results, it's better to provide actual results:
```python
# Good: Provide actual tool results
messages = [
{"role": "user", "content": "Search for Python"},
{"role": "assistant", "tool_calls": [...]},
{"role": "tool", "tool_call_id": "call_123", "content": "Actual search results"}
]
# Fallback: Sanitization adds dummy result if missing
messages = [
{"role": "user", "content": "Search for Python"},
{"role": "assistant", "tool_calls": [...]},
# Missing tool result - sanitization adds dummy
]
```
### 3. Monitor Sanitization Events
Use logging to track when sanitization occurs:
```python
import litellm
import logging
# Enable debug logging
litellm.set_verbose = True
logging.basicConfig(level=logging.DEBUG)
# Track sanitization events in your application
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=messages
)
```
### 4. Test Edge Cases
Ensure your application handles sanitized messages correctly:
```python
import litellm
litellm.modify_params = True
# Test orphaned tool calls
test_messages = [
{"role": "user", "content": "Test"},
{"role": "assistant", "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "test", "arguments": "{}"}}]},
{"role": "user", "content": "Continue"} # No tool result
]
response = litellm.completion(
model="anthropic/claude-3-5-sonnet-20241022",
messages=test_messages,
tools=[...]
)
# Verify the response handles the dummy tool result appropriately
```
## Related Features
- **[Drop Params](./drop_params.md)** - Drop unsupported parameters for specific providers
- **[Message Trimming](./message_trimming.md)** - Trim messages to fit token limits
- **[Function Calling](./function_call.md)** - Complete guide to tool/function calling
- **[Reasoning Content](../reasoning_content.md)** - Extended thinking with tool calling
## Troubleshooting
### Sanitization Not Working
**Issue:** Messages still cause errors despite `modify_params=True`
**Solution:**
1. Verify `modify_params` is enabled:
```python
import litellm
print(litellm.modify_params) # Should be True
```
2. Check if the issue is provider-specific:
```python
litellm.set_verbose = True # Enable debug logging
```
3. Ensure you're using a recent version of LiteLLM:
```bash
pip install --upgrade litellm
```
### Unexpected Dummy Tool Results
**Issue:** Dummy tool results appear when you expect actual results
**Cause:** Tool result messages are missing or have incorrect `tool_call_id`
**Solution:**
1. Verify tool result messages have correct `tool_call_id`:
```python
# Correct
{"role": "tool", "tool_call_id": "call_123", "content": "result"}
# Incorrect - will be treated as orphaned
{"role": "tool", "tool_call_id": "wrong_id", "content": "result"}
```
2. Ensure tool results immediately follow assistant messages with tool_calls
### Performance Impact
**Issue:** Concerned about performance overhead
**Details:** Message sanitization has minimal performance impact:
- Runs in O(n) time where n = number of messages
- Only processes messages when `modify_params=True`
- Typically adds < 1ms to request processing time
## FAQ
**Q: Does sanitization modify my original messages?**
A: No, sanitization creates a new list of messages. Your original messages remain unchanged.
**Q: Can I disable specific sanitization cases?**
A: Currently, all three cases are handled together when `modify_params=True`. To disable sanitization entirely, set `modify_params=False`.
**Q: What happens to the dummy tool results?**
A: Dummy tool results are sent to the LLM provider along with other messages. The model sees them as regular tool results with informative error messages.
**Q: Does this work with streaming?**
A: Yes, message sanitization works with both streaming and non-streaming requests.
**Q: Is this related to `drop_params`?**
A: No, they're separate features:
- `modify_params` - Modifies/fixes message content and structure
- `drop_params` - Removes unsupported API parameters
Both can be enabled simultaneously.
## See Also
- [Reasoning Content with Tool Calling](../reasoning_content.md)
- [Function Calling Guide](./function_call.md)
- [Bedrock Provider Documentation](../providers/bedrock.md)
- [Anthropic Provider Documentation](../providers/anthropic.md)
docs/my-website/sidebars.js
+1
View File
@@ -944,6 +944,7 @@ const sidebars = {
"providers/anthropic_tool_search",
"guides/code_interpreter",
"completion/message_trimming",
"completion/message_sanitization",
"completion/model_alias",
"completion/mock_requests",
"completion/predict_outputs",
litellm/litellm_core_utils/prompt_templates/factory.py
+232
View File
@@ -2018,6 +2018,235 @@ def anthropic_process_openai_file_message(
)
def _sanitize_empty_text_content(
message: AllMessageValues,
) -> AllMessageValues:
"""
Case C: Sanitize empty text content
- Replace empty or whitespace-only text content with a placeholder message.
Returns:
The message with sanitized content if needed, otherwise the original message
"""
if message.get("role") in ["user", "assistant"]:
content = message.get("content")
if isinstance(content, str):
if not content or not content.strip():
message = cast(AllMessageValues, dict(message)) # Make a copy
message["content"] = "[System: Empty message content sanitised to satisfy protocol]"
verbose_logger.debug(
f"_sanitize_empty_text_content: Replaced empty text content in {message.get('role')} message"
)
return message
def _add_missing_tool_results( # noqa: PLR0915
current_message: AllMessageValues,
messages: List[AllMessageValues],
current_index: int,
) -> Tuple[List[AllMessageValues], int]:
"""
Case A: Missing tool_result for tool_use (orphaned tool calls)
- If an assistant message has tool_calls but no corresponding tool result follows,
add a dummy tool result message indicating the user did not provide the result.
Returns:
A tuple of:
- List containing the assistant message, followed by existing tool results,
followed by any dummy tool results needed
- Number of original messages consumed (to adjust iteration index)
"""
result_messages: List[AllMessageValues] = []
tool_calls = current_message.get("tool_calls")
if not tool_calls or len(cast(list, tool_calls)) == 0:
return ([current_message], 0)
# Collect all tool_call_ids from this assistant message
expected_tool_call_ids = set()
for tool_call in cast(list, tool_calls):
tool_call_id = None
if isinstance(tool_call, dict):
tool_call_id = tool_call.get("id")
else:
tool_call_id = getattr(tool_call, "id", None)
if tool_call_id:
expected_tool_call_ids.add(tool_call_id)
# Collect actual tool result messages that follow this assistant message
found_tool_call_ids = set()
actual_tool_results: List[AllMessageValues] = []
j = current_index + 1
while j < len(messages):
next_msg = messages[j]
next_role = next_msg.get("role")
if next_role == "assistant":
break
if next_role in ["tool", "function"]:
tool_call_id = next_msg.get("tool_call_id")
if tool_call_id and tool_call_id in expected_tool_call_ids:
found_tool_call_ids.add(tool_call_id)
actual_tool_results.append(next_msg)
j += 1
# Find missing tool results
missing_tool_call_ids = expected_tool_call_ids - found_tool_call_ids
if missing_tool_call_ids:
verbose_logger.debug(
f"_add_missing_tool_results: Found {len(missing_tool_call_ids)} orphaned tool calls. Adding dummy tool results."
)
result_messages.append(current_message)
# Add existing tool results FIRST
result_messages.extend(actual_tool_results)
# Then add dummy tool results for missing ones
for tool_call_id in missing_tool_call_ids:
tool_name = "unknown_tool"
for tool_call in cast(list, tool_calls):
tc_id = None
if isinstance(tool_call, dict):
tc_id = tool_call.get("id")
else:
tc_id = getattr(tool_call, "id", None)
if tc_id == tool_call_id:
if isinstance(tool_call, dict):
function = tool_call.get("function", {})
if isinstance(function, dict):
tool_name = function.get("name", "unknown_tool")
else:
tool_name = getattr(function, "name", "unknown_tool")
else:
function = getattr(tool_call, "function", None)
if function:
tool_name = getattr(function, "name", "unknown_tool")
break
dummy_tool_result: ChatCompletionToolMessage = {
"role": "tool",
"tool_call_id": tool_call_id,
"content": f"[System: Tool execution skipped/interrupted by user. No result provided for tool '{tool_name}'.]",
}
result_messages.append(dummy_tool_result)
# Return the messages and the number of original messages to skip
return (result_messages, len(actual_tool_results))
return ([current_message], 0)
def _is_orphaned_tool_result(
current_message: AllMessageValues,
sanitized_messages: List[AllMessageValues],
) -> bool:
"""
Case B: Orphaned tool_result (unexpected result)
- Check if a tool message references a tool_call_id that doesn't exist in the previous
assistant message.
Returns:
True if this is an orphaned tool result that should be removed, False otherwise
"""
if current_message.get("role") not in ["tool", "function"]:
return False
tool_call_id = current_message.get("tool_call_id")
if not tool_call_id:
return False
# Look back to find the most recent assistant message with tool_calls
found_matching_tool_call = False
for j in range(len(sanitized_messages) - 1, -1, -1):
prev_msg = sanitized_messages[j]
if prev_msg.get("role") == "assistant":
tool_calls = prev_msg.get("tool_calls")
if tool_calls:
for tool_call in cast(list, tool_calls):
tc_id = None
if isinstance(tool_call, dict):
tc_id = tool_call.get("id")
else:
tc_id = getattr(tool_call, "id", None)
if tc_id == tool_call_id:
found_matching_tool_call = True
break
break
if not found_matching_tool_call:
verbose_logger.debug(
"_is_orphaned_tool_result: Found orphaned tool result with redacted tool_call_id"
)
return True
return False
def sanitize_messages_for_tool_calling(
messages: List[AllMessageValues],
) -> List[AllMessageValues]:
"""
Sanitize messages for tool calling to handle common issues when modify_params=True:
Case A: Missing tool_result for tool_use (orphaned tool calls)
- If an assistant message has tool_calls but no corresponding tool result follows,
add a dummy tool result message indicating the user did not provide the result.
Case B: Orphaned tool_result (unexpected result)
- If a tool message references a tool_call_id that doesn't exist in the previous
assistant message, remove that tool message.
Case C: Empty text content
- Replace empty or whitespace-only text content with a placeholder message.
This function operates on OpenAI format messages before they are converted to
provider-specific formats.
"""
if not litellm.modify_params:
return messages
sanitized_messages: List[AllMessageValues] = []
i = 0
while i < len(messages):
current_message = messages[i]
# Case C: Sanitize empty text content
current_message = _sanitize_empty_text_content(current_message)
# Case A: Check if assistant message has tool_calls without following tool results
if current_message.get("role") == "assistant":
result_messages, messages_consumed = _add_missing_tool_results(current_message, messages, i)
# If dummy tool results were added, extend sanitized_messages and skip consumed messages
if len(result_messages) > 1:
sanitized_messages.extend(result_messages)
# Skip the assistant message and any actual tool results that were included
i += 1 + messages_consumed
continue
# Case B: Check for orphaned tool results
if _is_orphaned_tool_result(current_message, sanitized_messages):
i += 1
continue # Skip this orphaned tool result
# Add the message to sanitized list
sanitized_messages.append(current_message)
i += 1
return sanitized_messages
def anthropic_messages_pt( # noqa: PLR0915
messages: List[AllMessageValues],
model: str,
@@ -2037,6 +2266,9 @@ def anthropic_messages_pt( # noqa: PLR0915
5. System messages are a separate param to the Messages API
6. Ensure we only accept role, content. (message.name is not supported)
"""
# Sanitize messages for tool calling issues when modify_params=True
messages = sanitize_messages_for_tool_calling(messages)
# add role=tool support to allow function call result/error submission
user_message_types = {"user", "tool", "function"}
# reformat messages to ensure user/assistant are alternating, if there's either 2 consecutive 'user' messages or 2 consecutive 'assistant' message, merge them.
litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py
+7 -5
View File
@@ -31,11 +31,15 @@ from litellm import Router
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.utils import GuardrailTracingDetail, ModelResponseStream
from litellm.types.utils import (
GenericGuardrailAPIInputs,
GuardrailStatus,
GuardrailTracingDetail,
ModelResponseStream,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailStatus
from litellm.types.guardrails import (
BlockedWord,
@@ -1546,8 +1550,6 @@ class ContentFilterGuardrail(CustomGuardrail):
Raises:
HTTPException: If sensitive content is detected and action is BLOCK
"""
from litellm.types.utils import GuardrailStatus
start_time = datetime.now()
detections: List[ContentFilterDetection] = []
masked_entity_count: Dict[str, int] = {}
@@ -1693,4 +1695,4 @@ class ContentFilterGuardrail(CustomGuardrail):
LitellmContentFilterGuardrailConfigModel,
)
return LitellmContentFilterGuardrailConfigModel
return LitellmContentFilterGuardrailConfigModel
tests/test_litellm/llms/anthropic/test_message_sanitization.py
+380
View File
@@ -0,0 +1,380 @@
"""
Test message sanitization for Anthropic API when modify_params=True
Tests three cases:
A. Missing tool_result for tool_use (orphaned tool calls)
B. Orphaned tool_result without matching tool_use
C. Empty text content
"""
import pytest
import sys
import os
# Add the parent directory to the path so we can import litellm
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../..")))
import litellm
from litellm.litellm_core_utils.prompt_templates.factory import (
sanitize_messages_for_tool_calling,
anthropic_messages_pt,
)
class TestMessageSanitization:
"""Test message sanitization for tool calling scenarios"""
def setup_method(self):
"""Setup for each test"""
# Save original modify_params value
self.original_modify_params = litellm.modify_params
litellm.modify_params = True
def teardown_method(self):
"""Cleanup after each test"""
# Restore original modify_params value
litellm.modify_params = self.original_modify_params
def test_case_a_orphaned_tool_call_single(self):
"""
Test Case A: Assistant message with tool_calls but no tool result
Should add a dummy tool result message
"""
messages = [
{
"role": "user",
"content": "What is the weather in Nashik?"
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "toolu_01Kus2cC3ydjBW7UK4GJqBP4",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"location": "Nashik, India"}'
}
}
]
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
# Should have 3 messages: user, assistant, and dummy tool result
assert len(sanitized) == 3
assert sanitized[0]["role"] == "user"
assert sanitized[1]["role"] == "assistant"
assert sanitized[2]["role"] == "tool"
assert sanitized[2]["tool_call_id"] == "toolu_01Kus2cC3ydjBW7UK4GJqBP4"
assert "skipped" in sanitized[2]["content"].lower() or "interrupted" in sanitized[2]["content"].lower()
assert "get_weather" in sanitized[2]["content"]
def test_case_a_orphaned_tool_call_multiple(self):
"""
Test Case A: Assistant message with multiple tool_calls, some missing results
"""
messages = [
{
"role": "user",
"content": "Get weather for Nashik and Mumbai"
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"location": "Nashik"}'
}
},
{
"id": "call_2",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"location": "Mumbai"}'
}
}
]
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": "Weather in Nashik: 25°C"
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
# Should have 4 messages: user, assistant, tool result for call_1, dummy for call_2
assert len(sanitized) == 4
assert sanitized[0]["role"] == "user"
assert sanitized[1]["role"] == "assistant"
assert sanitized[2]["tool_call_id"] == "call_1" # Original tool result (first in tool_calls)
assert sanitized[3]["tool_call_id"] == "call_2" # Dummy added for missing call_2
def test_case_b_orphaned_tool_result(self):
"""
Test Case B: Tool result without matching tool_call in previous assistant message
Should remove the orphaned tool result
"""
messages = [
{
"role": "user",
"content": "Hello"
},
{
"role": "assistant",
"content": "Hi there!"
},
{
"role": "tool",
"tool_call_id": "nonexistent_id",
"content": "Some result"
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
# Should have only 2 messages, orphaned tool result removed
assert len(sanitized) == 2
assert sanitized[0]["role"] == "user"
assert sanitized[1]["role"] == "assistant"
def test_case_b_valid_tool_result_preserved(self):
"""
Test Case B: Valid tool result with matching tool_call should be preserved
"""
messages = [
{
"role": "user",
"content": "What's the weather?"
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_123",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"location": "Boston"}'
}
}
]
},
{
"role": "tool",
"tool_call_id": "call_123",
"content": "Weather: 20°C"
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
# All messages should be preserved
assert len(sanitized) == 3
assert sanitized[2]["role"] == "tool"
assert sanitized[2]["tool_call_id"] == "call_123"
def test_case_c_empty_text_content_user(self):
"""
Test Case C: Empty text content in user message
Should replace with placeholder
"""
messages = [
{
"role": "user",
"content": ""
},
{
"role": "assistant",
"content": "Hello!"
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
assert len(sanitized) == 2
assert sanitized[0]["role"] == "user"
assert sanitized[0]["content"] == "[System: Empty message content sanitised to satisfy protocol]"
def test_case_c_whitespace_only_content(self):
"""
Test Case C: Whitespace-only content
Should replace with placeholder
"""
messages = [
{
"role": "user",
"content": " \n \t "
},
{
"role": "assistant",
"content": " "
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
assert len(sanitized) == 2
assert sanitized[0]["content"] == "[System: Empty message content sanitised to satisfy protocol]"
assert sanitized[1]["content"] == "[System: Empty message content sanitised to satisfy protocol]"
def test_case_c_valid_content_preserved(self):
"""
Test Case C: Valid non-empty content should be preserved
"""
messages = [
{
"role": "user",
"content": "Hello"
},
{
"role": "assistant",
"content": "Hi there!"
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
assert len(sanitized) == 2
assert sanitized[0]["content"] == "Hello"
assert sanitized[1]["content"] == "Hi there!"
def test_combined_cases(self):
"""
Test combination of multiple cases
"""
messages = [
{
"role": "user",
"content": "Get weather"
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"location": "NYC"}'
}
}
]
},
# Missing tool result for call_1
{
"role": "user",
"content": "" # Empty content
},
{
"role": "assistant",
"content": "Response"
},
{
"role": "tool",
"tool_call_id": "orphaned_id", # Orphaned tool result
"content": "Some data"
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
# Should have: user, assistant, dummy tool result, user (sanitized), assistant
# Orphaned tool result should be removed
assert len(sanitized) == 5
assert sanitized[0]["role"] == "user"
assert sanitized[1]["role"] == "assistant"
assert sanitized[2]["role"] == "tool"
assert sanitized[2]["tool_call_id"] == "call_1" # Dummy added
assert sanitized[3]["role"] == "user"
assert sanitized[3]["content"] == "[System: Empty message content sanitised to satisfy protocol]"
assert sanitized[4]["role"] == "assistant"
def test_modify_params_false_no_sanitization(self):
"""
Test that sanitization is skipped when modify_params=False
"""
litellm.modify_params = False
messages = [
{
"role": "user",
"content": ""
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{}'
}
}
]
}
]
sanitized = sanitize_messages_for_tool_calling(messages)
# Messages should be unchanged
assert len(sanitized) == 2
assert sanitized[0]["content"] == ""
assert len(sanitized[1].get("tool_calls", [])) == 1
def test_anthropic_messages_pt_integration(self):
"""
Test that sanitization is integrated into anthropic_messages_pt
"""
litellm.modify_params = True
messages = [
{
"role": "user",
"content": "What is the weather in Nashik?"
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "toolu_01Kus2cC3ydjBW7UK4GJqBP4",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"location": "Nashik, India"}'
}
}
]
}
]
# This should not raise an error and should add dummy tool result
result = anthropic_messages_pt(
messages=messages,
model="claude-sonnet-4-5",
llm_provider="anthropic"
)
# Should have at least 2 messages (user and assistant)
# The tool result will be merged into user content
assert len(result) >= 2
assert result[0]["role"] == "user"
assert result[1]["role"] == "assistant"
if __name__ == "__main__":
pytest.main([__file__, "-v"])