Google Toolbox

Instructions

Implementation Reference

• server.py:674-783 (handler)

  The main handler function that executes the read_gdrive_file tool. It uses Google Drive API to retrieve file metadata, handles Google Apps documents by exporting to appropriate formats, downloads regular files, and returns structured content with metadata and success status.
  async def read_gdrive_file(file_id: str) -> Dict[str, Any]: """ Read contents of a file from Google Drive Args: file_id (str): ID of the file to read Returns: Dict[str, Any]: A dictionary containing: - success (bool): Whether the operation was successful - name (str): Name of the file - mime_type (str): MIME type of the file - content (str): Text content or base64-encoded binary content - is_text (bool): Whether the content is text or binary - error (str): Error message (when unsuccessful) """ creds = get_google_credentials() if not creds: return { "success": False, "error": "Google authentication failed." } try: # Initialize Google Drive API service service = build('drive', 'v3', credentials=creds) # First get file metadata to check mime type file = service.files().get( fileId=file_id, fields="mimeType,name" ).execute() file_name = file.get('name', file_id) mime_type = file.get('mimeType', 'application/octet-stream') # For Google Docs/Sheets/etc we need to export if mime_type.startswith('application/vnd.google-apps'): export_mime_type = 'text/plain' # Default # Determine appropriate export format based on file type if mime_type == 'application/vnd.google-apps.document': export_mime_type = 'text/markdown' elif mime_type == 'application/vnd.google-apps.spreadsheet': export_mime_type = 'text/csv' elif mime_type == 'application/vnd.google-apps.presentation': export_mime_type = 'text/plain' elif mime_type == 'application/vnd.google-apps.drawing': export_mime_type = 'image/png' # Export the file response = service.files().export( fileId=file_id, mimeType=export_mime_type ).execute() # Handle response based on mime type is_text = export_mime_type.startswith('text/') or export_mime_type == 'application/json' content = response if is_text else base64.b64encode(response).decode('utf-8') return { "success": True, "name": file_name, "mime_type": export_mime_type, "content": content, "is_text": is_text } # For regular files, download content request = service.files().get_media(fileId=file_id) file_content = io.BytesIO() downloader = MediaIoBaseDownload(file_content, request) done = False while not done: _, done = downloader.next_chunk() # Determine if content is text based on mime type is_text = mime_type.startswith('text/') or mime_type == 'application/json' content_bytes = file_content.getvalue() # Prepare response based on content type if is_text: content = content_bytes.decode('utf-8') else: content = base64.b64encode(content_bytes).decode('utf-8') logger.info(f"읽은 파일: {file_name} ({mime_type}), 크기: {len(content_bytes)} 바이트") return { "success": True, "name": file_name, "mime_type": mime_type, "content": content, "is_text": is_text } except HttpError as error: logger.error(f"Drive API 오류 발생: {error}") return { "success": False, "error": f"Google Drive API Error: {str(error)}" } except Exception as e: logger.exception("파일 읽기 중 오류:") return { "success": False, "error": f"예상치 못한 오류 발생: {str(e)}" }
• server.py:670-673 (registration)

  The @mcp.tool decorator that registers the read_gdrive_file tool with the MCP server, specifying its name and description.
  @mcp.tool( name="read_gdrive_file", description="Read contents of a file from Google Drive", )
• server.py:674-689 (schema)

  The function signature and docstring that define the input schema (file_id: str) and output schema (Dict[str, Any] with specific keys: success, name, mime_type, content, is_text, error).
  async def read_gdrive_file(file_id: str) -> Dict[str, Any]: """ Read contents of a file from Google Drive Args: file_id (str): ID of the file to read Returns: Dict[str, Any]: A dictionary containing: - success (bool): Whether the operation was successful - name (str): Name of the file - mime_type (str): MIME type of the file - content (str): Text content or base64-encoded binary content - is_text (bool): Whether the content is text or binary - error (str): Error message (when unsuccessful) """
• server.py:113-160 (helper)

  Helper function to obtain Google API credentials from token.json or environment variables, used by the read_gdrive_file handler for authentication.
  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
• server.py:203-203 (registration)

  The read_gdrive_file tool is listed in the available_google_tools resource, which provides a list of all available Google tools on the MCP server.
  "search_google", "read_gdrive_file", "search_gdrive"