jason/google-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #47 from taylorwilsdon/sheets_support

Browse Source 
feat: Full Google Sheets support
This commit is contained in:
Taylor Wilsdon
2025-06-06 13:16:32 -04:00
committed by GitHub
parent d4a07dcef5 047c44b472
commit 21dccf5915
7 changed files with 593 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
README.md
View File

@@ -42,6 +42,7 @@
- [📁 Google Drive](#-google-drive)
- [📧 Gmail](#-gmail)
- [📝 Google Docs](#-google-docs)
- [📊 Google Sheets](#-google-sheets)
- [💬 Google Chat](#-google-chat)
- [🛠️ Development](#️-development)
- [Project Structure](#project-structure)
@@ -56,7 +57,7 @@
## 🌐 Overview
The Google Workspace MCP Server integrates Google Workspace services (Calendar, Drive, Gmail, and Docs) with AI assistants and other applications using the Model Context Protocol (MCP). This allows AI systems to access and interact with user data from Google Workspace applications securely and efficiently.
The Google Workspace MCP Server integrates Google Workspace services (Calendar, Drive, Gmail, Docs, Sheets, and Chat) with AI assistants and other applications using the Model Context Protocol (MCP). This allows AI systems to access and interact with user data from Google Workspace applications securely and efficiently.
---
@@ -67,6 +68,7 @@ The Google Workspace MCP Server integrates Google Workspace services (Calendar,
- **📁 Google Drive Integration**: Search files, list folder contents, read file content, and create new files. Supports extraction and retrieval of .docx, .xlsx and other Microsoft Office formats natively!
- **📧 Gmail Integration**: Complete email management - search messages, retrieve content, send emails, and create drafts with full support for all query syntax
- **📄 Google Docs Integration**: Search for documents, read document content, list documents in folders, and create new documents right from your chat!
- **📊 Google Sheets Integration**: Complete spreadsheet management - list spreadsheets, read/write/clear cell values, create spreadsheets and sheets, with flexible value modification
- **🔄 Multiple Transport Options**: Streamable HTTP + SSE fallback
- **🔌 `mcpo` Compatibility**: Easily expose the server as an OpenAPI endpoint for integration with tools like Open WebUI
- **🧩 Extensible Design**: Simple structure for adding support for more Google Workspace APIs and tools
@@ -81,7 +83,7 @@ The Google Workspace MCP Server integrates Google Workspace services (Calendar,
- **Python 3.11+**
- **[uv](https://github.com/astral-sh/uv)** package installer (or pip)
- **Google Cloud Project** with OAuth 2.0 credentials enabled for required APIs (Calendar, Drive, Gmail, Docs)
- **Google Cloud Project** with OAuth 2.0 credentials enabled for required APIs (Calendar, Drive, Gmail, Docs, Sheets, Chat)
### Installation
@@ -100,7 +102,7 @@ uv pip install -e .
### Configuration
1. Create **OAuth 2.0 Credentials** (web application type) in the [Google Cloud Console](https://console.cloud.google.com/).
2. Enable the **Google Calendar API**, **Google Drive API**, **Gmail API**, and **Google Docs API** for your project.
2. Enable the **Google Calendar API**, **Google Drive API**, **Gmail API**, **Google Docs API**, and **Google Sheets API** for your project.
3. Download the OAuth client credentials as `client_secret.json` and place it in the project's root directory.
4. Add the following redirect URI to your OAuth client configuration in the Google Cloud Console. Note that `http://localhost:8000` is the default base URI and port, which can be customized via environment variables (`WORKSPACE_MCP_BASE_URI` and `WORKSPACE_MCP_PORT`). If you change these, you must update the redirect URI in the Google Cloud Console accordingly.
```
@@ -345,6 +347,30 @@ Source: [`gmail/gmail_tools.py`](gmail/gmail_tools.py)
Source: [`gdocs/docs_tools.py`](gdocs/docs_tools.py)
| Tool | Description | Parameters |
|----------------------|-------------------------------------------------------------------------------------|------------|
| `search_docs` | Search for Google Docs by name (using Drive API). | • `query` (required): Text to search for in Doc names<br>• `user_google_email` (optional)<br>• `page_size` (optional, default: 10)<br>• `mcp_session_id` (injected automatically) |
| `get_doc_content` | Retrieve the plain text content of a Google Doc by its document ID. | • `document_id` (required)<br>• `user_google_email` (optional)<br>• `mcp_session_id` (injected automatically) |
| `list_docs_in_folder`| List all Google Docs inside a given Drive folder (by folder ID, default = `root`). | • `folder_id` (optional, default: `'root'`)<br>• `user_google_email` (optional)<br>• `page_size` (optional, default: 100)<br>• `mcp_session_id` (injected automatically) |
| `create_doc` | Create a new Google Doc, optionally with initial content. | • `title` (required): Name for the doc<br>• `content` (optional, default: empty)<br>• `user_google_email` (optional)<br>• `mcp_session_id` (injected automatically) |
### 📊 Google Sheets
Source: [`gsheets/sheets_tools.py`](gsheets/sheets_tools.py)
| Tool | Description | Parameters |
|------|-------------|------------|
| `list_spreadsheets` | Lists spreadsheets from Google Drive that the user has access to. | • `user_google_email` (required): The user's Google email address<br>• `max_results` (optional, default: 25): Maximum number of spreadsheets to return |
| `get_spreadsheet_info` | Gets information about a specific spreadsheet including its sheets. | • `user_google_email` (required): The user's Google email address<br>• `spreadsheet_id` (required): The ID of the spreadsheet to get info for |
| `read_sheet_values` | Reads values from a specific range in a Google Sheet. | • `user_google_email` (required): The user's Google email address<br>• `spreadsheet_id` (required): The ID of the spreadsheet<br>• `range_name` (optional, default: "A1:Z1000"): The range to read (e.g., "Sheet1!A1:D10", "A1:D10") |
| `modify_sheet_values` | Modifies values in a specific range of a Google Sheet - can write, update, or clear values. | • `user_google_email` (required): The user's Google email address<br>• `spreadsheet_id` (required): The ID of the spreadsheet<br>• `range_name` (required): The range to modify<br>• `values` (optional): 2D array of values to write/update. Required unless clear_values=True<br>• `value_input_option` (optional, default: "USER_ENTERED"): How to interpret input values ("RAW" or "USER_ENTERED")<br>• `clear_values` (optional, default: False): If True, clears the range instead of writing values |
| `create_spreadsheet` | Creates a new Google Spreadsheet. | • `user_google_email` (required): The user's Google email address<br>• `title` (required): The title of the new spreadsheet<br>• `sheet_names` (optional): List of sheet names to create. If not provided, creates one sheet with default name |
| `create_sheet` | Creates a new sheet within an existing spreadsheet. | • `user_google_email` (required): The user's Google email address<br>• `spreadsheet_id` (required): The ID of the spreadsheet<br>• `sheet_name` (required): The name of the new sheet |
> ℹ️ All Sheets tools require `user_google_email` for authentication. If authentication fails or is required, the tool will return an error prompting the LLM to use the centralized `start_google_auth` tool with the user's email and `service_name="Google Sheets"`.
> 📊 **Sheet Operations**: The `modify_sheet_values` tool consolidates write, update, and clear operations into a single flexible function. Use `clear_values=True` to clear a range, or provide `values` to write/update data.
### 💬 Google Chat
Source: [`gchat/chat_tools.py`](gchat/chat_tools.py)
@@ -357,12 +383,6 @@ Source: [`gchat/chat_tools.py`](gchat/chat_tools.py)
| `search_messages`| Searches for messages across Chat spaces by text content. | • `user_google_email` (required)<br>• `query` (required): Text to search for<br>• `space_id` (optional): If provided, searches only in this space<br>• `page_size` (optional, default: 25) |
> ℹ️ All Chat tools require `user_google_email` for authentication. If authentication fails or is required, the tool will return an error prompting the LLM to use the centralized `start_google_auth` tool with the user's email and `service_name="Google Chat"`.
| Tool | Description | Parameters |
|----------------------|-------------------------------------------------------------------------------------|------------|
| `search_docs` | Search for Google Docs by name (using Drive API). | • `query` (required): Text to search for in Doc names<br>• `user_google_email` (optional)<br>• `page_size` (optional, default: 10)<br>• `mcp_session_id` (injected automatically) |
| `get_doc_content` | Retrieve the plain text content of a Google Doc by its document ID. | • `document_id` (required)<br>• `user_google_email` (optional)<br>• `mcp_session_id` (injected automatically) |
| `list_docs_in_folder`| List all Google Docs inside a given Drive folder (by folder ID, default = `root`). | • `folder_id` (optional, default: `'root'`)<br>• `user_google_email` (optional)<br>• `page_size` (optional, default: 100)<br>• `mcp_session_id` (injected automatically) |
| `create_doc` | Create a new Google Doc, optionally with initial content. | • `title` (required): Name for the doc<br>• `content` (optional, default: empty)<br>• `user_google_email` (optional)<br>• `mcp_session_id` (injected automatically) |
---
@@ -379,6 +399,7 @@ google_workspace_mcp/
├── gdocs/ # Google Docs tools (docs_tools.py)
├── gdrive/ # Google Drive tools (drive_tools.py)
├── gmail/ # Gmail tools (gmail_tools.py)
├── gsheets/ # Google Sheets tools (sheets_tools.py)
├── .gitignore # Git ignore file
├── client_secret.json # Google OAuth Credentials (DO NOT COMMIT)
├── config.json # Example mcpo configuration
@@ -472,7 +493,7 @@ if not credentials or not credentials.valid:
- Running `mcpo` behind a reverse proxy (like Nginx or Caddy) to handle HTTPS termination, proper logging, and more robust authentication
- Binding `mcpo` only to trusted network interfaces if exposing it beyond localhost
- **Scope Management**: The server requests specific OAuth scopes (permissions) for Calendar, Drive, and Gmail. Users grant access based on these scopes during the initial authentication. Do not request broader scopes than necessary for the implemented tools.
- **Scope Management**: The server requests specific OAuth scopes (permissions) for Calendar, Drive, Gmail, Docs, Sheets, and Chat. Users grant access based on these scopes during the initial authentication. Do not request broader scopes than necessary for the implemented tools.
---

12
config/google_config.py
View File

@@ -39,6 +39,10 @@ CHAT_READONLY_SCOPE = 'https://www.googleapis.com/auth/chat.messages.readonly'
CHAT_WRITE_SCOPE = 'https://www.googleapis.com/auth/chat.messages'
CHAT_SPACES_SCOPE = 'https://www.googleapis.com/auth/chat.spaces'
# Google Sheets API scopes
SHEETS_READONLY_SCOPE = 'https://www.googleapis.com/auth/spreadsheets.readonly'
SHEETS_WRITE_SCOPE = 'https://www.googleapis.com/auth/spreadsheets'
# Base OAuth scopes required for user identification
BASE_SCOPES = [
USERINFO_EMAIL_SCOPE,
@@ -79,5 +83,11 @@ CHAT_SCOPES = [
CHAT_SPACES_SCOPE
]
# Sheets-specific scopes
SHEETS_SCOPES = [
SHEETS_READONLY_SCOPE,
SHEETS_WRITE_SCOPE
]
# Combined scopes for all supported Google Workspace operations
SCOPES = list(set(BASE_SCOPES + CALENDAR_SCOPES + DRIVE_SCOPES + GMAIL_SCOPES + DOCS_SCOPES + CHAT_SCOPES))
SCOPES = list(set(BASE_SCOPES + CALENDAR_SCOPES + DRIVE_SCOPES + GMAIL_SCOPES + DOCS_SCOPES + CHAT_SCOPES + SHEETS_SCOPES))

3
core/server.py
View File

@@ -36,6 +36,9 @@ from config.google_config import (
CHAT_WRITE_SCOPE,
CHAT_SPACES_SCOPE,
CHAT_SCOPES,
SHEETS_READONLY_SCOPE,
SHEETS_WRITE_SCOPE,
SHEETS_SCOPES,
SCOPES
)

13
gmail/gmail_tools.py
View File

@@ -11,7 +11,6 @@ from typing import Optional, List, Dict, Literal
from email.mime.text import MIMEText
from mcp import types
from fastapi import Body
from googleapiclient.errors import HttpError
@@ -797,10 +796,10 @@ async def list_gmail_labels(
)
lines = [f"Found {len(labels)} labels:", ""]
system_labels = []
user_labels = []
for label in labels:
if label.get("type") == "system":
system_labels.append(label)
@@ -867,7 +866,7 @@ async def manage_gmail_label(
isError=True,
content=[types.TextContent(type="text", text="Label name is required for create action.")],
)
if action in ["update", "delete"] and not label_id:
return types.CallToolResult(
isError=True,
@@ -908,14 +907,14 @@ async def manage_gmail_label(
current_label = await asyncio.to_thread(
service.users().labels().get(userId="me", id=label_id).execute
)
label_object = {
"id": label_id,
"name": name if name is not None else current_label["name"],
"labelListVisibility": label_list_visibility,
"messageListVisibility": message_list_visibility,
}
updated_label = await asyncio.to_thread(
service.users().labels().update(userId="me", id=label_id, body=label_object).execute
)
@@ -933,7 +932,7 @@ async def manage_gmail_label(
service.users().labels().get(userId="me", id=label_id).execute
)
label_name = label["name"]
await asyncio.to_thread(
service.users().labels().delete(userId="me", id=label_id).execute
)

23
gsheets/__init__.py Normal file
View File

@@ -0,0 +1,23 @@
"""
Google Sheets MCP Integration
This module provides MCP tools for interacting with Google Sheets API.
"""
from .sheets_tools import (
list_spreadsheets,
get_spreadsheet_info,
read_sheet_values,
modify_sheet_values,
create_spreadsheet,
create_sheet,
)
__all__ = [
"list_spreadsheets",
"get_spreadsheet_info",
"read_sheet_values",
"modify_sheet_values",
"create_spreadsheet",
"create_sheet",
]

517
gsheets/sheets_tools.py Normal file
View File

@@ -0,0 +1,517 @@
"""
Google Sheets MCP Tools
This module provides MCP tools for interacting with Google Sheets API.
"""
import logging
import asyncio
from typing import List, Optional
from mcp import types
from googleapiclient.errors import HttpError
from auth.google_auth import get_authenticated_google_service
from core.server import server
from config.google_config import SHEETS_READONLY_SCOPE, SHEETS_WRITE_SCOPE
# Configure module logger
logger = logging.getLogger(__name__)
@server.tool()
async def list_spreadsheets(
user_google_email: str,
max_results: int = 25,
) -> types.CallToolResult:
"""
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:
types.CallToolResult: Contains a list of spreadsheet files (name, ID, modified time),
an error message if the API call fails,
or an authentication guidance message if credentials are required.
"""
tool_name = "list_spreadsheets"
logger.info(f"[{tool_name}] Invoked. Email: '{user_google_email}'")
auth_result = await get_authenticated_google_service(
service_name="drive",
version="v3",
tool_name=tool_name,
user_google_email=user_google_email,
required_scopes=[SHEETS_READONLY_SCOPE],
)
if isinstance(auth_result, types.CallToolResult):
return auth_result
service, user_email = auth_result
try:
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 types.CallToolResult(
content=[
types.TextContent(
type="text", text=f"No spreadsheets found for {user_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_email}:\n"
+ "\n".join(spreadsheets_list)
)
logger.info(f"Successfully listed {len(files)} spreadsheets for {user_email}.")
return types.CallToolResult(
content=[types.TextContent(type="text", text=text_output)]
)
except HttpError as error:
message = f"API error listing spreadsheets: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with user's email and service_name='Google Sheets'."
logger.error(message, exc_info=True)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
except Exception as e:
message = f"Unexpected error listing spreadsheets: {e}."
logger.exception(message)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
@server.tool()
async def get_spreadsheet_info(
user_google_email: str,
spreadsheet_id: str,
) -> types.CallToolResult:
"""
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:
types.CallToolResult: Contains spreadsheet information (title, sheets list),
an error message if the API call fails,
or an authentication guidance message if credentials are required.
"""
tool_name = "get_spreadsheet_info"
logger.info(f"[{tool_name}] Invoked. Email: '{user_google_email}', Spreadsheet ID: {spreadsheet_id}")
auth_result = await get_authenticated_google_service(
service_name="sheets",
version="v4",
tool_name=tool_name,
user_google_email=user_google_email,
required_scopes=[SHEETS_READONLY_SCOPE],
)
if isinstance(auth_result, types.CallToolResult):
return auth_result
service, user_email = auth_result
try:
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_email}.")
return types.CallToolResult(
content=[types.TextContent(type="text", text=text_output)]
)
except HttpError as error:
message = f"API error getting spreadsheet info: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with user's email and service_name='Google Sheets'."
logger.error(message, exc_info=True)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
except Exception as e:
message = f"Unexpected error getting spreadsheet info: {e}."
logger.exception(message)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
@server.tool()
async def read_sheet_values(
user_google_email: str,
spreadsheet_id: str,
range_name: str = "A1:Z1000",
) -> types.CallToolResult:
"""
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:
types.CallToolResult: Contains the values from the specified range,
an error message if the API call fails,
or an authentication guidance message if credentials are required.
"""
tool_name = "read_sheet_values"
logger.info(f"[{tool_name}] Invoked. Email: '{user_google_email}', Spreadsheet: {spreadsheet_id}, Range: {range_name}")
auth_result = await get_authenticated_google_service(
service_name="sheets",
version="v4",
tool_name=tool_name,
user_google_email=user_google_email,
required_scopes=[SHEETS_READONLY_SCOPE],
)
if isinstance(auth_result, types.CallToolResult):
return auth_result
service, user_email = auth_result
try:
result = await asyncio.to_thread(
service.spreadsheets()
.values()
.get(spreadsheetId=spreadsheet_id, range=range_name)
.execute
)
values = result.get("values", [])
if not values:
return types.CallToolResult(
content=[
types.TextContent(
type="text", text=f"No data found in range '{range_name}' for {user_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_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_email}.")
return types.CallToolResult(
content=[types.TextContent(type="text", text=text_output)]
)
except HttpError as error:
message = f"API error reading sheet values: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with user's email and service_name='Google Sheets'."
logger.error(message, exc_info=True)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
except Exception as e:
message = f"Unexpected error reading sheet values: {e}."
logger.exception(message)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
@server.tool()
async def modify_sheet_values(
user_google_email: str,
spreadsheet_id: str,
range_name: str,
values: Optional[List[List[str]]] = None,
value_input_option: str = "USER_ENTERED",
clear_values: bool = False,
) -> types.CallToolResult:
"""
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[List[List[str]]]): 2D array of values to write/update. 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:
types.CallToolResult: Confirms successful modification operation,
an error message if the API call fails,
or an authentication guidance message if credentials are required.
"""
tool_name = "modify_sheet_values"
operation = "clear" if clear_values else "write"
logger.info(f"[{tool_name}] Invoked. Operation: {operation}, Email: '{user_google_email}', Spreadsheet: {spreadsheet_id}, Range: {range_name}")
if not clear_values and not values:
message = "Either 'values' must be provided or 'clear_values' must be True."
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
auth_result = await get_authenticated_google_service(
service_name="sheets",
version="v4",
tool_name=tool_name,
user_google_email=user_google_email,
required_scopes=[SHEETS_WRITE_SCOPE],
)
if isinstance(auth_result, types.CallToolResult):
return auth_result
service, user_email = auth_result
try:
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_email}."
logger.info(f"Successfully cleared range '{cleared_range}' for {user_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_email}. "
f"Updated: {updated_cells} cells, {updated_rows} rows, {updated_columns} columns."
)
logger.info(f"Successfully updated {updated_cells} cells for {user_email}.")
return types.CallToolResult(
content=[types.TextContent(type="text", text=text_output)]
)
except HttpError as error:
message = f"API error modifying sheet values: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with user's email and service_name='Google Sheets'."
logger.error(message, exc_info=True)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
except Exception as e:
message = f"Unexpected error modifying sheet values: {e}."
logger.exception(message)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
@server.tool()
async def create_spreadsheet(
user_google_email: str,
title: str,
sheet_names: Optional[List[str]] = None,
) -> types.CallToolResult:
"""
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:
types.CallToolResult: Contains the new spreadsheet information (ID, URL),
an error message if the API call fails,
or an authentication guidance message if credentials are required.
"""
tool_name = "create_spreadsheet"
logger.info(f"[{tool_name}] Invoked. Email: '{user_google_email}', Title: {title}")
auth_result = await get_authenticated_google_service(
service_name="sheets",
version="v4",
tool_name=tool_name,
user_google_email=user_google_email,
required_scopes=[SHEETS_WRITE_SCOPE],
)
if isinstance(auth_result, types.CallToolResult):
return auth_result
service, user_email = auth_result
try:
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_email}. "
f"ID: {spreadsheet_id} | URL: {spreadsheet_url}"
)
logger.info(f"Successfully created spreadsheet for {user_email}. ID: {spreadsheet_id}")
return types.CallToolResult(
content=[types.TextContent(type="text", text=text_output)]
)
except HttpError as error:
message = f"API error creating spreadsheet: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with user's email and service_name='Google Sheets'."
logger.error(message, exc_info=True)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
except Exception as e:
message = f"Unexpected error creating spreadsheet: {e}."
logger.exception(message)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
@server.tool()
async def create_sheet(
user_google_email: str,
spreadsheet_id: str,
sheet_name: str,
) -> types.CallToolResult:
"""
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:
types.CallToolResult: Confirms successful sheet creation,
an error message if the API call fails,
or an authentication guidance message if credentials are required.
"""
tool_name = "create_sheet"
logger.info(f"[{tool_name}] Invoked. Email: '{user_google_email}', Spreadsheet: {spreadsheet_id}, Sheet: {sheet_name}")
auth_result = await get_authenticated_google_service(
service_name="sheets",
version="v4",
tool_name=tool_name,
user_google_email=user_google_email,
required_scopes=[SHEETS_WRITE_SCOPE],
)
if isinstance(auth_result, types.CallToolResult):
return auth_result
service, user_email = auth_result
try:
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_email}."
)
logger.info(f"Successfully created sheet for {user_email}. Sheet ID: {sheet_id}")
return types.CallToolResult(
content=[types.TextContent(type="text", text=text_output)]
)
except HttpError as error:
message = f"API error creating sheet: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with user's email and service_name='Google Sheets'."
logger.error(message, exc_info=True)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)
except Exception as e:
message = f"Unexpected error creating sheet: {e}."
logger.exception(message)
return types.CallToolResult(
isError=True, content=[types.TextContent(type="text", text=message)]
)

3
main.py
View File

@@ -31,12 +31,13 @@ try:
except Exception as e:
sys.stderr.write(f"CRITICAL: Failed to set up file logging to '{log_file_path}': {e}\n")
# Import calendar tools to register them with the MCP server via decorators
# Import tool modules to register them with the MCP server via decorators
import gcalendar.calendar_tools
import gdrive.drive_tools
import gmail.gmail_tools
import gdocs.docs_tools
import gchat.chat_tools
import gsheets.sheets_tools
def main():