Coverage for pydantic_ai_slim/pydantic_ai/

1"""Used to build pydantic validators and JSON schemas from functions.

3This module has to use numerous internal Pydantic APIs and is therefore brittle to changes in Pydantic.

4"""

6from __future__ import annotations as _annotations (empty)

8from inspect import Parameter, signature (empty)

9from typing import TYPE_CHECKING, Any, Callable, TypedDict, cast, get_origin (empty)

11from pydantic import ConfigDict (empty)

12from pydantic._internal import _decorators, _generate_schema, _typing_extra (empty)

13from pydantic._internal._config import ConfigWrapper (empty)

14from pydantic.fields import FieldInfo (empty)

15from pydantic.json_schema import GenerateJsonSchema (empty)

16from pydantic.plugin._schema_validator import create_schema_validator (empty)

17from pydantic_core import SchemaValidator, core_schema (empty)

19from ._griffe import doc_descriptions (empty)

20from ._utils import check_object_json_schema, is_model_like (empty)

22if TYPE_CHECKING: (empty)

23 from .tools import DocstringFormat, ObjectJsonSchema

26__all__ = ('function_schema',) (empty)

29class FunctionSchema(TypedDict): (empty)

30 """Internal information about a function schema."""

32 description: str (empty)

33 validator: SchemaValidator (empty)

34 json_schema: ObjectJsonSchema (empty)

35 # if not None, the function takes a single by that name (besides potentially `info`)

36 single_arg_name: str | None (empty)

37 positional_fields: list[str] (empty)

38 var_positional_field: str | None (empty)

41def function_schema( # noqa: C901 (empty)

42 function: Callable[..., Any],

43 takes_ctx: bool,

44 docstring_format: DocstringFormat,

45 require_parameter_descriptions: bool,

46) -> FunctionSchema:

47 """Build a Pydantic validator and JSON schema from a tool function.

49 Args:

50 function: The function to build a validator and JSON schema for.

51 takes_ctx: Whether the function takes a `RunContext` first argument.

52 docstring_format: The docstring format to use.

53 require_parameter_descriptions: Whether to require descriptions for all tool function parameters.

55 Returns:

56 A `FunctionSchema` instance.

57 """

58 config = ConfigDict(title=function.__name__) (empty)

59 config_wrapper = ConfigWrapper(config) (empty)

60 gen_schema = _generate_schema.GenerateSchema(config_wrapper) (empty)

62 sig = signature(function) (empty)

64 type_hints = _typing_extra.get_function_type_hints(function) (empty)

66 var_kwargs_schema: core_schema.CoreSchema | None = None (empty)

67 fields: dict[str, core_schema.TypedDictField] = {} (empty)

68 positional_fields: list[str] = [] (empty)

69 var_positional_field: str | None = None (empty)

70 errors: list[str] = [] (empty)

71 decorators = _decorators.DecoratorInfos() (empty)

73 description, field_descriptions = doc_descriptions(function, sig, docstring_format=docstring_format) (empty)

75 if require_parameter_descriptions: (empty)

76 if len(field_descriptions) != len(sig.parameters): 76 ↛ 80line 76 didn't jump to line 80 because the condition on line 76 was always true(empty)

77 missing_params = set(sig.parameters) - set(field_descriptions) (empty)

78 errors.append(f'Missing parameter descriptions for {", ".join(missing_params)}') (empty)

80 for index, (name, p) in enumerate(sig.parameters.items()): (empty)

81 if p.annotation is sig.empty: (empty)

82 if takes_ctx and index == 0: (empty)

83 # should be the `context` argument, skip

84 continue (empty)

85 # TODO warn?

86 annotation = Any (empty)

87 else:

88 annotation = type_hints[name] (empty)

90 if index == 0 and takes_ctx: (empty)

91 if not _is_call_ctx(annotation): (empty)

92 errors.append('First parameter of tools that take context must be annotated with RunContext[...]') (empty)

93 continue (empty)

94 elif not takes_ctx and _is_call_ctx(annotation): (empty)

95 errors.append('RunContext annotations can only be used with tools that take context') (empty)

96 continue (empty)

97 elif index != 0 and _is_call_ctx(annotation): (empty)

98 errors.append('RunContext annotations can only be used as the first argument') (empty)

99 continue (empty)

100

101 field_name = p.name (empty)

102 if p.kind == Parameter.VAR_KEYWORD: (empty)

103 var_kwargs_schema = gen_schema.generate_schema(annotation) (empty)

104 else:

105 if p.kind == Parameter.VAR_POSITIONAL: (empty)

106 annotation = list[annotation] (empty)

107

108 # FieldInfo.from_annotation expects a type, `annotation` is Any

109 annotation = cast(type[Any], annotation) (empty)

110 field_info = FieldInfo.from_annotation(annotation) (empty)

111 if field_info.description is None: (empty)

112 field_info.description = field_descriptions.get(field_name) (empty)

113

