feat(backend): add getting user profile, drafts, update send email to use mulitple to, cc, bcc (#10482)

Need: The Gmail integration had several parsing issues that were causing
data loss and
  workflow incompatibilities:
1. Email recipient parsing only captured the first recipient, losing
CC/BCC and multiple TO
   recipients
2. Email body parsing was inconsistent between blocks, sometimes showing
"This email does
  not contain a readable body" for valid emails
3. Type mismatches between blocks caused serialization issues when
connecting them in
workflows (lists being converted to string representations like
"[\"email@example.com\"]")

  # Changes 🏗️

  1. Enhanced Email Model:
    - Added cc and bcc fields to capture all recipients
    - Changed to field from string to list for consistency
    - Now captures all recipients instead of just the first one
  2. Improved Email Parsing:
- Updated GmailReadBlock and GmailGetThreadBlock to parse all recipients
using
  getaddresses()
- Unified email body parsing logic across blocks with robust multipart
handling
    - Added support for HTML to plain text conversion
    - Fixed handling of emails with attachments as body content
  3. Fixed Block Compatibility:
- Updated GmailSendBlock and GmailCreateDraftBlock to accept lists for
recipient fields
    - Added validation to ensure at least one recipient is provided
- All blocks now consistently use lists for recipient fields, preventing
serialization
  issues
  4. Updated Test Data:
- Modified all test inputs/outputs to use the new list format for
recipients
    - Ensures tests reflect the new data structure

  # Checklist 📋

  For code changes:

  - I have clearly listed my changes in the PR description
  - I have made a test plan
  - I have tested my changes according to the test plan:
    - Run existing Gmail block unit tests with poetry run test
- Create a workflow that reads emails with multiple recipients and
verify all TO, CC, BCC
   recipients are captured
- Test email body parsing with plain text, HTML, and multipart emails
- Connect GmailReadBlock → GmailSendBlock in a workflow and verify
recipient data flows
  correctly
- Connect GmailReplyBlock → GmailSendBlock and verify no serialization
errors occur
    - Test sending emails with multiple recipients via GmailSendBlock
- Test creating drafts with multiple recipients via
GmailCreateDraftBlock
- Verify backwards compatibility by testing with single recipient
strings (should now
  require lists)

  - Create from scratch and execute an agent with at least 3 blocks
  - Import an agent from file upload, and confirm it executes correctly
  - Upload agent to marketplace
  - Import an agent from marketplace and confirm it executes correctly
  - Edit an agent from monitor, and confirm it executes correctly

# Breaking Change Note: The to field in GmailSendBlock and
GmailCreateDraftBlock now requires
a list instead of accepting both string and list. Existing workflows
using strings will
need to be updated to use lists (e.g., ["email@example.com"] instead of
  "email@example.com").

---------

Co-authored-by: Zamil Majdy <zamil.majdy@agpt.co>

autogpt_platform/backend/backend/blocks/google/gmail.py

@@ -1,4 +1,3 @@
-import asyncio
 import base64
 from email.utils import getaddresses, parseaddr
 from pathlib import Path
@@ -6,7 +5,7 @@ from typing import List
 
 from google.oauth2.credentials import Credentials
 from googleapiclient.discovery import build
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
 from backend.data.model import SchemaField
@@ -23,6 +22,56 @@ from ._auth import (
 )
 
 
+def serialize_email_recipients(recipients: list[str]) -> str:
+    """Serialize recipients list to comma-separated string."""
+    return ", ".join(recipients)
+
+
+async def create_mime_message(
+    input_data,
+    graph_exec_id: str,
+    user_id: str,
+) -> str:
+    """Create a MIME message with attachments and return base64-encoded raw message."""
+    from email import encoders
+    from email.mime.base import MIMEBase
+    from email.mime.multipart import MIMEMultipart
+    from email.mime.text import MIMEText
+
+    message = MIMEMultipart()
+    message["to"] = serialize_email_recipients(input_data.to)
+    message["subject"] = input_data.subject
+
+    if input_data.cc:
+        message["cc"] = ", ".join(input_data.cc)
+    if input_data.bcc:
+        message["bcc"] = ", ".join(input_data.bcc)
+
+    message.attach(MIMEText(input_data.body))
+
+    # Handle attachments if any
+    if input_data.attachments:
+        for attach in input_data.attachments:
+            local_path = await store_media_file(
+                user_id=user_id,
+                graph_exec_id=graph_exec_id,
+                file=attach,
+                return_content=False,
+            )
+            abs_path = get_exec_file_path(graph_exec_id, local_path)
+            part = MIMEBase("application", "octet-stream")
+            with open(abs_path, "rb") as f:
+                part.set_payload(f.read())
+            encoders.encode_base64(part)
+            part.add_header(
+                "Content-Disposition",
+                f"attachment; filename={Path(abs_path).name}",
+            )
+            message.attach(part)
+
+    return base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")
+
+
 class Attachment(BaseModel):
     filename: str
     content_type: str
@@ -37,7 +86,11 @@ class Email(BaseModel):
     subject: str
     snippet: str
     from_: str
-    to: str
+    to: list[str]  # List of recipient email addresses
+    cc: list[str] = Field(default_factory=list)  # CC recipients
+    bcc: list[str] = Field(
+        default_factory=list
+    )  # BCC recipients (rarely available in received emails)
     date: str
     body: str = ""  # Default to an empty string
     sizeEstimate: int
@@ -55,11 +108,24 @@ class GmailSendResult(BaseModel):
     status: str
 
 
+class GmailDraftResult(BaseModel):
+    id: str
+    message_id: str
+    status: str
+
+
 class GmailLabelResult(BaseModel):
     label_id: str
     status: str
 
 
+class Profile(BaseModel):
+    emailAddress: str
+    messagesTotal: int
+    threadsTotal: int
+    historyId: str
+
+
 class GmailReadBlock(Block):
     class Input(BlockSchema):
         credentials: GoogleCredentialsInput = GoogleCredentialsField(
@@ -109,7 +175,9 @@ class GmailReadBlock(Block):
                         "subject": "Test Email",
                         "snippet": "This is a test email",
                         "from_": "test@example.com",
-                        "to": "recipient@example.com",
+                        "to": ["recipient@example.com"],
+                        "cc": [],
+                        "bcc": [],
                         "date": "2024-01-01",
                         "body": "This is a test email",
                         "sizeEstimate": 100,
@@ -126,7 +194,9 @@ class GmailReadBlock(Block):
                             "subject": "Test Email",
                             "snippet": "This is a test email",
                             "from_": "test@example.com",
-                            "to": "recipient@example.com",
+                            "to": ["recipient@example.com"],
+                            "cc": [],
+                            "bcc": [],
                             "date": "2024-01-01",
                             "body": "This is a test email",
                             "sizeEstimate": 100,
@@ -144,7 +214,9 @@ class GmailReadBlock(Block):
                         "subject": "Test Email",
                         "snippet": "This is a test email",
                         "from_": "test@example.com",
-                        "to": "recipient@example.com",
+                        "to": ["recipient@example.com"],
+                        "cc": [],
+                        "bcc": [],
                         "date": "2024-01-01",
                         "body": "This is a test email",
                         "sizeEstimate": 100,
@@ -159,8 +231,7 @@ class GmailReadBlock(Block):
         self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
     ) -> BlockOutput:
         service = GmailReadBlock._build_service(credentials, **kwargs)
-        messages = await asyncio.to_thread(
-            self._read_emails,
+        messages = await self._read_emails(
             service,
             input_data.query,
             input_data.max_results,
@@ -190,7 +261,7 @@ class GmailReadBlock(Block):
         )
         return build("gmail", "v1", credentials=creds)
 
-    def _read_emails(
+    async def _read_emails(
         self,
         service,
         query: str | None,
@@ -203,6 +274,7 @@ class GmailReadBlock(Block):
             list_kwargs["q"] = query
 
         results = service.users().messages().list(**list_kwargs).execute()
+
         messages = results.get("messages", [])
 
         email_data = []
@@ -224,7 +296,18 @@ class GmailReadBlock(Block):
                 for header in msg["payload"]["headers"]
             }
 
-            attachments = self._get_attachments(service, msg)
+            attachments = await self._get_attachments(service, msg)
+
+            # Parse all recipients
+            to_recipients = [
+                addr.strip() for _, addr in getaddresses([headers.get("to", "")])
+            ]
+            cc_recipients = [
+                addr.strip() for _, addr in getaddresses([headers.get("cc", "")])
+            ]
+            bcc_recipients = [
+                addr.strip() for _, addr in getaddresses([headers.get("bcc", "")])
+            ]
 
             email = Email(
                 threadId=msg["threadId"],
@@ -233,9 +316,11 @@ class GmailReadBlock(Block):
                 subject=headers.get("subject", "No Subject"),
                 snippet=msg["snippet"],
                 from_=parseaddr(headers.get("from", ""))[1],
-                to=parseaddr(headers.get("to", ""))[1],
+                to=to_recipients if to_recipients else [],
+                cc=cc_recipients,
+                bcc=bcc_recipients,
                 date=headers.get("date", ""),
-                body=self._get_email_body(msg, service),
+                body=await self._get_email_body(msg, service),
                 sizeEstimate=msg["sizeEstimate"],
                 attachments=attachments,
             )
@@ -243,12 +328,12 @@ class GmailReadBlock(Block):
 
         return email_data
 
-    def _get_email_body(self, msg, service):
+    async def _get_email_body(self, msg, service):
         """Extract email body content with support for multipart messages and HTML conversion."""
-        text = self._walk_for_body(msg["payload"], msg["id"], service)
+        text = await self._walk_for_body(msg["payload"], msg["id"], service)
         return text or "This email does not contain a readable body."
 
-    def _walk_for_body(self, part, msg_id, service, depth=0):
+    async def _walk_for_body(self, part, msg_id, service, depth=0):
         """Recursively walk through email parts to find readable body content."""
         # Prevent infinite recursion by limiting depth
         if depth > 10:
@@ -278,7 +363,7 @@ class GmailReadBlock(Block):
 
         # Handle content stored as attachment
         if body.get("attachmentId"):
-            attachment_data = self._download_attachment_body(
+            attachment_data = await self._download_attachment_body(
                 body["attachmentId"], msg_id, service
             )
             if attachment_data:
@@ -286,7 +371,7 @@ class GmailReadBlock(Block):
 
         # Recursively search in parts
         for sub_part in part.get("parts", []):
-            text = self._walk_for_body(sub_part, msg_id, service, depth + 1)
+            text = await self._walk_for_body(sub_part, msg_id, service, depth + 1)
             if text:
                 return text
 
@@ -305,7 +390,7 @@ class GmailReadBlock(Block):
         except Exception:
             return None
 
-    def _download_attachment_body(self, attachment_id, msg_id, service):
+    async def _download_attachment_body(self, attachment_id, msg_id, service):
         """Download attachment content when email body is stored as attachment."""
         try:
             attachment = (
@@ -319,7 +404,7 @@ class GmailReadBlock(Block):
         except Exception:
             return None
 
-    def _get_attachments(self, service, message):
+    async def _get_attachments(self, service, message):
         attachments = []
         if "parts" in message["payload"]:
             for part in message["payload"]["parts"]:
@@ -351,8 +436,8 @@ class GmailSendBlock(Block):
         credentials: GoogleCredentialsInput = GoogleCredentialsField(
             ["https://www.googleapis.com/auth/gmail.send"]
         )
-        to: str = SchemaField(
-            description="Recipient email address",
+        to: list[str] = SchemaField(
+            description="Recipient email addresses",
         )
         subject: str = SchemaField(
             description="Email subject",
@@ -360,6 +445,11 @@ class GmailSendBlock(Block):
         body: str = SchemaField(
             description="Email body",
         )
+        cc: list[str] = SchemaField(description="CC recipients", default_factory=list)
+        bcc: list[str] = SchemaField(description="BCC recipients", default_factory=list)
+        attachments: list[MediaFileType] = SchemaField(
+            description="Files to attach", default_factory=list, advanced=True
+        )
 
     class Output(BlockSchema):
         result: GmailSendResult = SchemaField(
@@ -378,7 +468,7 @@ class GmailSendBlock(Block):
             output_schema=GmailSendBlock.Output,
             disabled=not GOOGLE_OAUTH_IS_CONFIGURED,
             test_input={
-                "to": "recipient@example.com",
+                "to": ["recipient@example.com"],
                 "subject": "Test Email",
                 "body": "This is a test email sent from GmailSendBlock.",
                 "credentials": TEST_CREDENTIALS_INPUT,
@@ -393,35 +483,136 @@ class GmailSendBlock(Block):
         )
 
     async def run(
-        self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
+        self,
+        input_data: Input,
+        *,
+        credentials: GoogleCredentials,
+        graph_exec_id: str,
+        user_id: str,
+        **kwargs,
     ) -> BlockOutput:
         service = GmailReadBlock._build_service(credentials, **kwargs)
-        result = await asyncio.to_thread(
-            self._send_email,
+        result = await self._send_email(
             service,
-            input_data.to,
-            input_data.subject,
-            input_data.body,
+            input_data,
+            graph_exec_id,
+            user_id,
         )
         yield "result", result
 
-    def _send_email(self, service, to: str, subject: str, body: str) -> dict:
-        if not to or not subject or not body:
-            raise ValueError("To, subject, and body are required for sending an email")
-        message = self._create_message(to, subject, body)
+    async def _send_email(
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
+    ) -> dict:
+        if not input_data.to or not input_data.subject or not input_data.body:
+            raise ValueError(
+                "At least one recipient, subject, and body are required for sending an email"
+            )
+        raw_message = await create_mime_message(input_data, graph_exec_id, user_id)
         sent_message = (
-            service.users().messages().send(userId="me", body=message).execute()
+            service.users()
+            .messages()
+            .send(userId="me", body={"raw": raw_message})
+            .execute()
         )
         return {"id": sent_message["id"], "status": "sent"}
 
-    def _create_message(self, to: str, subject: str, body: str) -> dict:
-        from email.mime.text import MIMEText
 
-        message = MIMEText(body)
-        message["to"] = to
-        message["subject"] = subject
-        raw_message = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")
-        return {"raw": raw_message}
+class GmailCreateDraftBlock(Block):
+    class Input(BlockSchema):
+        credentials: GoogleCredentialsInput = GoogleCredentialsField(
+            ["https://www.googleapis.com/auth/gmail.modify"]
+        )
+        to: list[str] = SchemaField(
+            description="Recipient email addresses",
+        )
+        subject: str = SchemaField(
+            description="Email subject",
+        )
+        body: str = SchemaField(
+            description="Email body",
+        )
+        cc: list[str] = SchemaField(description="CC recipients", default_factory=list)
+        bcc: list[str] = SchemaField(description="BCC recipients", default_factory=list)
+        attachments: list[MediaFileType] = SchemaField(
+            description="Files to attach", default_factory=list, advanced=True
+        )
+
+    class Output(BlockSchema):
+        result: GmailDraftResult = SchemaField(
+            description="Draft creation result",
+        )
+        error: str = SchemaField(
+            description="Error message if any",
+        )
+
+    def __init__(self):
+        super().__init__(
+            id="e1eeead4-46cb-491e-8281-17b6b9c44a55",
+            description="This block creates a draft email in Gmail.",
+            categories={BlockCategory.COMMUNICATION},
+            input_schema=GmailCreateDraftBlock.Input,
+            output_schema=GmailCreateDraftBlock.Output,
+            disabled=not GOOGLE_OAUTH_IS_CONFIGURED,
+            test_input={
+                "to": ["recipient@example.com"],
+                "subject": "Draft Test Email",
+                "body": "This is a test draft email.",
+                "credentials": TEST_CREDENTIALS_INPUT,
+            },
+            test_credentials=TEST_CREDENTIALS,
+            test_output=[
+                (
+                    "result",
+                    GmailDraftResult(
+                        id="draft1", message_id="msg1", status="draft_created"
+                    ),
+                ),
+            ],
+            test_mock={
+                "_create_draft": lambda *args, **kwargs: {
+                    "id": "draft1",
+                    "message": {"id": "msg1"},
+                },
+            },
+        )
+
+    async def run(
+        self,
+        input_data: Input,
+        *,
+        credentials: GoogleCredentials,
+        graph_exec_id: str,
+        user_id: str,
+        **kwargs,
+    ) -> BlockOutput:
+        service = GmailReadBlock._build_service(credentials, **kwargs)
+        result = await self._create_draft(
+            service,
+            input_data,
+            graph_exec_id,
+            user_id,
+        )
+        yield "result", GmailDraftResult(
+            id=result["id"], message_id=result["message"]["id"], status="draft_created"
+        )
+
+    async def _create_draft(
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
+    ) -> dict:
+        if not input_data.to or not input_data.subject:
+            raise ValueError(
+                "At least one recipient and subject are required for creating a draft"
+            )
+
+        raw_message = await create_mime_message(input_data, graph_exec_id, user_id)
+        draft = (
+            service.users()
+            .drafts()
+            .create(userId="me", body={"message": {"raw": raw_message}})
+            .execute()
+        )
+
+        return draft
 
 
 class GmailListLabelsBlock(Block):
@@ -471,10 +662,10 @@ class GmailListLabelsBlock(Block):
         self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
     ) -> BlockOutput:
         service = GmailReadBlock._build_service(credentials, **kwargs)
-        result = await asyncio.to_thread(self._list_labels, service)
+        result = await self._list_labels(service)
         yield "result", result
 
-    def _list_labels(self, service) -> list[dict]:
+    async def _list_labels(self, service) -> list[dict]:
         results = service.users().labels().list(userId="me").execute()
         labels = results.get("labels", [])
         return [{"id": label["id"], "name": label["name"]} for label in labels]
@@ -532,20 +723,29 @@ class GmailAddLabelBlock(Block):
         self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
     ) -> BlockOutput:
         service = GmailReadBlock._build_service(credentials, **kwargs)
-        result = await asyncio.to_thread(
-            self._add_label, service, input_data.message_id, input_data.label_name
+        result = await self._add_label(
+            service, input_data.message_id, input_data.label_name
         )
         yield "result", result
 
-    def _add_label(self, service, message_id: str, label_name: str) -> dict:
-        label_id = self._get_or_create_label(service, label_name)
-        service.users().messages().modify(
-            userId="me", id=message_id, body={"addLabelIds": [label_id]}
-        ).execute()
+    async def _add_label(self, service, message_id: str, label_name: str) -> dict:
+        label_id = await self._get_or_create_label(service, label_name)
+        result = (
+            service.users()
+            .messages()
+            .modify(userId="me", id=message_id, body={"addLabelIds": [label_id]})
+            .execute()
+        )
+        if not result.get("labelIds"):
+            return {
+                "status": "Label already applied or not found",
+                "label_id": label_id,
+            }
+
         return {"status": "Label added successfully", "label_id": label_id}
 
-    def _get_or_create_label(self, service, label_name: str) -> str:
-        label_id = self._get_label_id(service, label_name)
+    async def _get_or_create_label(self, service, label_name: str) -> str:
+        label_id = await self._get_label_id(service, label_name)
         if not label_id:
             label = (
                 service.users()
@@ -556,7 +756,7 @@ class GmailAddLabelBlock(Block):
             label_id = label["id"]
         return label_id
 
-    def _get_label_id(self, service, label_name: str) -> str | None:
+    async def _get_label_id(self, service, label_name: str) -> str | None:
         results = service.users().labels().list(userId="me").execute()
         labels = results.get("labels", [])
         for label in labels:
@@ -617,22 +817,30 @@ class GmailRemoveLabelBlock(Block):
         self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
     ) -> BlockOutput:
         service = GmailReadBlock._build_service(credentials, **kwargs)
-        result = await asyncio.to_thread(
-            self._remove_label, service, input_data.message_id, input_data.label_name
+        result = await self._remove_label(
+            service, input_data.message_id, input_data.label_name
         )
         yield "result", result
 
-    def _remove_label(self, service, message_id: str, label_name: str) -> dict:
-        label_id = self._get_label_id(service, label_name)
+    async def _remove_label(self, service, message_id: str, label_name: str) -> dict:
+        label_id = await self._get_label_id(service, label_name)
         if label_id:
-            service.users().messages().modify(
-                userId="me", id=message_id, body={"removeLabelIds": [label_id]}
-            ).execute()
+            result = (
+                service.users()
+                .messages()
+                .modify(userId="me", id=message_id, body={"removeLabelIds": [label_id]})
+                .execute()
+            )
+            if not result.get("labelIds"):
+                return {
+                    "status": "Label already removed or not applied",
+                    "label_id": label_id,
+                }
             return {"status": "Label removed successfully", "label_id": label_id}
         else:
             return {"status": "Label not found", "label_name": label_name}
 
-    def _get_label_id(self, service, label_name: str) -> str | None:
+    async def _get_label_id(self, service, label_name: str) -> str | None:
         results = service.users().labels().list(userId="me").execute()
         labels = results.get("labels", [])
         for label in labels:
@@ -672,7 +880,9 @@ class GmailGetThreadBlock(Block):
                         "messages": [
                             {
                                 "id": "188199feff9dc907",
-                                "to": "nick@example.co",
+                                "to": ["nick@example.co"],
+                                "cc": [],
+                                "bcc": [],
                                 "body": "This email does not contain a text body.",
                                 "date": "Thu, 17 Jul 2025 19:22:36 +0100",
                                 "from_": "bent@example.co",
@@ -701,7 +911,9 @@ class GmailGetThreadBlock(Block):
                     "messages": [
                         {
                             "id": "188199feff9dc907",
-                            "to": "nick@example.co",
+                            "to": ["nick@example.co"],
+                            "cc": [],
+                            "bcc": [],
                             "body": "This email does not contain a text body.",
                             "date": "Thu, 17 Jul 2025 19:22:36 +0100",
                             "from_": "bent@example.co",
@@ -729,10 +941,14 @@ class GmailGetThreadBlock(Block):
         self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
     ) -> BlockOutput:
         service = GmailReadBlock._build_service(credentials, **kwargs)
-        thread = self._get_thread(service, input_data.threadId, credentials.scopes)
+        thread = await self._get_thread(
+            service, input_data.threadId, credentials.scopes
+        )
         yield "thread", thread
 
-    def _get_thread(self, service, thread_id: str, scopes: list[str] | None) -> Thread:
+    async def _get_thread(
+        self, service, thread_id: str, scopes: list[str] | None
+    ) -> Thread:
         scopes = [s.lower() for s in (scopes or [])]
         format_type = (
             "metadata"
@@ -752,8 +968,20 @@ class GmailGetThreadBlock(Block):
                 h["name"].lower(): h["value"]
                 for h in msg.get("payload", {}).get("headers", [])
             }
-            body = self._get_email_body(msg)
-            attachments = self._get_attachments(service, msg)
+            body = await self._get_email_body(msg, service)
+            attachments = await self._get_attachments(service, msg)
+
+            # Parse all recipients
+            to_recipients = [
+                addr.strip() for _, addr in getaddresses([headers.get("to", "")])
+            ]
+            cc_recipients = [
+                addr.strip() for _, addr in getaddresses([headers.get("cc", "")])
+            ]
+            bcc_recipients = [
+                addr.strip() for _, addr in getaddresses([headers.get("bcc", "")])
+            ]
+
             email = Email(
                 threadId=msg.get("threadId", thread_id),
                 labelIds=msg.get("labelIds", []),
@@ -761,7 +989,9 @@ class GmailGetThreadBlock(Block):
                 subject=headers.get("subject", "No Subject"),
                 snippet=msg.get("snippet", ""),
                 from_=parseaddr(headers.get("from", ""))[1],
-                to=parseaddr(headers.get("to", ""))[1],
+                to=to_recipients if to_recipients else [],
+                cc=cc_recipients,
+                bcc=bcc_recipients,
                 date=headers.get("date", ""),
                 body=body,
                 sizeEstimate=msg.get("sizeEstimate", 0),
@@ -772,26 +1002,83 @@ class GmailGetThreadBlock(Block):
         thread["messages"] = parsed_messages
         return thread
 
-    def _get_email_body(self, msg):
-        payload = msg.get("payload")
-        if not payload:
-            return "This email does not contain a text body."
+    async def _get_email_body(self, msg, service):
+        """Extract email body content with support for multipart messages and HTML conversion."""
+        text = await self._walk_for_body(msg["payload"], msg["id"], service)
+        return text or "This email does not contain a readable body."
 
-        if "parts" in payload:
-            for part in payload["parts"]:
-                if part.get("mimeType") == "text/plain" and "data" in part.get(
-                    "body", {}
-                ):
-                    return base64.urlsafe_b64decode(part["body"]["data"]).decode(
-                        "utf-8"
-                    )
-        elif payload.get("mimeType") == "text/plain" and "data" in payload.get(
-            "body", {}
-        ):
-            return base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8")
-        return "This email does not contain a text body."
+    async def _walk_for_body(self, part, msg_id, service, depth=0):
+        """Recursively walk through email parts to find readable body content."""
+        # Prevent infinite recursion by limiting depth
+        if depth > 10:
+            return None
 
-    def _get_attachments(self, service, message):
+        mime_type = part.get("mimeType", "")
+        body = part.get("body", {})
+
+        # Handle text/plain content
+        if mime_type == "text/plain" and body.get("data"):
+            return self._decode_base64(body["data"])
+
+        # Handle text/html content (convert to plain text)
+        if mime_type == "text/html" and body.get("data"):
+            html_content = self._decode_base64(body["data"])
+            if html_content:
+                try:
+                    import html2text
+
+                    h = html2text.HTML2Text()
+                    h.ignore_links = False
+                    h.ignore_images = True
+                    return h.handle(html_content)
+                except ImportError:
+                    # Fallback: return raw HTML if html2text is not available
+                    return html_content
+
+        # Handle content stored as attachment
+        if body.get("attachmentId"):
+            attachment_data = await self._download_attachment_body(
+                body["attachmentId"], msg_id, service
+            )
+            if attachment_data:
+                return self._decode_base64(attachment_data)
+
+        # Recursively search in parts
+        for sub_part in part.get("parts", []):
+            text = await self._walk_for_body(sub_part, msg_id, service, depth + 1)
+            if text:
+                return text
+
+        return None
+
+    def _decode_base64(self, data):
+        """Safely decode base64 URL-safe data with proper padding."""
+        if not data:
+            return None
+        try:
+            # Add padding if necessary
+            missing_padding = len(data) % 4
+            if missing_padding:
+                data += "=" * (4 - missing_padding)
+            return base64.urlsafe_b64decode(data).decode("utf-8")
+        except Exception:
+            return None
+
+    async def _download_attachment_body(self, attachment_id, msg_id, service):
+        """Download attachment content when email body is stored as attachment."""
+        try:
+            attachment = (
+                service.users()
+                .messages()
+                .attachments()
+                .get(userId="me", messageId=msg_id, id=attachment_id)
+                .execute()
+            )
+            return attachment.get("data")
+        except Exception:
+            return None
+
+    async def _get_attachments(self, service, message):
         attachments = []
         if "parts" in message["payload"]:
             for part in message["payload"]["parts"]:
@@ -831,6 +1118,9 @@ class GmailReplyBlock(Block):
         messageId: str = SchemaField(description="Sent message ID")
         threadId: str = SchemaField(description="Thread ID")
         message: dict = SchemaField(description="Raw Gmail message object")
+        email: Email = SchemaField(
+            description="Parsed email object with decoded body and attachments"
+        )
         error: str = SchemaField(description="Error message if any")
 
     def __init__(self):
@@ -853,6 +1143,24 @@ class GmailReplyBlock(Block):
                 ("messageId", "m2"),
                 ("threadId", "t1"),
                 ("message", {"id": "m2", "threadId": "t1"}),
+                (
+                    "email",
+                    Email(
+                        threadId="t1",
+                        labelIds=[],
+                        id="m2",
+                        subject="",
+                        snippet="",
+                        from_="",
+                        to=[],
+                        cc=[],
+                        bcc=[],
+                        date="",
+                        body="Thanks",
+                        sizeEstimate=0,
+                        attachments=[],
+                    ),
+                ),
             ],
             test_mock={
                 "_reply": lambda *args, **kwargs: {
@@ -881,6 +1189,22 @@ class GmailReplyBlock(Block):
         yield "messageId", message["id"]
         yield "threadId", message.get("threadId", input_data.threadId)
         yield "message", message
+        email = Email(
+            threadId=message.get("threadId", input_data.threadId),
+            labelIds=message.get("labelIds", []),
+            id=message["id"],
+            subject=input_data.subject or "",
+            snippet=message.get("snippet", ""),
+            from_="",  # From address would need to be retrieved from the message headers
+            to=input_data.to if input_data.to else [],
+            cc=input_data.cc if input_data.cc else [],
+            bcc=input_data.bcc if input_data.bcc else [],
+            date="",  # Date would need to be retrieved from the message headers
+            body=input_data.body,
+            sizeEstimate=message.get("sizeEstimate", 0),
+            attachments=[],  # Attachments info not available from send response
+        )
+        yield "email", email
 
     async def _reply(
         self, service, input_data: Input, graph_exec_id: str, user_id: str
@@ -904,6 +1228,7 @@ class GmailReplyBlock(Block):
             )
             .execute()
         )
+
         headers = {
             h["name"].lower(): h["value"]
             for h in parent.get("payload", {}).get("headers", [])
@@ -975,3 +1300,63 @@ class GmailReplyBlock(Block):
             .send(userId="me", body={"threadId": input_data.threadId, "raw": raw})
             .execute()
         )
+
+
+class GmailGetProfileBlock(Block):
+    class Input(BlockSchema):
+        credentials: GoogleCredentialsInput = GoogleCredentialsField(
+            ["https://www.googleapis.com/auth/gmail.readonly"]
+        )
+
+    class Output(BlockSchema):
+        profile: Profile = SchemaField(description="Gmail user profile information")
+        error: str = SchemaField(description="Error message if any")
+
+    def __init__(self):
+        super().__init__(
+            id="04b0d996-0908-4a4b-89dd-b9697ff253d3",
+            description="Get the authenticated user's Gmail profile details including email address and message statistics.",
+            categories={BlockCategory.COMMUNICATION},
+            disabled=not GOOGLE_OAUTH_IS_CONFIGURED,
+            input_schema=GmailGetProfileBlock.Input,
+            output_schema=GmailGetProfileBlock.Output,
+            test_input={
+                "credentials": TEST_CREDENTIALS_INPUT,
+            },
+            test_credentials=TEST_CREDENTIALS,
+            test_output=[
+                (
+                    "profile",
+                    {
+                        "emailAddress": "test@example.com",
+                        "messagesTotal": 1000,
+                        "threadsTotal": 500,
+                        "historyId": "12345",
+                    },
+                ),
+            ],
+            test_mock={
+                "_get_profile": lambda *args, **kwargs: {
+                    "emailAddress": "test@example.com",
+                    "messagesTotal": 1000,
+                    "threadsTotal": 500,
+                    "historyId": "12345",
+                },
+            },
+        )
+
+    async def run(
+        self, input_data: Input, *, credentials: GoogleCredentials, **kwargs
+    ) -> BlockOutput:
+        service = GmailReadBlock._build_service(credentials, **kwargs)
+        profile = await self._get_profile(service)
+        yield "profile", profile
+
+    async def _get_profile(self, service) -> Profile:
+        result = service.users().getProfile(userId="me").execute()
+        return Profile(
+            emailAddress=result.get("emailAddress", ""),
+            messagesTotal=result.get("messagesTotal", 0),
+            threadsTotal=result.get("threadsTotal", 0),
+            historyId=result.get("historyId", ""),
+        )

autogpt_platform/backend/test/blocks/test_gmail.py

@@ -1,6 +1,8 @@
 import base64
 from unittest.mock import Mock, patch
 
+import pytest
+
 from backend.blocks.google.gmail import GmailReadBlock
 
 
@@ -16,7 +18,8 @@ class TestGmailReadBlock:
         """Helper to encode text as base64 URL-safe."""
         return base64.urlsafe_b64encode(text.encode("utf-8")).decode("utf-8")
 
-    def test_single_part_text_plain(self):
+    @pytest.mark.asyncio
+    async def test_single_part_text_plain(self):
         """Test parsing single-part text/plain email."""
         body_text = "This is a plain text email body."
         msg = {
@@ -27,10 +30,11 @@ class TestGmailReadBlock:
             },
         }
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         assert result == body_text
 
-    def test_multipart_alternative_plain_and_html(self):
+    @pytest.mark.asyncio
+    async def test_multipart_alternative_plain_and_html(self):
         """Test parsing multipart/alternative with both plain and HTML parts."""
         plain_text = "This is the plain text version."
         html_text = "<html><body><p>This is the HTML version.</p></body></html>"
@@ -52,11 +56,12 @@ class TestGmailReadBlock:
             },
         }
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         # Should prefer plain text over HTML
         assert result == plain_text
 
-    def test_html_only_email(self):
+    @pytest.mark.asyncio
+    async def test_html_only_email(self):
         """Test parsing HTML-only email with conversion to plain text."""
         html_text = (
             "<html><body><h1>Hello World</h1><p>This is HTML content.</p></body></html>"
@@ -75,11 +80,12 @@ class TestGmailReadBlock:
             mock_converter.handle.return_value = "Hello World\n\nThis is HTML content."
             mock_html2text.return_value = mock_converter
 
-            result = self.gmail_block._get_email_body(msg, self.mock_service)
+            result = await self.gmail_block._get_email_body(msg, self.mock_service)
             assert "Hello World" in result
             assert "This is HTML content" in result
 
-    def test_html_fallback_when_html2text_unavailable(self):
+    @pytest.mark.asyncio
+    async def test_html_fallback_when_html2text_unavailable(self):
         """Test fallback to raw HTML when html2text is not available."""
         html_text = "<html><body><p>HTML content</p></body></html>"
 
@@ -92,10 +98,11 @@ class TestGmailReadBlock:
         }
 
         with patch("html2text.HTML2Text", side_effect=ImportError):
-            result = self.gmail_block._get_email_body(msg, self.mock_service)
+            result = await self.gmail_block._get_email_body(msg, self.mock_service)
             assert result == html_text
 
-    def test_nested_multipart_structure(self):
+    @pytest.mark.asyncio
+    async def test_nested_multipart_structure(self):
         """Test parsing deeply nested multipart structure."""
         plain_text = "Nested plain text content."
 
@@ -117,10 +124,11 @@ class TestGmailReadBlock:
             },
         }
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         assert result == plain_text
 
-    def test_attachment_body_content(self):
+    @pytest.mark.asyncio
+    async def test_attachment_body_content(self):
         """Test parsing email where body is stored as attachment."""
         attachment_data = self._encode_base64("Body content from attachment.")
 
@@ -137,10 +145,11 @@ class TestGmailReadBlock:
             "data": attachment_data
         }
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         assert result == "Body content from attachment."
 
-    def test_no_readable_body(self):
+    @pytest.mark.asyncio
+    async def test_no_readable_body(self):
         """Test email with no readable body content."""
         msg = {
             "id": "test_msg_7",
@@ -150,10 +159,11 @@ class TestGmailReadBlock:
             },
         }
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         assert result == "This email does not contain a readable body."
 
-    def test_base64_padding_handling(self):
+    @pytest.mark.asyncio
+    async def test_base64_padding_handling(self):
         """Test proper handling of base64 data with missing padding."""
         # Create base64 data with missing padding
         text = "Test content"
@@ -164,7 +174,8 @@ class TestGmailReadBlock:
         result = self.gmail_block._decode_base64(encoded_no_padding)
         assert result == text
 
-    def test_recursion_depth_limit(self):
+    @pytest.mark.asyncio
+    async def test_recursion_depth_limit(self):
         """Test that recursion depth is properly limited."""
 
         # Create a deeply nested structure that would exceed the limit
@@ -184,21 +195,24 @@ class TestGmailReadBlock:
             "payload": create_nested_part(0),
         }
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         # Should return fallback message due to depth limit
         assert result == "This email does not contain a readable body."
 
-    def test_malformed_base64_handling(self):
+    @pytest.mark.asyncio
+    async def test_malformed_base64_handling(self):
         """Test handling of malformed base64 data."""
         result = self.gmail_block._decode_base64("invalid_base64_data!!!")
         assert result is None
 
-    def test_empty_data_handling(self):
+    @pytest.mark.asyncio
+    async def test_empty_data_handling(self):
         """Test handling of empty or None data."""
         assert self.gmail_block._decode_base64("") is None
         assert self.gmail_block._decode_base64(None) is None
 
-    def test_attachment_download_failure(self):
+    @pytest.mark.asyncio
+    async def test_attachment_download_failure(self):
         """Test handling of attachment download failure."""
         msg = {
             "id": "test_msg_9",
@@ -213,5 +227,5 @@ class TestGmailReadBlock:
             Exception("Download failed")
         )
 
-        result = self.gmail_block._get_email_body(msg, self.mock_service)
+        result = await self.gmail_block._get_email_body(msg, self.mock_service)
         assert result == "This email does not contain a readable body."