Google Workspace MCP Server

Instructions

Implementation Reference

• core/comments.py:183-202 (handler)

  Core handler logic that creates a comment using the Google Drive Comments API's comments.create method. This is the main execution logic for the tool.
  async def _create_comment_impl(service, app_name: str, file_id: str, comment_content: str) -> str: """Implementation for creating a comment on any Google Workspace file.""" logger.info(f"[create_{app_name}_comment] Creating comment in {app_name} {file_id}") body = {"content": comment_content} comment = await asyncio.to_thread( service.comments().create( fileId=file_id, body=body, fields="id,content,author,createdTime,modifiedTime" ).execute ) comment_id = comment.get('id', '') author = comment.get('author', {}).get('displayName', 'Unknown') created = comment.get('createdTime', '') return f"Comment created successfully!\\nComment ID: {comment_id}\\nAuthor: {author}\\nCreated: {created}\\nContent: {comment_content}"
• core/comments.py:39-43 (handler)

  The decorated handler function specifically for documents (create_document_comment), which calls the core impl. This is the direct MCP tool handler.
  @require_google_service("drive", "drive_read") @handle_http_errors(read_func_name, service_type="drive") async def read_comments(service, user_google_email: str, document_id: str) -> str: """Read all comments from a Google Document.""" return await _read_comments_impl(service, app_name, document_id)
• core/comments.py:114-123 (registration)

  Sets the function __name__ to 'create_document_comment' and registers it as an MCP tool using server.tool().
  read_comments.__name__ = read_func_name create_comment.__name__ = create_func_name reply_to_comment.__name__ = reply_func_name resolve_comment.__name__ = resolve_func_name # Register tools with the server using the proper names server.tool()(read_comments) server.tool()(create_comment) server.tool()(reply_to_comment) server.tool()(resolve_comment)
• core/comments.py:19-130 (helper)

  Factory function that generates and registers the create_document_comment tool when called with app_name='document'.
  def create_comment_tools(app_name: str, file_id_param: str): """ Factory function to create comment management tools for a specific Google Workspace app. Args: app_name: Name of the app (e.g., "document", "spreadsheet", "presentation") file_id_param: Parameter name for the file ID (e.g., "document_id", "spreadsheet_id", "presentation_id") Returns: Dict containing the four comment management functions with unique names """ # Create unique function names based on the app type read_func_name = f"read_{app_name}_comments" create_func_name = f"create_{app_name}_comment" reply_func_name = f"reply_to_{app_name}_comment" resolve_func_name = f"resolve_{app_name}_comment" # Create functions without decorators first, then apply decorators with proper names if file_id_param == "document_id": @require_google_service("drive", "drive_read") @handle_http_errors(read_func_name, service_type="drive") async def read_comments(service, user_google_email: str, document_id: str) -> str: """Read all comments from a Google Document.""" return await _read_comments_impl(service, app_name, document_id) @require_google_service("drive", "drive_file") @handle_http_errors(create_func_name, service_type="drive") async def create_comment(service, user_google_email: str, document_id: str, comment_content: str) -> str: """Create a new comment on a Google Document.""" return await _create_comment_impl(service, app_name, document_id, comment_content) @require_google_service("drive", "drive_file") @handle_http_errors(reply_func_name, service_type="drive") async def reply_to_comment(service, user_google_email: str, document_id: str, comment_id: str, reply_content: str) -> str: """Reply to a specific comment in a Google Document.""" return await _reply_to_comment_impl(service, app_name, document_id, comment_id, reply_content) @require_google_service("drive", "drive_file") @handle_http_errors(resolve_func_name, service_type="drive") async def resolve_comment(service, user_google_email: str, document_id: str, comment_id: str) -> str: """Resolve a comment in a Google Document.""" return await _resolve_comment_impl(service, app_name, document_id, comment_id) elif file_id_param == "spreadsheet_id": @require_google_service("drive", "drive_read") @handle_http_errors(read_func_name, service_type="drive") async def read_comments(service, user_google_email: str, spreadsheet_id: str) -> str: """Read all comments from a Google Spreadsheet.""" return await _read_comments_impl(service, app_name, spreadsheet_id) @require_google_service("drive", "drive_file") @handle_http_errors(create_func_name, service_type="drive") async def create_comment(service, user_google_email: str, spreadsheet_id: str, comment_content: str) -> str: """Create a new comment on a Google Spreadsheet.""" return await _create_comment_impl(service, app_name, spreadsheet_id, comment_content) @require_google_service("drive", "drive_file") @handle_http_errors(reply_func_name, service_type="drive") async def reply_to_comment(service, user_google_email: str, spreadsheet_id: str, comment_id: str, reply_content: str) -> str: """Reply to a specific comment in a Google Spreadsheet.""" return await _reply_to_comment_impl(service, app_name, spreadsheet_id, comment_id, reply_content) @require_google_service("drive", "drive_file") @handle_http_errors(resolve_func_name, service_type="drive") async def resolve_comment(service, user_google_email: str, spreadsheet_id: str, comment_id: str) -> str: """Resolve a comment in a Google Spreadsheet.""" return await _resolve_comment_impl(service, app_name, spreadsheet_id, comment_id) elif file_id_param == "presentation_id": @require_google_service("drive", "drive_read") @handle_http_errors(read_func_name, service_type="drive") async def read_comments(service, user_google_email: str, presentation_id: str) -> str: """Read all comments from a Google Presentation.""" return await _read_comments_impl(service, app_name, presentation_id) @require_google_service("drive", "drive_file") @handle_http_errors(create_func_name, service_type="drive") async def create_comment(service, user_google_email: str, presentation_id: str, comment_content: str) -> str: """Create a new comment on a Google Presentation.""" return await _create_comment_impl(service, app_name, presentation_id, comment_content) @require_google_service("drive", "drive_file") @handle_http_errors(reply_func_name, service_type="drive") async def reply_to_comment(service, user_google_email: str, presentation_id: str, comment_id: str, reply_content: str) -> str: """Reply to a specific comment in a Google Presentation.""" return await _reply_to_comment_impl(service, app_name, presentation_id, comment_id, reply_content) @require_google_service("drive", "drive_file") @handle_http_errors(resolve_func_name, service_type="drive") async def resolve_comment(service, user_google_email: str, presentation_id: str, comment_id: str) -> str: """Resolve a comment in a Google Presentation.""" return await _resolve_comment_impl(service, app_name, presentation_id, comment_id) # Set the proper function names and register with server read_comments.__name__ = read_func_name create_comment.__name__ = create_func_name reply_to_comment.__name__ = reply_func_name resolve_comment.__name__ = resolve_func_name # Register tools with the server using the proper names server.tool()(read_comments) server.tool()(create_comment) server.tool()(reply_to_comment) server.tool()(resolve_comment) return { 'read_comments': read_comments, 'create_comment': create_comment, 'reply_to_comment': reply_to_comment, 'resolve_comment': resolve_comment }
• gdocs/docs_tools.py:1174-1178 (helper)

  Invokes the factory to create document-specific comment tools, including create_document_comment, and aliases it locally as create_doc_comment.
  _comment_tools = create_comment_tools("document", "document_id") # Extract and register the functions read_doc_comments = _comment_tools['read_comments'] create_doc_comment = _comment_tools['create_comment']