Skip to content
Closed
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
23 changes: 20 additions & 3 deletions pydantic_ai_slim/pydantic_ai/models/anthropic.py
Original file line number Diff line number Diff line change
Expand Up @@ -835,14 +835,31 @@ async def _map_user_prompt(
else:
raise RuntimeError(f'Unsupported content type: {type(item)}') # pragma: no cover

@staticmethod
def _map_tool_definition(f: ToolDefinition) -> BetaToolParam:
return {
def _map_tool_definition(self, f: ToolDefinition) -> BetaToolParam:
"""Map tool definition to Anthropic format with strict mode support.

Follows the same pattern as OpenAI (line 768): checks if f.strict is truthy,
not just checking if it's not None (which was PR #3457's critical mistake).

Args:
f: Tool definition to map

Returns:
BetaToolParam with strict mode set if applicable
"""
tool_param: BetaToolParam = {
'name': f.name,
'description': f.description or '',
'input_schema': f.parameters_json_schema,
}

# ✅ CRITICAL: Only add strict if explicitly True (not None, not False)
# This prevents the strict field from leaking into tools that don't explicitly set it
if f.strict is True:
tool_param['strict'] = True # type: ignore[typeddict-item]

return tool_param


def _map_usage(
message: BetaMessage | BetaRawMessageStartEvent | BetaRawMessageDeltaEvent,
Expand Down
88 changes: 86 additions & 2 deletions pydantic_ai_slim/pydantic_ai/profiles/anthropic.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,92 @@
from __future__ import annotations as _annotations

from dataclasses import dataclass
from typing import Any

from .._json_schema import JsonSchema, JsonSchemaTransformer
from . import ModelProfile

try:
from anthropic.lib.tools import transform_schema as anthropic_transform_schema # type: ignore[import-not-found]
except ImportError:
anthropic_transform_schema = None


@dataclass(init=False)
class AnthropicJsonSchemaTransformer(JsonSchemaTransformer):
"""Transform JSON schemas for Anthropic structured outputs.

Anthropic requires schemas to not include 'title', '$schema', or 'discriminator' at root level.

`additionalProperties` is set to false for objects to ensure strict mode compatibility.

optionally use Anthropic's transform_schema helper for validation

see: https://docs.claude.com/en/docs/build-with-claude/structured-outputs
"""

def transform(self, schema: JsonSchema) -> JsonSchema:
"""Apply Anthropic-specific schema transformations.

Removes 'title', '$schema', and 'discriminator' fields which are not supported by Anthropic API,
and sets `additionalProperties` to false for objects to ensure strict mode compatibility.

If available, also validates the schema using Anthropic's transform_schema helper from their SDK.

Args:
schema: The JSON schema to transform

Returns:
Transformed schema compatible with Anthropic's structured outputs API
"""
# remove fields not supported by Anthropic
schema.pop('title', None)
schema.pop('$schema', None)
schema.pop('discriminator', None)

schema_type = schema.get('type')
if schema_type == 'object':
schema['additionalProperties'] = False
if self.strict is True:
if 'properties' not in schema:
schema['properties'] = dict[str, Any]()
schema['required'] = list(schema['properties'].keys())
elif self.strict is None:
if schema.get('additionalProperties', None) not in (None, False):
self.is_strict_compatible = False
else:
schema['additionalProperties'] = False

if 'properties' not in schema or 'required' not in schema:
self.is_strict_compatible = False
else:
required = schema['required']
for k in schema['properties'].keys():
if k not in required:
self.is_strict_compatible = False
else:
if 'additionalProperties' not in schema:
schema['additionalProperties'] = False

if anthropic_transform_schema is not None:
try:
validated_schema = anthropic_transform_schema(schema) # pyright: ignore[reportUnknownVariableType]
if isinstance(validated_schema, dict):
schema = validated_schema
except Exception:
pass

if self.strict is True:
self.is_strict_compatible = True

return schema


def anthropic_model_profile(model_name: str) -> ModelProfile | None:
def anthropic_model_profile(model_name: str) -> ModelProfile:
"""Get the model profile for an Anthropic model."""
return ModelProfile(thinking_tags=('<thinking>', '</thinking>'))
return ModelProfile(
json_schema_transformer=AnthropicJsonSchemaTransformer,
supports_json_schema_output=True,
supports_json_object_output=True,
thinking_tags=('<thinking>', '</thinking>'),
)
99 changes: 99 additions & 0 deletions tests/models/test_anthropic.py
Original file line number Diff line number Diff line change
Expand Up @@ -545,6 +545,7 @@ def my_tool(value: str) -> str: # pragma: no cover
'required': ['value'],
'type': 'object',
},
'strict': True,
'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
}
]
Expand Down Expand Up @@ -6516,3 +6517,101 @@ async def test_anthropic_bedrock_count_tokens_not_supported(env: TestEnv):

