Add ttl to CachePoint and Anthropic caching model settings

pydantic_ai_slim/pydantic_ai/messages.py

@@ -627,6 +627,9 @@ class CachePoint:
     kind: Literal['cache-point'] = 'cache-point'
     """Type identifier, this is available on all parts as a discriminator."""
 
+    ttl: Literal['5m', '1h'] = '5m'
+    """The cache time-to-live, either "5m" (5 minutes) or "1h" (1 hour). See https://docs.claude.com/en/docs/build-with-claude/prompt-caching#1-hour-cache-duration for more information."""
+
 
 MultiModalContent = ImageUrl | AudioUrl | DocumentUrl | VideoUrl | BinaryContent
 UserContent: TypeAlias = str | MultiModalContent | CachePoint

pydantic_ai_slim/pydantic_ai/models/anthropic.py

@@ -158,6 +158,12 @@ class AnthropicModelSettings(ModelSettings, total=False):
     See https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching for more information.
     """
 
+    anthropic_cache_tool_definitions_ttl: Literal['5m', '1h']
+    """The TTL for tool definitions cache control.
+
+    See https://docs.claude.com/en/docs/build-with-claude/prompt-caching#1-hour-cache-duration for more information.
+    """
+
     anthropic_cache_instructions: bool
     """Whether to add `cache_control` to the last system prompt block.
 
@@ -166,6 +172,12 @@ class AnthropicModelSettings(ModelSettings, total=False):
     See https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching for more information.
     """
 
+    anthropic_cache_instructions_ttl: Literal['5m', '1h']
+    """The TTL for system instructions cache control.
+
+    See https://docs.claude.com/en/docs/build-with-claude/prompt-caching#1-hour-cache-duration for more information.
+    """
+
 
 @dataclass(init=False)
 class AnthropicModel(Model):
@@ -439,7 +451,9 @@ def _get_tools(
         # Add cache_control to the last tool if enabled
         if tools and model_settings.get('anthropic_cache_tool_definitions'):
             last_tool = tools[-1]
-            last_tool['cache_control'] = BetaCacheControlEphemeralParam(type='ephemeral')
+            last_tool['cache_control'] = BetaCacheControlEphemeralParam(
+                type='ephemeral', ttl=model_settings.get('anthropic_cache_tool_definitions_ttl', '5m')
+            )
 
         return tools
 
@@ -510,7 +524,7 @@ async def _map_message(  # noqa: C901
                     elif isinstance(request_part, UserPromptPart):
                         async for content in self._map_user_prompt(request_part):
                             if isinstance(content, CachePoint):
-                                self._add_cache_control_to_last_param(user_content_params)
+                                self._add_cache_control_to_last_param(user_content_params, ttl=content.ttl)
                             else:
                                 user_content_params.append(content)
                     elif isinstance(request_part, ToolReturnPart):
@@ -677,15 +691,19 @@ async def _map_message(  # noqa: C901
         if system_prompt and model_settings.get('anthropic_cache_instructions'):
             system_prompt_blocks = [
                 BetaTextBlockParam(
-                    type='text', text=system_prompt, cache_control=BetaCacheControlEphemeralParam(type='ephemeral')
+                    type='text',
+                    text=system_prompt,
+                    cache_control=BetaCacheControlEphemeralParam(
+                        type='ephemeral', ttl=model_settings.get('anthropic_cache_instructions_ttl', '5m')
+                    ),
                 )
             ]
             return system_prompt_blocks, anthropic_messages
 
         return system_prompt, anthropic_messages
 
     @staticmethod
-    def _add_cache_control_to_last_param(params: list[BetaContentBlockParam]) -> None:
+    def _add_cache_control_to_last_param(params: list[BetaContentBlockParam], ttl: Literal['5m', '1h'] = '5m') -> None:
         """Add cache control to the last content block param.
 
         See https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching for more information.
@@ -706,7 +724,7 @@ def _add_cache_control_to_last_param(params: list[BetaContentBlockParam]) -> Non
             raise UserError(f'Cache control not supported for param type: {last_param["type"]}')
 
         # Add cache_control to the last param
-        last_param['cache_control'] = BetaCacheControlEphemeralParam(type='ephemeral')
+        last_param['cache_control'] = BetaCacheControlEphemeralParam(type='ephemeral', ttl=ttl)
 
     @staticmethod
     async def _map_user_prompt(

tests/models/test_anthropic.py

@@ -314,7 +314,11 @@ async def test_cache_point_adds_cache_control(allow_model_requests: None):
             {
                 'role': 'user',
                 'content': [
-                    {'text': 'Some context to cache', 'type': 'text', 'cache_control': {'type': 'ephemeral'}},
+                    {
+                        'text': 'Some context to cache',
+                        'type': 'text',
+                        'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
+                    },
                     {'text': 'Now the question', 'type': 'text'},
                 ],
             }
@@ -339,8 +343,8 @@ async def test_cache_point_multiple_markers(allow_model_requests: None):
 
     assert content == snapshot(
         [
-            {'text': 'First chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral'}},
-            {'text': 'Second chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral'}},
+            {'text': 'First chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral', 'ttl': '5m'}},
+            {'text': 'Second chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral', 'ttl': '5m'}},
             {'text': 'Question', 'type': 'text'},
         ]
     )
@@ -389,7 +393,7 @@ async def test_cache_point_with_image_content(allow_model_requests: None):
             {
                 'source': {'type': 'url', 'url': 'https://example.com/image.jpg'},
                 'type': 'image',
-                'cache_control': {'type': 'ephemeral'},
+                'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
             },
             {'text': 'What is in this image?', 'type': 'text'},
         ]
@@ -466,7 +470,7 @@ def tool_two() -> str:  # pragma: no cover
                 'name': 'tool_two',
                 'description': '',
                 'input_schema': {'additionalProperties': False, 'properties': {}, 'type': 'object'},
-                'cache_control': {'type': 'ephemeral'},
+                'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
             },
         ]
     )
@@ -496,7 +500,7 @@ async def test_anthropic_cache_instructions(allow_model_requests: None):
             {
                 'type': 'text',
                 'text': 'This is a test system prompt with instructions.',
-                'cache_control': {'type': 'ephemeral'},
+                'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
             }
         ]
     )
@@ -540,12 +544,12 @@ def my_tool(value: str) -> str:  # pragma: no cover
                     'required': ['value'],
                     'type': 'object',
                 },
-                'cache_control': {'type': 'ephemeral'},
+                'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
             }
         ]
     )
     assert system == snapshot(
-        [{'type': 'text', 'text': 'System instructions to cache.', 'cache_control': {'type': 'ephemeral'}}]
+        [{'type': 'text', 'text': 'System instructions to cache.', 'cache_control': {'type': 'ephemeral', 'ttl': '5m'}}]
     )