search_emails

Instructions

Implementation Reference

• server.py:265-315 (handler)

  The main handler function for the 'search_emails' tool. It authenticates with Google Gmail API, performs the search using the provided query, fetches full message details including body extraction via helper, and returns a list of email dictionaries.

  async def search_emails(query: str, max_results: int = 10) -> List[Dict[str, Any]]:
      """
      Search emails with advanced query
      
      Args:
          query (str): Gmail search query (e.g., "from:example@gmail.com has:attachment")
          max_results (int): Maximum number of emails to return (default: 10)
      
      Returns:
          List[Dict[str, Any]]: List of email details
      """
      creds = get_google_credentials()
      if not creds:
          return "Google authentication failed."
  
      try:
          service = build('gmail', 'v1', credentials=creds)
          results = service.users().messages().list(
              userId='me',
              maxResults=max_results,
              q=query
          ).execute()
  
          messages = results.get('messages', [])
          email_details = []
          for msg in messages:
              msg_full = service.users().messages().get(userId='me', id=msg['id'], format='full').execute()
              payload = msg_full.get('payload', {})
              headers = {h['name']: h['value'] for h in payload.get('headers', [])}
              body = _get_email_body(payload)
              email_details.append({
                  'id': msg['id'],
                  'threadId': msg.get('threadId'),
                  'subject': headers.get('Subject', '(No Subject)'),
                  'from': headers.get('From'),
                  'to': headers.get('To'),
                  'date': headers.get('Date'),
                  'body': body,
                  'labels': msg_full.get('labelIds', []),
                  'snippet': msg_full.get('snippet'),
              })
  
          return email_details
  
      except HttpError as error:
          logger.error(f"API 오류 발생: {error}")
          return f"Gmail API 오류: {error.resp.status} - {error.content.decode()}"
      except Exception as e:
          logger.exception("이메일 검색 중 오류:")
          return f"예상치 못한 오류 발생: {str(e)}"

• server.py:261-264 (registration)

  The @mcp.tool decorator that registers the search_emails function as an MCP tool with name and description.

  @mcp.tool(
      name="search_emails",
      description="Search emails with advanced query",
  )

• server.py:172-190 (helper)

  Helper function used by the search_emails handler to extract plain text body from Gmail message payloads.

  def _get_email_body(payload: Dict[str, Any]) -> str:
      """Helper function to extract plain text body from email payload."""
      # (Helper function 내용은 변경 없음)
      if not payload:
          return "(No body content)"
      if 'parts' in payload:
          for part in payload['parts']:
              if part['mimeType'] == 'text/plain' and 'body' in part and 'data' in part['body']:
                  return base64.urlsafe_b64decode(part['body']['data']).decode('utf-8')
          for part in payload['parts']:
              if part['mimeType'] == 'multipart/alternative' and 'parts' in part:
                  for sub_part in part['parts']:
                      if sub_part['mimeType'] == 'text/plain' and 'body' in sub_part and 'data' in sub_part['body']:
                          return base64.urlsafe_b64decode(sub_part['body']['data']).decode('utf-8')
      if 'body' in payload and 'data' in payload['body']:
          mime_type = payload.get('mimeType', 'text/plain')
          if 'text/plain' in mime_type:
              return base64.urlsafe_b64decode(payload['body']['data']).decode('utf-8')
      return "(Could not extract plain text body)"