with pytest.raises(UserError, match='AsyncAnthropicBedrock client does not support `count_tokens` api.'):
await agent.run('hello', usage_limits=UsageLimits(input_tokens_limit=20, count_tokens_before_request=True))


def test_anthropic_json_schema_transformer():
"""
Test AnthropicJsonSchemaTransformer removes unsupported fields.
"""
from pydantic_ai.profiles.anthropic import AnthropicJsonSchemaTransformer

schema = {
'type': 'object',
'title': 'testSchema',
'$schema': 'http://json-schema.org/draft-07/schema#',
'discriminator': 'some_value',
'properties': {'name': {'type': 'string'}, 'age': {'type': 'integer'}},
'required': ['name'],
}

transformer = AnthropicJsonSchemaTransformer(schema, strict=True)
result = transformer.walk()

assert 'title' not in result
assert '$schema' not in result
assert 'discriminator' not in result
assert result['type'] == 'object'
assert result['additionalProperties'] is False
assert result['properties'] == schema['properties']
# In strict mode, Anthropic requires ALL properties to be in required list
assert result['required'] == ['name', 'age']
assert transformer.is_strict_compatible is True

# test auto-detect mode (strict=False)
schema2 = {
'type': 'object',
'properties': {'id': {'type': 'integer'}},
'required': ['id'],
}
transformer2 = AnthropicJsonSchemaTransformer(schema2, strict=False)
result2 = transformer2.walk()
assert result2['additionalProperties'] is False
assert transformer2.is_strict_compatible is True


def test_anthropic_strict_mode_tool_mapping():
"""
Test AnthropicModel._map_tool_definition() supports strict mode.
"""
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.providers.anthropic import AnthropicProvider
from pydantic_ai.tools import ToolDefinition

# Use mock provider to avoid requiring real API key
model = AnthropicModel('claude-sonnet-4-5', provider=AnthropicProvider(api_key='test-key'))

tool_def_strict = ToolDefinition(
name='test_tool',
description='a test tool',
parameters_json_schema={'type': 'object', 'properties': {'x': {'type': 'integer'}}, 'required': ['x']},
strict=True,
)

result_strict = model._map_tool_definition(tool_def_strict) # pyright: ignore[reportPrivateUsage]

assert result_strict['strict'] is True # pyright: ignore[reportGeneralTypeIssues]
assert result_strict['name'] == 'test_tool'
assert result_strict.get('description') == 'a test tool'

tool_def_no_strict = ToolDefinition(
name='test_tool',
description='a test tool',
parameters_json_schema={'type': 'object', 'properties': {'x': {'type': 'integer'}}, 'required': ['x']},
strict=False,
)

result_no_strict = model._map_tool_definition(tool_def_no_strict) # pyright: ignore[reportPrivateUsage]
assert 'strict' not in result_no_strict
assert result_no_strict['name'] == 'test_tool'
assert result_no_strict.get('description') == 'a test tool'


async def test_anthropic_strucutred_output_with_test_model():
from pydantic import BaseModel

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

class CityInfo(BaseModel):
city: str
country: str
population: int

test_model = TestModel(custom_output_args={'city': 'Tokyo', 'country': 'Japan', 'population': 14000000})
agent = Agent(test_model, output_type=CityInfo)

result = await agent.run('Tell me about Tokyo')
assert isinstance(result.output, CityInfo)
assert result.output.city == 'Tokyo'
assert result.output.country == 'Japan'
assert result.output.population == 14000000
Loading