Skip to main content
Glama

Google Workspace MCP Server - Control Gmail, Calendar, Docs, Sheets, Slides, Chat, Forms & Drive

sheets_tools.py13 kB
""" Google Sheets MCP Tools This module provides MCP tools for interacting with Google Sheets API. """ import logging import asyncio import json from typing import List, Optional, Union from auth.service_decorator import require_google_service from core.server import server from core.utils import handle_http_errors from core.comments import create_comment_tools # Configure module logger logger = logging.getLogger(__name__) @server.tool() @handle_http_errors("list_spreadsheets", is_read_only=True, service_type="sheets") @require_google_service("drive", "drive_read") async def list_spreadsheets( service, user_google_email: str, max_results: int = 25, ) -> str: """ Lists spreadsheets from Google Drive that the user has access to. Args: user_google_email (str): The user's Google email address. Required. max_results (int): Maximum number of spreadsheets to return. Defaults to 25. Returns: str: A formatted list of spreadsheet files (name, ID, modified time). """ logger.info(f"[list_spreadsheets] Invoked. Email: '{user_google_email}'") files_response = await asyncio.to_thread( service.files() .list( q="mimeType='application/vnd.google-apps.spreadsheet'", pageSize=max_results, fields="files(id,name,modifiedTime,webViewLink)", orderBy="modifiedTime desc", ) .execute ) files = files_response.get("files", []) if not files: return f"No spreadsheets found for {user_google_email}." spreadsheets_list = [ f"- \"{file['name']}\" (ID: {file['id']}) | Modified: {file.get('modifiedTime', 'Unknown')} | Link: {file.get('webViewLink', 'No link')}" for file in files ] text_output = ( f"Successfully listed {len(files)} spreadsheets for {user_google_email}:\n" + "\n".join(spreadsheets_list) ) logger.info(f"Successfully listed {len(files)} spreadsheets for {user_google_email}.") return text_output @server.tool() @handle_http_errors("get_spreadsheet_info", is_read_only=True, service_type="sheets") @require_google_service("sheets", "sheets_read") async def get_spreadsheet_info( service, user_google_email: str, spreadsheet_id: str, ) -> str: """ Gets information about a specific spreadsheet including its sheets. Args: user_google_email (str): The user's Google email address. Required. spreadsheet_id (str): The ID of the spreadsheet to get info for. Required. Returns: str: Formatted spreadsheet information including title and sheets list. """ logger.info(f"[get_spreadsheet_info] Invoked. Email: '{user_google_email}', Spreadsheet ID: {spreadsheet_id}") spreadsheet = await asyncio.to_thread( service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute ) title = spreadsheet.get("properties", {}).get("title", "Unknown") sheets = spreadsheet.get("sheets", []) sheets_info = [] for sheet in sheets: sheet_props = sheet.get("properties", {}) sheet_name = sheet_props.get("title", "Unknown") sheet_id = sheet_props.get("sheetId", "Unknown") grid_props = sheet_props.get("gridProperties", {}) rows = grid_props.get("rowCount", "Unknown") cols = grid_props.get("columnCount", "Unknown") sheets_info.append( f" - \"{sheet_name}\" (ID: {sheet_id}) | Size: {rows}x{cols}" ) text_output = ( f"Spreadsheet: \"{title}\" (ID: {spreadsheet_id})\n" f"Sheets ({len(sheets)}):\n" + "\n".join(sheets_info) if sheets_info else " No sheets found" ) logger.info(f"Successfully retrieved info for spreadsheet {spreadsheet_id} for {user_google_email}.") return text_output @server.tool() @handle_http_errors("read_sheet_values", is_read_only=True, service_type="sheets") @require_google_service("sheets", "sheets_read") async def read_sheet_values( service, user_google_email: str, spreadsheet_id: str, range_name: str = "A1:Z1000", ) -> str: """ Reads values from a specific range in a Google Sheet. Args: user_google_email (str): The user's Google email address. Required. spreadsheet_id (str): The ID of the spreadsheet. Required. range_name (str): The range to read (e.g., "Sheet1!A1:D10", "A1:D10"). Defaults to "A1:Z1000". Returns: str: The formatted values from the specified range. """ logger.info(f"[read_sheet_values] Invoked. Email: '{user_google_email}', Spreadsheet: {spreadsheet_id}, Range: {range_name}") result = await asyncio.to_thread( service.spreadsheets() .values() .get(spreadsheetId=spreadsheet_id, range=range_name) .execute ) values = result.get("values", []) if not values: return f"No data found in range '{range_name}' for {user_google_email}." # Format the output as a readable table formatted_rows = [] for i, row in enumerate(values, 1): # Pad row with empty strings to show structure padded_row = row + [""] * max(0, len(values[0]) - len(row)) if values else row formatted_rows.append(f"Row {i:2d}: {padded_row}") text_output = ( f"Successfully read {len(values)} rows from range '{range_name}' in spreadsheet {spreadsheet_id} for {user_google_email}:\n" + "\n".join(formatted_rows[:50]) # Limit to first 50 rows for readability + (f"\n... and {len(values) - 50} more rows" if len(values) > 50 else "") ) logger.info(f"Successfully read {len(values)} rows for {user_google_email}.") return text_output @server.tool() @handle_http_errors("modify_sheet_values", service_type="sheets") @require_google_service("sheets", "sheets_write") async def modify_sheet_values( service, user_google_email: str, spreadsheet_id: str, range_name: str, values: Optional[Union[str, List[List[str]]]] = None, value_input_option: str = "USER_ENTERED", clear_values: bool = False, ) -> str: """ Modifies values in a specific range of a Google Sheet - can write, update, or clear values. Args: user_google_email (str): The user's Google email address. Required. spreadsheet_id (str): The ID of the spreadsheet. Required. range_name (str): The range to modify (e.g., "Sheet1!A1:D10", "A1:D10"). Required. values (Optional[Union[str, List[List[str]]]]): 2D array of values to write/update. Can be a JSON string or Python list. Required unless clear_values=True. value_input_option (str): How to interpret input values ("RAW" or "USER_ENTERED"). Defaults to "USER_ENTERED". clear_values (bool): If True, clears the range instead of writing values. Defaults to False. Returns: str: Confirmation message of the successful modification operation. """ operation = "clear" if clear_values else "write" logger.info(f"[modify_sheet_values] Invoked. Operation: {operation}, Email: '{user_google_email}', Spreadsheet: {spreadsheet_id}, Range: {range_name}") # Parse values if it's a JSON string (MCP passes parameters as JSON strings) if values is not None and isinstance(values, str): try: parsed_values = json.loads(values) if not isinstance(parsed_values, list): raise ValueError(f"Values must be a list, got {type(parsed_values).__name__}") # Validate it's a list of lists for i, row in enumerate(parsed_values): if not isinstance(row, list): raise ValueError(f"Row {i} must be a list, got {type(row).__name__}") values = parsed_values logger.info(f"[modify_sheet_values] Parsed JSON string to Python list with {len(values)} rows") except json.JSONDecodeError as e: raise Exception(f"Invalid JSON format for values: {e}") except ValueError as e: raise Exception(f"Invalid values structure: {e}") if not clear_values and not values: raise Exception("Either 'values' must be provided or 'clear_values' must be True.") if clear_values: result = await asyncio.to_thread( service.spreadsheets() .values() .clear(spreadsheetId=spreadsheet_id, range=range_name) .execute ) cleared_range = result.get("clearedRange", range_name) text_output = f"Successfully cleared range '{cleared_range}' in spreadsheet {spreadsheet_id} for {user_google_email}." logger.info(f"Successfully cleared range '{cleared_range}' for {user_google_email}.") else: body = {"values": values} result = await asyncio.to_thread( service.spreadsheets() .values() .update( spreadsheetId=spreadsheet_id, range=range_name, valueInputOption=value_input_option, body=body, ) .execute ) updated_cells = result.get("updatedCells", 0) updated_rows = result.get("updatedRows", 0) updated_columns = result.get("updatedColumns", 0) text_output = ( f"Successfully updated range '{range_name}' in spreadsheet {spreadsheet_id} for {user_google_email}. " f"Updated: {updated_cells} cells, {updated_rows} rows, {updated_columns} columns." ) logger.info(f"Successfully updated {updated_cells} cells for {user_google_email}.") return text_output @server.tool() @handle_http_errors("create_spreadsheet", service_type="sheets") @require_google_service("sheets", "sheets_write") async def create_spreadsheet( service, user_google_email: str, title: str, sheet_names: Optional[List[str]] = None, ) -> str: """ Creates a new Google Spreadsheet. Args: user_google_email (str): The user's Google email address. Required. title (str): The title of the new spreadsheet. Required. sheet_names (Optional[List[str]]): List of sheet names to create. If not provided, creates one sheet with default name. Returns: str: Information about the newly created spreadsheet including ID and URL. """ logger.info(f"[create_spreadsheet] Invoked. Email: '{user_google_email}', Title: {title}") spreadsheet_body = { "properties": { "title": title } } if sheet_names: spreadsheet_body["sheets"] = [ {"properties": {"title": sheet_name}} for sheet_name in sheet_names ] spreadsheet = await asyncio.to_thread( service.spreadsheets().create(body=spreadsheet_body).execute ) spreadsheet_id = spreadsheet.get("spreadsheetId") spreadsheet_url = spreadsheet.get("spreadsheetUrl") text_output = ( f"Successfully created spreadsheet '{title}' for {user_google_email}. " f"ID: {spreadsheet_id} | URL: {spreadsheet_url}" ) logger.info(f"Successfully created spreadsheet for {user_google_email}. ID: {spreadsheet_id}") return text_output @server.tool() @handle_http_errors("create_sheet", service_type="sheets") @require_google_service("sheets", "sheets_write") async def create_sheet( service, user_google_email: str, spreadsheet_id: str, sheet_name: str, ) -> str: """ Creates a new sheet within an existing spreadsheet. Args: user_google_email (str): The user's Google email address. Required. spreadsheet_id (str): The ID of the spreadsheet. Required. sheet_name (str): The name of the new sheet. Required. Returns: str: Confirmation message of the successful sheet creation. """ logger.info(f"[create_sheet] Invoked. Email: '{user_google_email}', Spreadsheet: {spreadsheet_id}, Sheet: {sheet_name}") request_body = { "requests": [ { "addSheet": { "properties": { "title": sheet_name } } } ] } response = await asyncio.to_thread( service.spreadsheets() .batchUpdate(spreadsheetId=spreadsheet_id, body=request_body) .execute ) sheet_id = response["replies"][0]["addSheet"]["properties"]["sheetId"] text_output = ( f"Successfully created sheet '{sheet_name}' (ID: {sheet_id}) in spreadsheet {spreadsheet_id} for {user_google_email}." ) logger.info(f"Successfully created sheet for {user_google_email}. Sheet ID: {sheet_id}") return text_output # Create comment management tools for sheets _comment_tools = create_comment_tools("spreadsheet", "spreadsheet_id") # Extract and register the functions read_sheet_comments = _comment_tools['read_comments'] create_sheet_comment = _comment_tools['create_comment'] reply_to_sheet_comment = _comment_tools['reply_to_comment'] resolve_sheet_comment = _comment_tools['resolve_comment']

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/taylorwilsdon/google_workspace_mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server