Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions src/strands/models/litellm.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,11 +220,14 @@ def _format_system_messages(
for block in system_prompt_content or []:
if "text" in block:
system_content.append({"type": "text", "text": block["text"]})
elif "cachePoint" in block and block["cachePoint"].get("type") == "default":
elif "cachePoint" in block and block["cachePoint"]["type"] == "default":
# Apply cache control to the immediately preceding content block
# for LiteLLM/Anthropic compatibility
if system_content:
system_content[-1]["cache_control"] = {"type": "ephemeral"}
cache_control: dict[str, Any] = {"type": "ephemeral"}
if ttl := block["cachePoint"].get("ttl"):
Comment thread
mkmeral marked this conversation as resolved.
cache_control["ttl"] = ttl
system_content[-1]["cache_control"] = cache_control

# Create single system message with content array rather than mulitple system messages
return [{"role": "system", "content": system_content}] if system_content else []
Expand Down
3 changes: 3 additions & 0 deletions src/strands/types/content.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,9 +67,12 @@ class CachePoint(TypedDict):
Attributes:
type: The type of cache point, typically "default".
ttl: Optional cache TTL duration (e.g. "5m", "1h"). Supported by providers
that accept Anthropic-compatible cache_control fields.
"""

type: str
ttl: NotRequired[str]


class ContentBlock(TypedDict, total=False):
Expand Down
33 changes: 33 additions & 0 deletions tests/strands/models/test_litellm.py
Original file line number Diff line number Diff line change
Expand Up @@ -955,6 +955,39 @@ def test_format_request_message_tool_call_no_reasoning_signature():
assert "__thought__" not in result["id"]


def test_format_system_messages_preserves_cache_point_ttl():
"""CachePoint with ttl="1h" should produce cache_control with ttl field."""
result = LiteLLMModel._format_system_messages(
system_prompt_content=[
{"text": "You are a helpful assistant."},
{"cachePoint": {"type": "default", "ttl": "1h"}},
]
)
assert result[0]["content"][0]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}


def test_format_system_messages_cache_point_without_ttl():
"""CachePoint without ttl should produce cache_control with no ttl key (backward compat)."""
result = LiteLLMModel._format_system_messages(
system_prompt_content=[
{"text": "You are a helpful assistant."},
{"cachePoint": {"type": "default"}},
]
)
assert result[0]["content"][0]["cache_control"] == {"type": "ephemeral"}
assert "ttl" not in result[0]["content"][0]["cache_control"]


def test_format_system_messages_cache_point_with_no_preceding_content():
"""CachePoint with no preceding text block should be silently ignored."""
result = LiteLLMModel._format_system_messages(
system_prompt_content=[
{"cachePoint": {"type": "default", "ttl": "1h"}},
]
)
assert result == []


def test_thought_signature_round_trip():
"""Test that thought signature is preserved through a full response -> internal -> request cycle."""
model = LiteLLMModel(model_id="test")
Expand Down
Loading