list_emails

Instructions

Implementation Reference

• server.py:209-259 (handler)

  The handler function for the 'list_emails' tool. It is decorated with @mcp.tool(name="list_emails"), takes query and max_results parameters, authenticates with Google, queries the Gmail API for messages, extracts metadata (id, threadId, subject, from, date, snippet), and returns a list of email dictionaries.

  @mcp.tool(
      name="list_emails",
      description="List recent emails from Gmail inbox",
  )
  async def list_emails(query: str, max_results: int = 10) -> List[Dict[str, Any]]:
      """
      List recent emails from Gmail inbox
      
      Args:
          query (str): Search query to filter emails
          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 or ""
          ).execute()
  
          messages = results.get('messages', [])
          email_details = []
          for msg in messages:
              msg_data = service.users().messages().get(userId='me', id=msg['id'], format='metadata', metadataHeaders=['Subject', 'From', 'Date']).execute()
              headers = {h['name']: h['value'] for h in msg_data.get('payload', {}).get('headers', [])}
              email_details.append({
                  'id': msg['id'],
                  'threadId': msg.get('threadId'),
                  'subject': headers.get('Subject', '(No Subject)'),
                  'from': headers.get('From'),
                  'date': headers.get('Date'),
                  'snippet': msg_data.get('snippet'),
              })
  
          # CallToolResponse 반환 (MCP 표준)
          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:209-212 (registration)

  The @mcp.tool decorator registers the 'list_emails' tool with MCP server, specifying its name and description.

  @mcp.tool(
      name="list_emails",
      description="List recent emails from Gmail inbox",
  )

• server.py:213-213 (schema)

  The function signature provides the input schema (query: str, max_results: int=10) and output type (List[Dict[str, Any]]), along with detailed docstring describing parameters and return value.

  async def list_emails(query: str, max_results: int = 10) -> List[Dict[str, Any]]:

• server.py:113-161 (helper)

  The get_google_credentials helper function used by list_emails to obtain Google API credentials from token.json or environment variables, handling refresh if needed.

  def get_google_credentials() -> Optional[Credentials]:
      """
      Google API 접근을 위한 인증 정보를 가져옵니다.
      기존 token.json 파일이 있으면 로드하고, 만료 시 리프레시합니다.
      없거나 유효하지 않으면 None을 반환합니다 (초기 인증 필요).
      """
      creds = None
      if os.path.exists(TOKEN_FILE):
          try:
              with open(TOKEN_FILE, 'r') as token:
                  creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES)
          except Exception as e:
              logger.error(f"토큰 파일 로딩 오류: {e}")
              creds = None
      if not creds or not creds.valid:
          if creds and creds.expired and creds.refresh_token:
              logger.info("Google API 자격 증명 갱신 중.")
              try:
                  creds.refresh(Request())
                  with open(TOKEN_FILE, 'w') as token:
                      token.write(creds.to_json())
                  logger.info("자격 증명 갱신 및 저장 완료.")
              except Exception as e:
                  logger.error(f"토큰 갱신 실패: {e}")
                  return None
          elif GOOGLE_REFRESH_TOKEN and GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET:
              logger.info("환경 변수의 리프레시 토큰 사용 중.")
              try:
                  creds = Credentials(
                      token=None,
                      refresh_token=GOOGLE_REFRESH_TOKEN,
                      token_uri='https://oauth2.googleapis.com/token',
                      client_id=GOOGLE_CLIENT_ID,
                      client_secret=GOOGLE_CLIENT_SECRET,
                      scopes=SCOPES
                  ) 
                  creds.refresh(Request())
                  with open(TOKEN_FILE, 'w') as token:
                      token.write(creds.to_json())
                  logger.info("리프레시 토큰으로 자격 증명 얻고 저장 완료.")
              except Exception as e:
                  logger.error(f"리프레시 토큰으로 토큰 얻기 실패: {e}")
                  return None
          else:
              logger.error("유효한 자격 증명 또는 리프레시 토큰을 찾을 수 없습니다. 수동 인증이 필요합니다.")
              return None
  
      return creds