Google Workspace MCP Server

""" Google Docs Integration Module Create, read, update, and format Google Documents """ from googleapiclient.discovery import build from src.google_auth import get_credentials def get_docs_service(): creds = get_credentials() return build('docs', 'v1', credentials=creds) def get_drive_service(): creds = get_credentials() return build('drive', 'v3', credentials=creds) def create_document(title, initial_content=None): """Create a new Google Document with optional initial content""" try: service = get_docs_service() # Create empty document document = service.documents().create(body={'title': title}).execute() document_id = document.get('documentId') # Add initial content if provided if initial_content: requests = [{ 'insertText': { 'location': {'index': 1}, 'text': initial_content } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() return { 'success': True, 'documentId': document_id, 'title': document.get('title'), 'documentUrl': f"https://docs.google.com/document/d/{document_id}/edit" } except Exception as error: return {'success': False, 'error': str(error)} def read_document(document_id): """Read the content of a Google Document""" try: service = get_docs_service() document = service.documents().get(documentId=document_id).execute() # Extract text content from document body content = document.get('body', {}).get('content', []) text_content = _extract_text(content) return { 'success': True, 'documentId': document_id, 'title': document.get('title', ''), 'content': text_content, 'revisionId': document.get('revisionId', '') } except Exception as error: return {'success': False, 'error': str(error)} def _extract_text(content): """Extract plain text from document content structure""" text_parts = [] for element in content: if 'paragraph' in element: paragraph = element['paragraph'] for elem in paragraph.get('elements', []): if 'textRun' in elem: text_parts.append(elem['textRun'].get('content', '')) elif 'table' in element: # Extract text from tables table = element['table'] for row in table.get('tableRows', []): for cell in row.get('tableCells', []): cell_content = cell.get('content', []) text_parts.append(_extract_text(cell_content)) return ''.join(text_parts) def append_to_document(document_id, content): """Append text to the end of a document""" try: service = get_docs_service() # Get current document to find end index document = service.documents().get(documentId=document_id).execute() body_content = document.get('body', {}).get('content', []) # Find the end index (last element's endIndex - 1) end_index = 1 if body_content: last_element = body_content[-1] end_index = last_element.get('endIndex', 1) - 1 requests = [{ 'insertText': { 'location': {'index': end_index}, 'text': content } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() return { 'success': True, 'documentId': document_id, 'appendedLength': len(content) } except Exception as error: return {'success': False, 'error': str(error)} def insert_text_at(document_id, text, index): """Insert text at a specific index in the document""" try: service = get_docs_service() requests = [{ 'insertText': { 'location': {'index': index}, 'text': text } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() return { 'success': True, 'documentId': document_id, 'insertedAt': index, 'textLength': len(text) } except Exception as error: return {'success': False, 'error': str(error)} def format_text(document_id, start_index, end_index, format_options): """ Apply formatting to a range of text format_options can include: - bold: bool - italic: bool - underline: bool - strikethrough: bool - fontSize: int (in points) - fontFamily: str - foregroundColor: {red, green, blue} (0-1 values) - backgroundColor: {red, green, blue} (0-1 values) - link: str (URL to link to) """ try: service = get_docs_service() text_style = {} fields = [] if 'bold' in format_options: text_style['bold'] = format_options['bold'] fields.append('bold') if 'italic' in format_options: text_style['italic'] = format_options['italic'] fields.append('italic') if 'underline' in format_options: text_style['underline'] = format_options['underline'] fields.append('underline') if 'strikethrough' in format_options: text_style['strikethrough'] = format_options['strikethrough'] fields.append('strikethrough') if 'fontSize' in format_options: text_style['fontSize'] = { 'magnitude': format_options['fontSize'], 'unit': 'PT' } fields.append('fontSize') if 'fontFamily' in format_options: text_style['weightedFontFamily'] = { 'fontFamily': format_options['fontFamily'] } fields.append('weightedFontFamily') if 'foregroundColor' in format_options: text_style['foregroundColor'] = { 'color': {'rgbColor': format_options['foregroundColor']} } fields.append('foregroundColor') if 'backgroundColor' in format_options: text_style['backgroundColor'] = { 'color': {'rgbColor': format_options['backgroundColor']} } fields.append('backgroundColor') if 'link' in format_options: text_style['link'] = {'url': format_options['link']} fields.append('link') requests = [{ 'updateTextStyle': { 'range': { 'startIndex': start_index, 'endIndex': end_index }, 'textStyle': text_style, 'fields': ','.join(fields) } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() return { 'success': True, 'documentId': document_id, 'formattedRange': {'start': start_index, 'end': end_index}, 'appliedFormat': format_options } except Exception as error: return {'success': False, 'error': str(error)} def apply_paragraph_style(document_id, start_index, end_index, style_options): """ Apply paragraph styling style_options can include: - namedStyleType: 'HEADING_1', 'HEADING_2', 'HEADING_3', 'TITLE', 'SUBTITLE', 'NORMAL_TEXT' - alignment: 'START', 'CENTER', 'END', 'JUSTIFIED' - lineSpacing: float (100 = single spacing) - spaceAbove: int (in points) - spaceBelow: int (in points) - indentFirstLine: int (in points) - indentStart: int (in points) """ try: service = get_docs_service() paragraph_style = {} fields = [] if 'namedStyleType' in style_options: paragraph_style['namedStyleType'] = style_options['namedStyleType'] fields.append('namedStyleType') if 'alignment' in style_options: paragraph_style['alignment'] = style_options['alignment'] fields.append('alignment') if 'lineSpacing' in style_options: paragraph_style['lineSpacing'] = style_options['lineSpacing'] fields.append('lineSpacing') if 'spaceAbove' in style_options: paragraph_style['spaceAbove'] = { 'magnitude': style_options['spaceAbove'], 'unit': 'PT' } fields.append('spaceAbove') if 'spaceBelow' in style_options: paragraph_style['spaceBelow'] = { 'magnitude': style_options['spaceBelow'], 'unit': 'PT' } fields.append('spaceBelow') requests = [{ 'updateParagraphStyle': { 'range': { 'startIndex': start_index, 'endIndex': end_index }, 'paragraphStyle': paragraph_style, 'fields': ','.join(fields) } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() return { 'success': True, 'documentId': document_id, 'styledRange': {'start': start_index, 'end': end_index}, 'appliedStyle': style_options } except Exception as error: return {'success': False, 'error': str(error)} def insert_table(document_id, rows, cols, data=None, index=None): """ Insert a table into the document data: 2D array of cell values (optional) index: position to insert (if None, appends to end) """ try: service = get_docs_service() # Get current document to find insert position if index is None: document = service.documents().get(documentId=document_id).execute() body_content = document.get('body', {}).get('content', []) index = 1 if body_content: last_element = body_content[-1] index = last_element.get('endIndex', 1) - 1 # Insert table requests = [{ 'insertTable': { 'rows': rows, 'columns': cols, 'location': {'index': index} } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() # If data provided, populate table cells if data: # Re-fetch document to get table structure document = service.documents().get(documentId=document_id).execute() table_element = _find_table_at_index(document, index) if table_element: populate_requests = _create_table_populate_requests(table_element, data) if populate_requests: service.documents().batchUpdate( documentId=document_id, body={'requests': populate_requests} ).execute() return { 'success': True, 'documentId': document_id, 'tableSize': {'rows': rows, 'cols': cols}, 'insertedAt': index } except Exception as error: return {'success': False, 'error': str(error)} def _find_table_at_index(document, target_index): """Find a table near the target index""" content = document.get('body', {}).get('content', []) for element in content: if 'table' in element: start = element.get('startIndex', 0) if start >= target_index - 1: return element return None def _create_table_populate_requests(table_element, data): """Create requests to populate table cells with data""" requests = [] table = table_element.get('table', {}) table_rows = table.get('tableRows', []) for row_idx, row in enumerate(table_rows): if row_idx >= len(data): break cells = row.get('tableCells', []) for col_idx, cell in enumerate(cells): if col_idx >= len(data[row_idx]): break cell_content = cell.get('content', []) if cell_content: # Find the paragraph in the cell for elem in cell_content: if 'paragraph' in elem: start_index = elem.get('startIndex', 0) text_value = str(data[row_idx][col_idx]) if text_value: requests.append({ 'insertText': { 'location': {'index': start_index}, 'text': text_value } }) break # Reverse to insert from end to start (to avoid index shifts) return list(reversed(requests)) def list_documents(max_results=50): """List all Google Docs in Drive""" try: drive_service = get_drive_service() results = drive_service.files().list( q="mimeType='application/vnd.google-apps.document'", spaces='drive', fields='files(id, name, createdTime, modifiedTime, webViewLink)', orderBy='modifiedTime desc', pageSize=max_results ).execute() files = results.get('files', []) return { 'success': True, 'documents': [ { 'id': f['id'], 'name': f['name'], 'createdTime': f.get('createdTime', ''), 'modifiedTime': f.get('modifiedTime', ''), 'webViewLink': f.get('webViewLink', '') } for f in files ], 'total': len(files) } except Exception as error: return {'success': False, 'error': str(error)} def share_document(document_id, email, role='reader'): """ Share a document with a user role: 'reader', 'writer', 'commenter' """ try: drive_service = get_drive_service() permission = { 'type': 'user', 'role': role, 'emailAddress': email } result = drive_service.permissions().create( fileId=document_id, body=permission, sendNotificationEmail=True ).execute() return { 'success': True, 'permissionId': result.get('id'), 'sharedWith': email, 'role': role } except Exception as error: return {'success': False, 'error': str(error)} def insert_bullet_list(document_id, items, index=None): """Insert a bullet list at the specified position""" try: service = get_docs_service() # Get current document to find insert position if index is None: document = service.documents().get(documentId=document_id).execute() body_content = document.get('body', {}).get('content', []) index = 1 if body_content: last_element = body_content[-1] index = last_element.get('endIndex', 1) - 1 # Insert all items as text first text_to_insert = '\n'.join(items) + '\n' insert_request = { 'insertText': { 'location': {'index': index}, 'text': text_to_insert } } service.documents().batchUpdate( documentId=document_id, body={'requests': [insert_request]} ).execute() # Apply bullet formatting end_index = index + len(text_to_insert) bullet_request = { 'createParagraphBullets': { 'range': { 'startIndex': index, 'endIndex': end_index }, 'bulletPreset': 'BULLET_DISC_CIRCLE_SQUARE' } } service.documents().batchUpdate( documentId=document_id, body={'requests': [bullet_request]} ).execute() return { 'success': True, 'documentId': document_id, 'itemCount': len(items), 'insertedAt': index } except Exception as error: return {'success': False, 'error': str(error)} def insert_numbered_list(document_id, items, index=None): """Insert a numbered list at the specified position""" try: service = get_docs_service() # Get current document to find insert position if index is None: document = service.documents().get(documentId=document_id).execute() body_content = document.get('body', {}).get('content', []) index = 1 if body_content: last_element = body_content[-1] index = last_element.get('endIndex', 1) - 1 # Insert all items as text first text_to_insert = '\n'.join(items) + '\n' insert_request = { 'insertText': { 'location': {'index': index}, 'text': text_to_insert } } service.documents().batchUpdate( documentId=document_id, body={'requests': [insert_request]} ).execute() # Apply numbered list formatting end_index = index + len(text_to_insert) numbered_request = { 'createParagraphBullets': { 'range': { 'startIndex': index, 'endIndex': end_index }, 'bulletPreset': 'NUMBERED_DECIMAL_ALPHA_ROMAN' } } service.documents().batchUpdate( documentId=document_id, body={'requests': [numbered_request]} ).execute() return { 'success': True, 'documentId': document_id, 'itemCount': len(items), 'insertedAt': index } except Exception as error: return {'success': False, 'error': str(error)} # Helper function for creating reports def create_daily_report(title=None, sections=None): """ Create a daily report document with predefined sections sections: dict with keys like 'summary', 'tasks_completed', 'tasks_pending', 'notes' """ try: from datetime import datetime if title is None: title = f"Daily Report - {datetime.now().strftime('%Y-%m-%d')}" # Create document result = create_document(title) if not result['success']: return result document_id = result['documentId'] service = get_docs_service() # Build content content_parts = [] if sections is None: sections = { 'summary': '', 'tasks_completed': [], 'tasks_pending': [], 'notes': '' } # Title content_parts.append(f"{title}\n\n") # Summary section content_parts.append("Summary\n") content_parts.append(f"{sections.get('summary', 'Enter summary here...')}\n\n") # Tasks Completed content_parts.append("Tasks Completed\n") tasks_completed = sections.get('tasks_completed', []) if tasks_completed: for task in tasks_completed: content_parts.append(f"- {task}\n") else: content_parts.append("- No tasks completed\n") content_parts.append("\n") # Tasks Pending content_parts.append("Tasks Pending\n") tasks_pending = sections.get('tasks_pending', []) if tasks_pending: for task in tasks_pending: content_parts.append(f"- {task}\n") else: content_parts.append("- No pending tasks\n") content_parts.append("\n") # Notes content_parts.append("Notes\n") content_parts.append(f"{sections.get('notes', 'Additional notes...')}\n") full_content = ''.join(content_parts) # Insert all content requests = [{ 'insertText': { 'location': {'index': 1}, 'text': full_content } }] service.documents().batchUpdate( documentId=document_id, body={'requests': requests} ).execute() # Apply heading styles # Title (first line) apply_paragraph_style(document_id, 1, len(title) + 1, {'namedStyleType': 'TITLE'}) return { 'success': True, 'documentId': document_id, 'documentUrl': result['documentUrl'], 'title': title, 'message': 'Daily report created with Summary, Tasks Completed, Tasks Pending, and Notes sections' } except Exception as error: return {'success': False, 'error': str(error)}