search-emails

Instructions

Implementation Reference

• src/email_client/server.py:317-389 (handler)

  Handler for 'search-emails' tool: parses arguments, builds IMAP search criteria from dates/keyword/folder, performs async search, formats and returns table of email summaries.

  if name == "search-emails":
      # 选择文件夹
      folder = arguments.get("folder", "inbox")  # 默认选择收件箱
      if folder == "sent":
          mail.select('"[Gmail]/Sent Mail"')  # 对于 Gmail
      else:
          mail.select("inbox")
      
      # Get optional parameters
      start_date = arguments.get("start_date")
      end_date = arguments.get("end_date")
      keyword = arguments.get("keyword")
      
      # If no dates provided, default to last 7 days
      if not start_date:
          start_date = datetime.now() - timedelta(days=7)
          start_date = start_date.strftime("%d-%b-%Y")
      else:
          start_date = datetime.strptime(start_date, "%Y-%m-%d").strftime("%d-%b-%Y")
          
      if not end_date:
          end_date = datetime.now().strftime("%d-%b-%Y")
      else:
          # Convert end_date to datetime object once
          end_date_obj = datetime.strptime(end_date, "%Y-%m-%d")
          end_date = end_date_obj.strftime("%d-%b-%Y")
      
      # Build search criteria
      if start_date == end_date:
          # If searching for a single day
          search_criteria = f'ON "{start_date}"'
      else:
          # Calculate next day using the already converted end_date_obj
          next_day = (end_date_obj + timedelta(days=1)).strftime("%d-%b-%Y")
          search_criteria = f'SINCE "{start_date}" BEFORE "{next_day}"'
          
      if keyword:
          # Fix: Properly combine keyword search with date criteria
          keyword_criteria = f'(OR SUBJECT "{keyword}" BODY "{keyword}")'
          search_criteria = f'({keyword_criteria} {search_criteria})'
      
      logging.debug(f"Search criteria: {search_criteria}")  # Add debug logging
      
      try:
          async with asyncio.timeout(SEARCH_TIMEOUT):
              email_list = await search_emails_async(mail, search_criteria)
              
          if not email_list:
              return [types.TextContent(
                  type="text",
                  text="No emails found matching the criteria."
              )]
          
          # Format the results as a table
          result_text = "Found emails:\n\n"
          result_text += "ID | From | Date | Subject\n"
          result_text += "-" * 80 + "\n"
          
          for email in email_list:
              result_text += f"{email['id']} | {email['from']} | {email['date']} | {email['subject']}\n"
          
          result_text += "\nUse get-email-content with an email ID to view the full content of a specific email."
          
          return [types.TextContent(
              type="text",
              text=result_text
          )]
          
      except asyncio.TimeoutError:
          return [types.TextContent(
              type="text",
              text="Search operation timed out. Please try with a more specific search criteria."
          )]

• src/email_client/server.py:175-200 (registration)

  Registration of the 'search-emails' tool in the @server.list_tools() function, including name, description, and input schema.

  types.Tool(
      name="search-emails",
      description="Search emails within a date range and/or with specific keywords",
      inputSchema={
          "type": "object",
          "properties": {
              "start_date": {
                  "type": "string",
                  "description": "Start date in YYYY-MM-DD format (optional)",
              },
              "end_date": {
                  "type": "string",
                  "description": "End date in YYYY-MM-DD format (optional)",
              },
              "keyword": {
                  "type": "string",
                  "description": "Keyword to search in email subject and body (optional)",
              },
              "folder": {
                  "type": "string",
                  "description": "Folder to search in ('inbox' or 'sent', defaults to 'inbox')",
                  "enum": ["inbox", "sent"],
              },
          },
      },
  ),

• src/email_client/server.py:178-199 (schema)

  JSON schema defining inputs for search-emails: optional start_date, end_date, keyword, folder (inbox/sent).

  inputSchema={
      "type": "object",
      "properties": {
          "start_date": {
              "type": "string",
              "description": "Start date in YYYY-MM-DD format (optional)",
          },
          "end_date": {
              "type": "string",
              "description": "End date in YYYY-MM-DD format (optional)",
          },
          "keyword": {
              "type": "string",
              "description": "Keyword to search in email subject and body (optional)",
          },
          "folder": {
              "type": "string",
              "description": "Folder to search in ('inbox' or 'sent', defaults to 'inbox')",
              "enum": ["inbox", "sent"],
          },
      },
  },

• src/email_client/server.py:81-97 (helper)

  Helper function that performs the actual IMAP search using the criteria, fetches up to MAX_EMAILS emails, formats summaries using format_email_summary.

  async def search_emails_async(mail: imaplib.IMAP4_SSL, search_criteria: str) -> list[dict]:
      """Asynchronously search emails with timeout."""
      loop = asyncio.get_event_loop()
      try:
          _, messages = await loop.run_in_executor(None, lambda: mail.search(None, search_criteria))
          if not messages[0]:
              return []
              
          email_list = []
          for num in messages[0].split()[:MAX_EMAILS]:  # Limit to MAX_EMAILS
              _, msg_data = await loop.run_in_executor(None, lambda: mail.fetch(num, '(RFC822)'))
              email_list.append(format_email_summary(msg_data))
              
          return email_list
      except Exception as e:
          raise Exception(f"Error searching emails: {str(e)}")

• src/email_client/server.py:42-51 (helper)

  Helper utility to parse raw IMAP email data into a summary dictionary with id, from, date, subject.

  def format_email_summary(msg_data: tuple) -> dict:
      """Format an email message into a summary dict with basic information."""
      email_body = email.message_from_bytes(msg_data[0][1])
      
      return {
          "id": msg_data[0][0].split()[0].decode(),  # Get the email ID
          "from": email_body.get("From", "Unknown"),
          "date": email_body.get("Date", "Unknown"),
          "subject": email_body.get("Subject", "No Subject"),
      }