114 fields[field_name] = td_schema = gen_schema._generate_td_field_schema( # pyright: ignore[reportPrivateUsage] (empty)

115 field_name,

116 field_info,

117 decorators,

118 required=p.default is Parameter.empty,

119 )

120 # noinspection PyTypeChecker

121 td_schema.setdefault('metadata', {})['is_model_like'] = is_model_like(annotation) (empty)

122

123 if p.kind == Parameter.POSITIONAL_ONLY: (empty)

124 positional_fields.append(field_name) (empty)

125 elif p.kind == Parameter.VAR_POSITIONAL: (empty)

126 var_positional_field = field_name (empty)

127

128 if errors: (empty)

129 from .exceptions import UserError (empty)

130

131 error_details = '\n '.join(errors) (empty)

132 raise UserError(f'Error generating schema for {function.__qualname__}:\n {error_details}') (empty)

133

134 core_config = config_wrapper.core_config(None) (empty)

135 # noinspection PyTypedDict

136 core_config['extra_fields_behavior'] = 'allow' if var_kwargs_schema else 'forbid' (empty)

137

138 schema, single_arg_name = _build_schema(fields, var_kwargs_schema, gen_schema, core_config) (empty)

139 schema = gen_schema.clean_schema(schema) (empty)

140 # noinspection PyUnresolvedReferences

141 schema_validator = create_schema_validator( (empty)

142 schema,

143 function,

144 function.__module__,

145 function.__qualname__,

146 'validate_call',

147 core_config,

148 config_wrapper.plugin_settings,

149 )

150 # PluggableSchemaValidator is api compatible with SchemaValidator

151 schema_validator = cast(SchemaValidator, schema_validator) (empty)

152 json_schema = GenerateJsonSchema().generate(schema) (empty)

153

154 # workaround for https://github.com/pydantic/pydantic/issues/10785

155 # if we build a custom TypeDict schema (matches when `single_arg_name is None`), we manually set

156 # `additionalProperties` in the JSON Schema

157 if single_arg_name is None: (empty)

158 json_schema['additionalProperties'] = bool(var_kwargs_schema) (empty)

159 elif not description: 159 ↛ 164line 159 didn't jump to line 164 because the condition on line 159 was always true(empty)

160 # if the tool description is not set, and we have a single parameter, take the description from that

161 # and set it on the tool

162 description = json_schema.pop('description', None) (empty)

163

164 return FunctionSchema( (empty)

165 description=description,

166 validator=schema_validator,

167 json_schema=check_object_json_schema(json_schema),

168 single_arg_name=single_arg_name,

169 positional_fields=positional_fields,

170 var_positional_field=var_positional_field,

171 )

172

173

174def takes_ctx(function: Callable[..., Any]) -> bool: (empty)

175 """Check if a function takes a `RunContext` first argument.

176

177 Args:

178 function: The function to check.

179

180 Returns:

181 `True` if the function takes a `RunContext` as first argument, `False` otherwise.

182 """

183 sig = signature(function) (empty)

184 try: (empty)

185 first_param_name = next(iter(sig.parameters.keys())) (empty)

186 except StopIteration: (empty)

187 return False (empty)

188 else:

189 type_hints = _typing_extra.get_function_type_hints(function) (empty)

190 annotation = type_hints[first_param_name] (empty)

191 return annotation is not sig.empty and _is_call_ctx(annotation) (empty)

192

193

194def _build_schema( (empty)

195 fields: dict[str, core_schema.TypedDictField],

196 var_kwargs_schema: core_schema.CoreSchema | None,

197 gen_schema: _generate_schema.GenerateSchema,

198 core_config: core_schema.CoreConfig,

199) -> tuple[core_schema.CoreSchema, str | None]:

200 """Generate a typed dict schema for function parameters.

201

202 Args:

203 fields: The fields to generate a typed dict schema for.

204 var_kwargs_schema: The variable keyword arguments schema.

205 gen_schema: The `GenerateSchema` instance.

206 core_config: The core configuration.

207

208 Returns:

209 tuple of (generated core schema, single arg name).

210 """

211 if len(fields) == 1 and var_kwargs_schema is None: (empty)

212 name = next(iter(fields)) (empty)

213 td_field = fields[name] (empty)

214 if td_field['metadata']['is_model_like']: # type: ignore (empty)

215 return td_field['schema'], name (empty)

216

217 td_schema = core_schema.typed_dict_schema( (empty)

218 fields,

219 config=core_config,

220 extras_schema=gen_schema.generate_schema(var_kwargs_schema) if var_kwargs_schema else None,

221 )

222 return td_schema, None (empty)

223

224

225def _is_call_ctx(annotation: Any) -> bool: (empty)

226 from .tools import RunContext (empty)

227

228 return annotation is RunContext or ( (empty)

229 _typing_extra.is_generic_alias(annotation) and get_origin(annotation) is RunContext

230 )

Coverage for pydantic_ai_slim/pydantic_ai/_pydantic.py: 98.59%

104 statements