create_email_draft

Instructions

Implementation Reference

• src/adeu/mcp_components/tools/email.py:450-542 (handler)

  The actual implementation of the create_email_draft tool. It validates inputs (reply_to_email_id or subject+to_recipients), resolves email IDs, parses recipients/attachments, reads attachment bytes, encodes a multipart form-data request, and POSTs to the Adeu Cloud backend to create an email draft.

  async def create_email_draft(
      ctx: Context,
      body_markdown: Annotated[str, "The body of the email in Markdown format. Will be converted to HTML."],
      reply_to_email_id: Annotated[Optional[str], "Provide the short email ID to reply to an existing thread."] = None,
      subject: Annotated[Optional[str], "The subject line. Required if starting a NEW email."] = None,
      to_recipients: Annotated[Optional[list[str] | str], "List of emails. Required if starting a NEW email."] = None,
      attachment_paths: Annotated[
          Optional[list[str] | str],
          "List of absolute file paths on the local system to attach to the draft.",
      ] = None,
      api_key: str = Depends(get_cloud_auth_token),
  ) -> ToolResult:
      # 1. Validation
      if not reply_to_email_id and (not subject or not to_recipients):
          return ToolResult(
              "Error: You must provide either 'reply_to_email_id' (to reply) OR "
              "both 'subject' and 'to_recipients' (to start a new email).",
          )
  
      await ctx.info(
          "Creating email draft",
          extra={"reply_to": reply_to_email_id, "subject": subject},
      )
      url = f"{BACKEND_URL}/api/v1/emails/drafts/new"
  
      # Helper to safely parse stringified lists from Claude
      def _parse_list(val) -> list[str]:
          if not val:
              return []
          if isinstance(val, list):
              return val
          try:
              parsed = json.loads(val)
              return parsed if isinstance(parsed, list) else [val]
          except Exception:
              return [x.strip() for x in val.split(",") if x.strip()]
  
      parsed_recipients = _parse_list(to_recipients)
      parsed_attachments = _parse_list(attachment_paths)
  
      # 2. Resolve Minified ID to Real Graph ID
      real_reply_to_id = resolve_email_id(reply_to_email_id) if reply_to_email_id else None
  
      # Prepare text fields
      fields = {
          "body_markdown": body_markdown,
      }
      if real_reply_to_id:
          fields["reply_to_email_id"] = real_reply_to_id
      if subject:
          fields["subject"] = subject
      if parsed_recipients:
          fields["to_recipients"] = json.dumps(parsed_recipients)
  
      # Prepare file bytes
      files_to_upload = []
      if parsed_attachments:
          for path in parsed_attachments:
              try:
                  file_bytes = read_file_bytes(path).getvalue()
                  filename = Path(path).name
                  files_to_upload.append(("files", filename, file_bytes))
              except Exception as e:
                  raise ToolError(f"Failed to read attachment {path}: {e}") from e
  
      # Encode payload
      body, content_type = encode_multipart_formdata(fields=fields, files=files_to_upload)
  
      req = urllib.request.Request(
          url,
          data=body,
          headers={
              "Authorization": f"Bearer {api_key}",
              "Content-Type": content_type,
              "Accept": "application/json",
          },
          method="POST",
      )
  
      try:
          await ctx.debug("Sending draft request to Adeu Cloud")
          with urllib.request.urlopen(req) as response:
              data = json.loads(response.read().decode("utf-8"))
              draft_id = data.get("id")
              return ToolResult(content=f"Successfully created email draft! Draft ID: {draft_id}")
      except urllib.error.HTTPError as e:
          if e.code == 401:
              DesktopAuthManager.clear_api_key()
              raise ToolError("Authentication expired. Please call `login_to_adeu_cloud` to re-authenticate.") from e
          error_body = e.read().decode("utf-8")
          raise ToolError(f"Cloud draft creation failed (HTTP {e.code}): {error_body}") from e
      except Exception as e:
          raise ToolError(f"Failed to communicate with Adeu Cloud: {str(e)}") from e

• src/adeu/mcp_components/tools/email.py:450-460 (schema)

  The function signature uses Annotated type hints to define the input parameters: body_markdown (str), reply_to_email_id (Optional[str]), subject (Optional[str]), to_recipients (Optional[list[str]|str]), attachment_paths (Optional[list[str]|str]), and api_key (injected via Depends).

  async def create_email_draft(
      ctx: Context,
      body_markdown: Annotated[str, "The body of the email in Markdown format. Will be converted to HTML."],
      reply_to_email_id: Annotated[Optional[str], "Provide the short email ID to reply to an existing thread."] = None,
      subject: Annotated[Optional[str], "The subject line. Required if starting a NEW email."] = None,
      to_recipients: Annotated[Optional[list[str] | str], "List of emails. Required if starting a NEW email."] = None,
      attachment_paths: Annotated[
          Optional[list[str] | str],
          "List of absolute file paths on the local system to attach to the draft.",
      ] = None,
      api_key: str = Depends(get_cloud_auth_token),

• src/adeu/mcp_components/tools/email.py:439-448 (registration)

  The @tool decorator registers 'create_email_draft' with its name and description, explaining how to start a new email vs reply, and that attachments and Markdown body are supported.

  @tool(
      name="create_email_draft",
      description=(
          "Creates an email draft in the user's native draft box (e.g., Outlook/Gmail). "
          "Can either start a NEW email, or REPLY to an existing thread. "
          "To REPLY, provide 'reply_to_email_id' (the short ID from search_and_fetch_emails). "
          "To start a NEW email, omit the ID but provide 'subject' and 'to_recipients'. "
          "Allows attaching local files (PDF/DOCX) by providing their absolute paths. "
          "The body should be formatted in Markdown."
      ),

• src/adeu/mcp_components/shared.py:52-78 (helper)

  encode_multipart_formdata helper used to encode the email draft payload (fields + file attachments) into multipart/form-data format for the HTTP POST request.

  def encode_multipart_formdata(
      fields: Optional[Dict[str, str]] = None,
      files: Optional[List[tuple[str, str, bytes]]] = None,
  ) -> tuple[bytes, str]:
      boundary = uuid.uuid4().hex
      buffer = BytesIO()
  
      if fields:
          for key, value in fields.items():
              buffer.write(f"--{boundary}\r\n".encode("utf-8"))
              buffer.write(f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode("utf-8"))
              buffer.write(value.encode("utf-8"))
              buffer.write(b"\r\n")
  
      if files:
          for field_name, file_name, file_bytes in files:
              buffer.write(f"--{boundary}\r\n".encode("utf-8"))
              buffer.write(
                  f'Content-Disposition: form-data; name="{field_name}"; filename="{file_name}"\r\n'.encode("utf-8")
              )
              content_type = mimetypes.guess_type(file_name)[0] or "application/octet-stream"
              buffer.write(f"Content-Type: {content_type}\r\n\r\n".encode("utf-8"))
              buffer.write(file_bytes)
              buffer.write(b"\r\n")
  
      buffer.write(f"--{boundary}--\r\n".encode("utf-8"))
      return buffer.getvalue(), f"multipart/form-data; boundary={boundary}"

• src/adeu/mcp_components/shared.py:16-21 (helper)

  read_file_bytes helper used to read attachment files from the local filesystem into BytesIO objects.

  def read_file_bytes(path: str) -> BytesIO:
      p = Path(path)
      if not p.exists():
          raise FileNotFoundError(f"File not found: {path}")
      with open(p, "rb") as f:
          return BytesIO(f.read())