[docs]classSearchArgsSchema(BaseModel):"""Input for GetMessageTool."""message_id:str=Field(...,description="The unique ID of the email message, retrieved from a search.",)
[docs]classGmailGetMessage(GmailBaseTool):# type: ignore[override, override]"""Tool that gets a message by ID from Gmail."""name:str="get_gmail_message"description:str=("Use this tool to fetch an email by message ID."" Returns the thread ID, snippet, body, subject, and sender.")args_schema:Type[SearchArgsSchema]=SearchArgsSchemadef_run(self,message_id:str,run_manager:Optional[CallbackManagerForToolRun]=None,)->Dict:"""Run the tool."""query=(self.api_resource.users().messages().get(userId="me",format="raw",id=message_id))message_data=query.execute()raw_message=base64.urlsafe_b64decode(message_data["raw"])email_msg=email.message_from_bytes(raw_message)subject=email_msg["Subject"]sender=email_msg["From"]message_body=""ifemail_msg.is_multipart():forpartinemail_msg.walk():ctype=part.get_content_type()cdispo=str(part.get("Content-Disposition"))ifctype=="text/plain"and"attachment"notincdispo:message_body=part.get_payload(decode=True).decode("utf-8")# type: ignore[union-attr]breakelse:message_body=email_msg.get_payload(decode=True).decode("utf-8")# type: ignore[union-attr]body=clean_email_body(message_body)return{"id":message_id,"threadId":message_data["threadId"],"snippet":message_data["snippet"],"body":body,"subject":subject,"sender":sender,}