Google Toolbox

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