@dataclass(repr=False)classSystemPromptPart:"""A system prompt, generally written by the application developer. This gives the model context and guidance on how to respond. """content:str"""The content of the prompt."""timestamp:datetime=field(default_factory=_now_utc)"""The timestamp of the prompt."""dynamic_ref:str|None=None"""The ref of the dynamic system prompt function that generated this part. Only set if system prompt is dynamic, see [`system_prompt`][pydantic_ai.Agent.system_prompt] for more information. """part_kind:Literal['system-prompt']='system-prompt'"""Part type identifier, this is available on all parts as a discriminator."""defotel_event(self,_settings:InstrumentationSettings)->Event:returnEvent('gen_ai.system.message',body={'content':self.content,'role':'system'})__repr__=_utils.dataclasses_no_defaults_repr
@dataclass(repr=False)classFileUrl(ABC):"""Abstract base class for any URL-based file."""url:str"""The URL of the file."""force_download:bool=False"""If the model supports it: * If True, the file is downloaded and the data is sent to the model as bytes. * If False, the URL is sent directly to the model and no download is performed. """@property@abstractmethoddefmedia_type(self)->str:"""Return the media type of the file, based on the url."""@property@abstractmethoddefformat(self)->str:"""The file format."""__repr__=_utils.dataclasses_no_defaults_repr
@dataclass(repr=False)classVideoUrl(FileUrl):"""A URL to a video."""url:str"""The URL of the video."""kind:Literal['video-url']='video-url'"""Type identifier, this is available on all parts as a discriminator."""@propertydefmedia_type(self)->VideoMediaType:"""Return the media type of the video, based on the url."""ifself.url.endswith('.mkv'):return'video/x-matroska'elifself.url.endswith('.mov'):return'video/quicktime'elifself.url.endswith('.mp4'):return'video/mp4'elifself.url.endswith('.webm'):return'video/webm'elifself.url.endswith('.flv'):return'video/x-flv'elifself.url.endswith(('.mpeg','.mpg')):return'video/mpeg'elifself.url.endswith('.wmv'):return'video/x-ms-wmv'elifself.url.endswith('.three_gp'):return'video/3gpp'# Assume that YouTube videos are mp4 because there would be no extension# to infer from. This should not be a problem, as Gemini disregards media# type for YouTube URLs.elifself.is_youtube:return'video/mp4'else:raiseValueError(f'Unknown video file extension: {self.url}')@propertydefis_youtube(self)->bool:"""True if the URL has a YouTube domain."""returnself.url.startswith(('https://youtu.be/','https://youtube.com/','https://www.youtube.com/'))@propertydefformat(self)->VideoFormat:"""The file format of the video. The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. """return_video_format_lookup[self.media_type]
@dataclass(repr=False)classAudioUrl(FileUrl):"""A URL to an audio file."""url:str"""The URL of the audio file."""kind:Literal['audio-url']='audio-url'"""Type identifier, this is available on all parts as a discriminator."""@propertydefmedia_type(self)->AudioMediaType:"""Return the media type of the audio file, based on the url."""ifself.url.endswith('.mp3'):return'audio/mpeg'elifself.url.endswith('.wav'):return'audio/wav'else:raiseValueError(f'Unknown audio file extension: {self.url}')@propertydefformat(self)->AudioFormat:"""The file format of the audio file."""return_audio_format_lookup[self.media_type]
@dataclass(repr=False)classImageUrl(FileUrl):"""A URL to an image."""url:str"""The URL of the image."""kind:Literal['image-url']='image-url'"""Type identifier, this is available on all parts as a discriminator."""@propertydefmedia_type(self)->ImageMediaType:"""Return the media type of the image, based on the url."""ifself.url.endswith(('.jpg','.jpeg')):return'image/jpeg'elifself.url.endswith('.png'):return'image/png'elifself.url.endswith('.gif'):return'image/gif'elifself.url.endswith('.webp'):return'image/webp'else:raiseValueError(f'Unknown image file extension: {self.url}')@propertydefformat(self)->ImageFormat:"""The file format of the image. The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. """return_image_format_lookup[self.media_type]
@dataclass(repr=False)classDocumentUrl(FileUrl):"""The URL of the document."""url:str"""The URL of the document."""kind:Literal['document-url']='document-url'"""Type identifier, this is available on all parts as a discriminator."""@propertydefmedia_type(self)->str:"""Return the media type of the document, based on the url."""type_,_=guess_type(self.url)iftype_isNone:raiseValueError(f'Unknown document file extension: {self.url}')returntype_@propertydefformat(self)->DocumentFormat:"""The file format of the document. The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. """media_type=self.media_typetry:return_document_format_lookup[media_type]exceptKeyErrorase:raiseValueError(f'Unknown document media type: {media_type}')frome
@dataclass(repr=False)classBinaryContent:"""Binary content, e.g. an audio or image file."""data:bytes"""The binary data."""media_type:AudioMediaType|ImageMediaType|DocumentMediaType|str"""The media type of the binary data."""kind:Literal['binary']='binary'"""Type identifier, this is available on all parts as a discriminator."""@propertydefis_audio(self)->bool:"""Return `True` if the media type is an audio type."""returnself.media_type.startswith('audio/')@propertydefis_image(self)->bool:"""Return `True` if the media type is an image type."""returnself.media_type.startswith('image/')@propertydefis_video(self)->bool:"""Return `True` if the media type is a video type."""returnself.media_type.startswith('video/')@propertydefis_document(self)->bool:"""Return `True` if the media type is a document type."""returnself.media_typein_document_format_lookup@propertydefformat(self)->str:"""The file format of the binary content."""try:ifself.is_audio:return_audio_format_lookup[self.media_type]elifself.is_image:return_image_format_lookup[self.media_type]elifself.is_video:return_video_format_lookup[self.media_type]else:return_document_format_lookup[self.media_type]exceptKeyErrorase:raiseValueError(f'Unknown media type: {self.media_type}')frome__repr__=_utils.dataclasses_no_defaults_repr
@dataclass(repr=False)classUserPromptPart:"""A user prompt, generally written by the end user. Content comes from the `user_prompt` parameter of [`Agent.run`][pydantic_ai.Agent.run], [`Agent.run_sync`][pydantic_ai.Agent.run_sync], and [`Agent.run_stream`][pydantic_ai.Agent.run_stream]. """content:str|Sequence[UserContent]"""The content of the prompt."""timestamp:datetime=field(default_factory=_now_utc)"""The timestamp of the prompt."""part_kind:Literal['user-prompt']='user-prompt'"""Part type identifier, this is available on all parts as a discriminator."""defotel_event(self,settings:InstrumentationSettings)->Event:content:str|list[dict[str,Any]|str]ifisinstance(self.content,str):content=self.contentelse:content=[]forpartinself.content:ifisinstance(part,str):content.append(part)elifisinstance(part,(ImageUrl,AudioUrl,DocumentUrl,VideoUrl)):content.append({'kind':part.kind,'url':part.url})elifisinstance(part,BinaryContent):converted_part={'kind':part.kind,'media_type':part.media_type}ifsettings.include_binary_content:converted_part['binary_content']=base64.b64encode(part.data).decode()content.append(converted_part)else:content.append({'kind':part.kind})# pragma: no coverreturnEvent('gen_ai.user.message',body={'content':content,'role':'user'})__repr__=_utils.dataclasses_no_defaults_repr
@dataclass(repr=False)classToolReturnPart:"""A tool return message, this encodes the result of running a tool."""tool_name:str"""The name of the "tool" was called."""content:Any"""The return value."""tool_call_id:str"""The tool call identifier, this is used by some models including OpenAI."""timestamp:datetime=field(default_factory=_now_utc)"""The timestamp, when the tool returned."""part_kind:Literal['tool-return']='tool-return'"""Part type identifier, this is available on all parts as a discriminator."""defmodel_response_str(self)->str:"""Return a string representation of the content for the model."""ifisinstance(self.content,str):returnself.contentelse:returntool_return_ta.dump_json(self.content).decode()defmodel_response_object(self)->dict[str,Any]:"""Return a dictionary representation of the content, wrapping non-dict types appropriately."""# gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dictifisinstance(self.content,dict):returntool_return_ta.dump_python(self.content,mode='json')# pyright: ignore[reportUnknownMemberType]else:return{'return_value':tool_return_ta.dump_python(self.content,mode='json')}defotel_event(self,_settings:InstrumentationSettings)->Event:returnEvent('gen_ai.tool.message',body={'content':self.content,'role':'tool','id':self.tool_call_id,'name':self.tool_name},)__repr__=_utils.dataclasses_no_defaults_repr
Return a string representation of the content for the model.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
403404405406407408
defmodel_response_str(self)->str:"""Return a string representation of the content for the model."""ifisinstance(self.content,str):returnself.contentelse:returntool_return_ta.dump_json(self.content).decode()
Return a dictionary representation of the content, wrapping non-dict types appropriately.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
410411412413414415416
defmodel_response_object(self)->dict[str,Any]:"""Return a dictionary representation of the content, wrapping non-dict types appropriately."""# gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dictifisinstance(self.content,dict):returntool_return_ta.dump_python(self.content,mode='json')# pyright: ignore[reportUnknownMemberType]else:return{'return_value':tool_return_ta.dump_python(self.content,mode='json')}
RetryPromptPartdataclass
A message back to a model asking it to try again.
This can be sent for a number of reasons:
Pydantic validation of tool arguments failed, here content is derived from a Pydantic
ValidationError
@dataclass(repr=False)classRetryPromptPart:"""A message back to a model asking it to try again. This can be sent for a number of reasons: * Pydantic validation of tool arguments failed, here content is derived from a Pydantic [`ValidationError`][pydantic_core.ValidationError] * a tool raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception * no tool was found for the tool name * the model returned plain text when a structured response was expected * Pydantic validation of a structured response failed, here content is derived from a Pydantic [`ValidationError`][pydantic_core.ValidationError] * an output validator raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception """content:list[pydantic_core.ErrorDetails]|str"""Details of why and how the model should retry. If the retry was triggered by a [`ValidationError`][pydantic_core.ValidationError], this will be a list of error details. """tool_name:str|None=None"""The name of the tool that was called, if any."""tool_call_id:str=field(default_factory=_generate_tool_call_id)"""The tool call identifier, this is used by some models including OpenAI. In case the tool call id is not provided by the model, PydanticAI will generate a random one. """timestamp:datetime=field(default_factory=_now_utc)"""The timestamp, when the retry was triggered."""part_kind:Literal['retry-prompt']='retry-prompt'"""Part type identifier, this is available on all parts as a discriminator."""defmodel_response(self)->str:"""Return a string message describing why the retry is requested."""ifisinstance(self.content,str):description=self.contentelse:json_errors=error_details_ta.dump_json(self.content,exclude={'__all__':{'ctx'}},indent=2)description=f'{len(self.content)} validation errors: {json_errors.decode()}'returnf'{description}\n\nFix the errors and try again.'defotel_event(self,_settings:InstrumentationSettings)->Event:ifself.tool_nameisNone:returnEvent('gen_ai.user.message',body={'content':self.model_response(),'role':'user'})else:returnEvent('gen_ai.tool.message',body={'content':self.model_response(),'role':'tool','id':self.tool_call_id,'name':self.tool_name,},)__repr__=_utils.dataclasses_no_defaults_repr
Return a string message describing why the retry is requested.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
468469470471472473474475
defmodel_response(self)->str:"""Return a string message describing why the retry is requested."""ifisinstance(self.content,str):description=self.contentelse:json_errors=error_details_ta.dump_json(self.content,exclude={'__all__':{'ctx'}},indent=2)description=f'{len(self.content)} validation errors: {json_errors.decode()}'returnf'{description}\n\nFix the errors and try again.'
@dataclass(repr=False)classModelRequest:"""A request generated by PydanticAI and sent to a model, e.g. a message from the PydanticAI app to the model."""parts:list[ModelRequestPart]"""The parts of the user message."""instructions:str|None=None"""The instructions for the model."""kind:Literal['request']='request'"""Message type identifier, this is available on all parts as a discriminator."""@classmethoddefuser_text_prompt(cls,user_prompt:str,*,instructions:str|None=None)->ModelRequest:"""Create a `ModelRequest` with a single user prompt as text."""returncls(parts=[UserPromptPart(user_prompt)],instructions=instructions)__repr__=_utils.dataclasses_no_defaults_repr
Create a ModelRequest with a single user prompt as text.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
513514515516
@classmethoddefuser_text_prompt(cls,user_prompt:str,*,instructions:str|None=None)->ModelRequest:"""Create a `ModelRequest` with a single user prompt as text."""returncls(parts=[UserPromptPart(user_prompt)],instructions=instructions)
TextPartdataclass
A plain text response from a model.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
521522523524525526527528529530531532533534535
@dataclass(repr=False)classTextPart:"""A plain text response from a model."""content:str"""The text content of the response."""part_kind:Literal['text']='text'"""Part type identifier, this is available on all parts as a discriminator."""defhas_content(self)->bool:"""Return `True` if the text content is non-empty."""returnbool(self.content)__repr__=_utils.dataclasses_no_defaults_repr
@dataclass(repr=False)classThinkingPart:"""A thinking response from a model."""content:str"""The thinking content of the response."""id:str|None=None"""The identifier of the thinking part."""signature:str|None=None"""The signature of the thinking. The signature is only available on the Anthropic models. """part_kind:Literal['thinking']='thinking'"""Part type identifier, this is available on all parts as a discriminator."""defhas_content(self)->bool:"""Return `True` if the thinking content is non-empty."""returnbool(self.content)__repr__=_utils.dataclasses_no_defaults_repr
@dataclass(repr=False)classToolCallPart:"""A tool call from a model."""tool_name:str"""The name of the tool to call."""args:str|dict[str,Any]|None=None"""The arguments to pass to the tool. This is stored either as a JSON string or a Python dictionary depending on how data was received. """tool_call_id:str=field(default_factory=_generate_tool_call_id)"""The tool call identifier, this is used by some models including OpenAI. In case the tool call id is not provided by the model, PydanticAI will generate a random one. """part_kind:Literal['tool-call']='tool-call'"""Part type identifier, this is available on all parts as a discriminator."""defargs_as_dict(self)->dict[str,Any]:"""Return the arguments as a Python dictionary. This is just for convenience with models that require dicts as input. """ifnotself.args:return{}ifisinstance(self.args,dict):returnself.argsargs=pydantic_core.from_json(self.args)assertisinstance(args,dict),'args should be a dict'returncast(dict[str,Any],args)defargs_as_json_str(self)->str:"""Return the arguments as a JSON string. This is just for convenience with models that require JSON strings as input. """ifnotself.args:return'{}'ifisinstance(self.args,str):returnself.argsreturnpydantic_core.to_json(self.args).decode()defhas_content(self)->bool:"""Return `True` if the arguments contain any data."""ifisinstance(self.args,dict):# TODO: This should probably return True if you have the value False, or 0, etc.# It makes sense to me to ignore empty strings, but not sure about empty lists or dictsreturnany(self.args.values())else:returnbool(self.args)__repr__=_utils.dataclasses_no_defaults_repr
This is just for convenience with models that require dicts as input.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
586587588589590591592593594595596597
defargs_as_dict(self)->dict[str,Any]:"""Return the arguments as a Python dictionary. This is just for convenience with models that require dicts as input. """ifnotself.args:return{}ifisinstance(self.args,dict):returnself.argsargs=pydantic_core.from_json(self.args)assertisinstance(args,dict),'args should be a dict'returncast(dict[str,Any],args)
This is just for convenience with models that require JSON strings as input.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
599600601602603604605606607608
defargs_as_json_str(self)->str:"""Return the arguments as a JSON string. This is just for convenience with models that require JSON strings as input. """ifnotself.args:return'{}'ifisinstance(self.args,str):returnself.argsreturnpydantic_core.to_json(self.args).decode()
Source code in pydantic_ai_slim/pydantic_ai/messages.py
610611612613614615616617
defhas_content(self)->bool:"""Return `True` if the arguments contain any data."""ifisinstance(self.args,dict):# TODO: This should probably return True if you have the value False, or 0, etc.# It makes sense to me to ignore empty strings, but not sure about empty lists or dictsreturnany(self.args.values())else:returnbool(self.args)
@dataclass(repr=False)classModelResponse:"""A response from a model, e.g. a message from the model to the PydanticAI app."""parts:list[ModelResponsePart]"""The parts of the model message."""usage:Usage=field(default_factory=Usage)"""Usage information for the request. This has a default to make tests easier, and to support loading old messages where usage will be missing. """model_name:str|None=None"""The name of the model that generated the response."""timestamp:datetime=field(default_factory=_now_utc)"""The timestamp of the response. If the model provides a timestamp in the response (as OpenAI does) that will be used. """kind:Literal['response']='response'"""Message type identifier, this is available on all parts as a discriminator."""vendor_details:dict[str,Any]|None=field(default=None)"""Additional vendor-specific details in a serializable format. This allows storing selected vendor-specific data that isn't mapped to standard ModelResponse fields. For OpenAI models, this may include 'logprobs', 'finish_reason', etc. """vendor_id:str|None=None"""Vendor ID as specified by the model provider. This can be used to track the specific request to the model."""defotel_events(self)->list[Event]:"""Return OpenTelemetry events for the response."""result:list[Event]=[]defnew_event_body():new_body:dict[str,Any]={'role':'assistant'}ev=Event('gen_ai.assistant.message',body=new_body)result.append(ev)returnnew_bodybody=new_event_body()forpartinself.parts:ifisinstance(part,ToolCallPart):body.setdefault('tool_calls',[]).append({'id':part.tool_call_id,'type':'function',# TODO https://github.com/pydantic/pydantic-ai/issues/888'function':{'name':part.tool_name,'arguments':part.args,},})elifisinstance(part,TextPart):ifbody.get('content'):body=new_event_body()body['content']=part.contentreturnresult__repr__=_utils.dataclasses_no_defaults_repr
Additional vendor-specific details in a serializable format.
This allows storing selected vendor-specific data that isn't mapped to standard ModelResponse fields.
For OpenAI models, this may include 'logprobs', 'finish_reason', etc.
defotel_events(self)->list[Event]:"""Return OpenTelemetry events for the response."""result:list[Event]=[]defnew_event_body():new_body:dict[str,Any]={'role':'assistant'}ev=Event('gen_ai.assistant.message',body=new_body)result.append(ev)returnnew_bodybody=new_event_body()forpartinself.parts:ifisinstance(part,ToolCallPart):body.setdefault('tool_calls',[]).append({'id':part.tool_call_id,'type':'function',# TODO https://github.com/pydantic/pydantic-ai/issues/888'function':{'name':part.tool_name,'arguments':part.args,},})elifisinstance(part,TextPart):ifbody.get('content'):body=new_event_body()body['content']=part.contentreturnresult
@dataclass(repr=False)classTextPartDelta:"""A partial update (delta) for a `TextPart` to append new text content."""content_delta:str"""The incremental text content to add to the existing `TextPart` content."""part_delta_kind:Literal['text']='text'"""Part delta type identifier, used as a discriminator."""defapply(self,part:ModelResponsePart)->TextPart:"""Apply this text delta to an existing `TextPart`. Args: part: The existing model response part, which must be a `TextPart`. Returns: A new `TextPart` with updated text content. Raises: ValueError: If `part` is not a `TextPart`. """ifnotisinstance(part,TextPart):raiseValueError('Cannot apply TextPartDeltas to non-TextParts')# pragma: no coverreturnreplace(part,content=part.content+self.content_delta)__repr__=_utils.dataclasses_no_defaults_repr
Source code in pydantic_ai_slim/pydantic_ai/messages.py
713714715716717718719720721722723724725726727
defapply(self,part:ModelResponsePart)->TextPart:"""Apply this text delta to an existing `TextPart`. Args: part: The existing model response part, which must be a `TextPart`. Returns: A new `TextPart` with updated text content. Raises: ValueError: If `part` is not a `TextPart`. """ifnotisinstance(part,TextPart):raiseValueError('Cannot apply TextPartDeltas to non-TextParts')# pragma: no coverreturnreplace(part,content=part.content+self.content_delta)
ThinkingPartDeltadataclass
A partial update (delta) for a ThinkingPart to append new thinking content.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
@dataclass(repr=False)classThinkingPartDelta:"""A partial update (delta) for a `ThinkingPart` to append new thinking content."""content_delta:str|None=None"""The incremental thinking content to add to the existing `ThinkingPart` content."""signature_delta:str|None=None"""Optional signature delta. Note this is never treated as a delta — it can replace None. """part_delta_kind:Literal['thinking']='thinking'"""Part delta type identifier, used as a discriminator."""@overloaddefapply(self,part:ModelResponsePart)->ThinkingPart:...@overloaddefapply(self,part:ModelResponsePart|ThinkingPartDelta)->ThinkingPart|ThinkingPartDelta:...defapply(self,part:ModelResponsePart|ThinkingPartDelta)->ThinkingPart|ThinkingPartDelta:"""Apply this thinking delta to an existing `ThinkingPart`. Args: part: The existing model response part, which must be a `ThinkingPart`. Returns: A new `ThinkingPart` with updated thinking content. Raises: ValueError: If `part` is not a `ThinkingPart`. """ifisinstance(part,ThinkingPart):returnreplace(part,content=part.content+self.content_deltaifself.content_deltaelseNone)elifisinstance(part,ThinkingPartDelta):ifself.content_deltaisNoneandself.signature_deltaisNone:raiseValueError('Cannot apply ThinkingPartDelta with no content or signature')ifself.signature_deltaisnotNone:returnreplace(part,signature_delta=self.signature_delta)ifself.content_deltaisnotNone:returnreplace(part,content_delta=self.content_delta)raiseValueError(f'Cannot apply ThinkingPartDeltas to non-ThinkingParts or non-ThinkingPartDeltas ({part=}, {self=})')__repr__=_utils.dataclasses_no_defaults_repr
defapply(self,part:ModelResponsePart|ThinkingPartDelta)->ThinkingPart|ThinkingPartDelta:"""Apply this thinking delta to an existing `ThinkingPart`. Args: part: The existing model response part, which must be a `ThinkingPart`. Returns: A new `ThinkingPart` with updated thinking content. Raises: ValueError: If `part` is not a `ThinkingPart`. """ifisinstance(part,ThinkingPart):returnreplace(part,content=part.content+self.content_deltaifself.content_deltaelseNone)elifisinstance(part,ThinkingPartDelta):ifself.content_deltaisNoneandself.signature_deltaisNone:raiseValueError('Cannot apply ThinkingPartDelta with no content or signature')ifself.signature_deltaisnotNone:returnreplace(part,signature_delta=self.signature_delta)ifself.content_deltaisnotNone:returnreplace(part,content_delta=self.content_delta)raiseValueError(f'Cannot apply ThinkingPartDeltas to non-ThinkingParts or non-ThinkingPartDeltas ({part=}, {self=})')
ToolCallPartDeltadataclass
A partial update (delta) for a ToolCallPart to modify tool name, arguments, or tool call ID.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
@dataclass(repr=False)classToolCallPartDelta:"""A partial update (delta) for a `ToolCallPart` to modify tool name, arguments, or tool call ID."""tool_name_delta:str|None=None"""Incremental text to add to the existing tool name, if any."""args_delta:str|dict[str,Any]|None=None"""Incremental data to add to the tool arguments. If this is a string, it will be appended to existing JSON arguments. If this is a dict, it will be merged with existing dict arguments. """tool_call_id:str|None=None"""Optional tool call identifier, this is used by some models including OpenAI. Note this is never treated as a delta — it can replace None, but otherwise if a non-matching value is provided an error will be raised."""part_delta_kind:Literal['tool_call']='tool_call'"""Part delta type identifier, used as a discriminator."""defas_part(self)->ToolCallPart|None:"""Convert this delta to a fully formed `ToolCallPart` if possible, otherwise return `None`. Returns: A `ToolCallPart` if `tool_name_delta` is set, otherwise `None`. """ifself.tool_name_deltaisNone:returnNonereturnToolCallPart(self.tool_name_delta,self.args_delta,self.tool_call_idor_generate_tool_call_id())@overloaddefapply(self,part:ModelResponsePart)->ToolCallPart:...@overloaddefapply(self,part:ModelResponsePart|ToolCallPartDelta)->ToolCallPart|ToolCallPartDelta:...defapply(self,part:ModelResponsePart|ToolCallPartDelta)->ToolCallPart|ToolCallPartDelta:"""Apply this delta to a part or delta, returning a new part or delta with the changes applied. Args: part: The existing model response part or delta to update. Returns: Either a new `ToolCallPart` or an updated `ToolCallPartDelta`. Raises: ValueError: If `part` is neither a `ToolCallPart` nor a `ToolCallPartDelta`. UnexpectedModelBehavior: If applying JSON deltas to dict arguments or vice versa. """ifisinstance(part,ToolCallPart):returnself._apply_to_part(part)ifisinstance(part,ToolCallPartDelta):returnself._apply_to_delta(part)raiseValueError(# pragma: no coverf'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}')def_apply_to_delta(self,delta:ToolCallPartDelta)->ToolCallPart|ToolCallPartDelta:"""Internal helper to apply this delta to another delta."""ifself.tool_name_delta:# Append incremental text to the existing tool_name_deltaupdated_tool_name_delta=(delta.tool_name_deltaor'')+self.tool_name_deltadelta=replace(delta,tool_name_delta=updated_tool_name_delta)ifisinstance(self.args_delta,str):ifisinstance(delta.args_delta,dict):raiseUnexpectedModelBehavior(f'Cannot apply JSON deltas to non-JSON tool arguments ({delta=}, {self=})')updated_args_delta=(delta.args_deltaor'')+self.args_deltadelta=replace(delta,args_delta=updated_args_delta)elifisinstance(self.args_delta,dict):ifisinstance(delta.args_delta,str):raiseUnexpectedModelBehavior(f'Cannot apply dict deltas to non-dict tool arguments ({delta=}, {self=})')updated_args_delta={**(delta.args_deltaor{}),**self.args_delta}delta=replace(delta,args_delta=updated_args_delta)ifself.tool_call_id:delta=replace(delta,tool_call_id=self.tool_call_id)# If we now have enough data to create a full ToolCallPart, do soifdelta.tool_name_deltaisnotNone:returnToolCallPart(delta.tool_name_delta,delta.args_delta,delta.tool_call_idor_generate_tool_call_id())returndeltadef_apply_to_part(self,part:ToolCallPart)->ToolCallPart:"""Internal helper to apply this delta directly to a `ToolCallPart`."""ifself.tool_name_delta:# Append incremental text to the existing tool_nametool_name=part.tool_name+self.tool_name_deltapart=replace(part,tool_name=tool_name)ifisinstance(self.args_delta,str):ifisinstance(part.args,dict):raiseUnexpectedModelBehavior(f'Cannot apply JSON deltas to non-JSON tool arguments ({part=}, {self=})')updated_json=(part.argsor'')+self.args_deltapart=replace(part,args=updated_json)elifisinstance(self.args_delta,dict):ifisinstance(part.args,str):raiseUnexpectedModelBehavior(f'Cannot apply dict deltas to non-dict tool arguments ({part=}, {self=})')updated_dict={**(part.argsor{}),**self.args_delta}part=replace(part,args=updated_dict)ifself.tool_call_id:part=replace(part,tool_call_id=self.tool_call_id)returnpart__repr__=_utils.dataclasses_no_defaults_repr
A ToolCallPart if tool_name_delta is set, otherwise None.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
805806807808809810811812813814
defas_part(self)->ToolCallPart|None:"""Convert this delta to a fully formed `ToolCallPart` if possible, otherwise return `None`. Returns: A `ToolCallPart` if `tool_name_delta` is set, otherwise `None`. """ifself.tool_name_deltaisNone:returnNonereturnToolCallPart(self.tool_name_delta,self.args_delta,self.tool_call_idor_generate_tool_call_id())
defapply(self,part:ModelResponsePart|ToolCallPartDelta)->ToolCallPart|ToolCallPartDelta:"""Apply this delta to a part or delta, returning a new part or delta with the changes applied. Args: part: The existing model response part or delta to update. Returns: Either a new `ToolCallPart` or an updated `ToolCallPartDelta`. Raises: ValueError: If `part` is neither a `ToolCallPart` nor a `ToolCallPartDelta`. UnexpectedModelBehavior: If applying JSON deltas to dict arguments or vice versa. """ifisinstance(part,ToolCallPart):returnself._apply_to_part(part)ifisinstance(part,ToolCallPartDelta):returnself._apply_to_delta(part)raiseValueError(# pragma: no coverf'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}')
@dataclass(repr=False)classPartStartEvent:"""An event indicating that a new part has started. If multiple `PartStartEvent`s are received with the same index, the new one should fully replace the old one. """index:int"""The index of the part within the overall response parts list."""part:ModelResponsePart"""The newly started `ModelResponsePart`."""event_kind:Literal['part_start']='part_start'"""Event type identifier, used as a discriminator."""__repr__=_utils.dataclasses_no_defaults_repr
An event indicating a delta update for an existing part.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
927928929930931932933934935936937938939940
@dataclass(repr=False)classPartDeltaEvent:"""An event indicating a delta update for an existing part."""index:int"""The index of the part within the overall response parts list."""delta:ModelResponsePartDelta"""The delta to apply to the specified part."""event_kind:Literal['part_delta']='part_delta'"""Event type identifier, used as a discriminator."""__repr__=_utils.dataclasses_no_defaults_repr
An event indicating the response to the current model request matches the output schema and will produce a result.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
943944945946947948949950951952953954
@dataclass(repr=False)classFinalResultEvent:"""An event indicating the response to the current model request matches the output schema and will produce a result."""tool_name:str|None"""The name of the output tool that was called. `None` if the result is from text content and not from a tool."""tool_call_id:str|None"""The tool call ID, if any, that this result is associated with."""event_kind:Literal['final_result']='final_result'"""Event type identifier, used as a discriminator."""__repr__=_utils.dataclasses_no_defaults_repr
An event indicating the start to a call to a function tool.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
966967968969970971972973974975976977978979980
@dataclass(repr=False)classFunctionToolCallEvent:"""An event indicating the start to a call to a function tool."""part:ToolCallPart"""The (function) tool call to make."""call_id:str=field(init=False)"""An ID used for matching details about the call to its result. If present, defaults to the part's tool_call_id."""event_kind:Literal['function_tool_call']='function_tool_call'"""Event type identifier, used as a discriminator."""def__post_init__(self):self.call_id=self.part.tool_call_idorstr(uuid.uuid4())__repr__=_utils.dataclasses_no_defaults_repr
An event indicating the result of a function tool call.
Source code in pydantic_ai_slim/pydantic_ai/messages.py
983984985986987988989990991992993994
@dataclass(repr=False)classFunctionToolResultEvent:"""An event indicating the result of a function tool call."""result:ToolReturnPart|RetryPromptPart"""The result of the call to the function tool."""tool_call_id:str"""An ID used to match the result to its original call."""event_kind:Literal['function_tool_result']='function_tool_result'"""Event type identifier, used as a discriminator."""__repr__=_utils.dataclasses_no_defaults_repr