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

Merge branch 'main' of https://github.com/taylorwilsdon/google_workspace_mcp into fix/pkce-code-verifier

Browse Source
This commit is contained in:
Taylor Wilsdon
2026-03-01 17:34:12 -05:00
parent 8463e4fd29 4d17c83057
commit 662383b8b2
20 changed files with 1712 additions and 2179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
README.md
View File

@@ -846,9 +846,7 @@ cp .env.oauth21 .env
|------|------|-------------| |------|------|-------------|
| `list_calendars` | **Core** | List accessible calendars | | `list_calendars` | **Core** | List accessible calendars |
| `get_events` | **Core** | Retrieve events with time range filtering | | `get_events` | **Core** | Retrieve events with time range filtering |
| `create_event` | **Core** | Create events with attachments & reminders | | `manage_event` | **Core** | Create, update, or delete calendar events |
| `modify_event` | **Core** | Update existing events |
| `delete_event` | Extended | Remove events |
</td> </td>
<td width="50%" valign="top"> <td width="50%" valign="top">
@@ -863,15 +861,11 @@ cp .env.oauth21 .env
| `create_drive_file` | **Core** | Create files or fetch from URLs | | `create_drive_file` | **Core** | Create files or fetch from URLs |
| `create_drive_folder` | **Core** | Create empty folders in Drive or shared drives | | `create_drive_folder` | **Core** | Create empty folders in Drive or shared drives |
| `import_to_google_doc` | **Core** | Import files (MD, DOCX, HTML, etc.) as Google Docs | | `import_to_google_doc` | **Core** | Import files (MD, DOCX, HTML, etc.) as Google Docs |
| `share_drive_file` | **Core** | Share file with users/groups/domains/anyone |
| `get_drive_shareable_link` | **Core** | Get shareable links for a file | | `get_drive_shareable_link` | **Core** | Get shareable links for a file |
| `list_drive_items` | Extended | List folder contents | | `list_drive_items` | Extended | List folder contents |
| `copy_drive_file` | Extended | Copy existing files (templates) with optional renaming | | `copy_drive_file` | Extended | Copy existing files (templates) with optional renaming |
| `update_drive_file` | Extended | Update file metadata, move between folders | | `update_drive_file` | Extended | Update file metadata, move between folders |
| `batch_share_drive_file` | Extended | Share file with multiple recipients | | `manage_drive_access` | Extended | Grant, update, revoke permissions, and transfer ownership |
| `update_drive_permission` | Extended | Modify permission role |
| `remove_drive_permission` | Extended | Revoke file access |
| `transfer_drive_ownership` | Extended | Transfer file ownership to another user |
| `set_drive_file_permissions` | Extended | Set link sharing and file-level sharing settings | | `set_drive_file_permissions` | Extended | Set link sharing and file-level sharing settings |
| `get_drive_file_permissions` | Complete | Get detailed file permissions | | `get_drive_file_permissions` | Complete | Get detailed file permissions |
| `check_drive_file_public_access` | Complete | Check public sharing status | | `check_drive_file_public_access` | Complete | Check public sharing status |
@@ -894,7 +888,9 @@ cp .env.oauth21 .env
| `get_gmail_thread_content` | Extended | Get full thread content | | `get_gmail_thread_content` | Extended | Get full thread content |
| `modify_gmail_message_labels` | Extended | Modify message labels | | `modify_gmail_message_labels` | Extended | Modify message labels |
| `list_gmail_labels` | Extended | List available labels | | `list_gmail_labels` | Extended | List available labels |
| `list_gmail_filters` | Extended | List Gmail filters |
| `manage_gmail_label` | Extended | Create/update/delete labels | | `manage_gmail_label` | Extended | Create/update/delete labels |
| `manage_gmail_filter` | Extended | Create or delete Gmail filters |
| `draft_gmail_message` | Extended | Create drafts | | `draft_gmail_message` | Extended | Create drafts |
| `get_gmail_threads_content_batch` | Complete | Batch retrieve thread content | | `get_gmail_threads_content_batch` | Complete | Batch retrieve thread content |
| `batch_modify_gmail_message_labels` | Complete | Batch modify labels | | `batch_modify_gmail_message_labels` | Complete | Batch modify labels |
@@ -965,7 +961,8 @@ Saved files expire after 1 hour and are cleaned up automatically.
| `export_doc_to_pdf` | Extended | Export document to PDF | | `export_doc_to_pdf` | Extended | Export document to PDF |
| `create_table_with_data` | Complete | Create data tables | | `create_table_with_data` | Complete | Create data tables |
| `debug_table_structure` | Complete | Debug table issues | | `debug_table_structure` | Complete | Debug table issues |
| `*_document_comments` | Complete | Read, Reply, Create, Resolve | | `list_document_comments` | Complete | List all document comments |
| `manage_document_comment` | Complete | Create, reply to, or resolve comments |
</td> </td>
</tr> </tr>
@@ -984,7 +981,9 @@ Saved files expire after 1 hour and are cleaned up automatically.
| `get_spreadsheet_info` | Extended | Get spreadsheet metadata | | `get_spreadsheet_info` | Extended | Get spreadsheet metadata |
| `format_sheet_range` | Extended | Apply colors, number formats, text wrapping, alignment, bold/italic, font size | | `format_sheet_range` | Extended | Apply colors, number formats, text wrapping, alignment, bold/italic, font size |
| `create_sheet` | Complete | Add sheets to existing files | | `create_sheet` | Complete | Add sheets to existing files |
| `*_sheet_comment` | Complete | Read/create/reply/resolve comments | | `list_spreadsheet_comments` | Complete | List all spreadsheet comments |
| `manage_spreadsheet_comment` | Complete | Create, reply to, or resolve comments |
| `manage_conditional_formatting` | Complete | Add, update, or delete conditional formatting rules |
</td> </td>
<td width="50%" valign="top"> <td width="50%" valign="top">
@@ -998,7 +997,8 @@ Saved files expire after 1 hour and are cleaned up automatically.
| `batch_update_presentation` | Extended | Apply multiple updates | | `batch_update_presentation` | Extended | Apply multiple updates |
| `get_page` | Extended | Get specific slide information | | `get_page` | Extended | Get specific slide information |
| `get_page_thumbnail` | Extended | Generate slide thumbnails | | `get_page_thumbnail` | Extended | Generate slide thumbnails |
| `*_presentation_comment` | Complete | Read/create/reply/resolve comments | | `list_presentation_comments` | Complete | List all presentation comments |
| `manage_presentation_comment` | Complete | Create, reply to, or resolve comments |
</td> </td>
</tr> </tr>
@@ -1025,12 +1025,10 @@ Saved files expire after 1 hour and are cleaned up automatically.
|------|------|-------------| |------|------|-------------|
| `list_tasks` | **Core** | List tasks with filtering | | `list_tasks` | **Core** | List tasks with filtering |
| `get_task` | **Core** | Retrieve task details | | `get_task` | **Core** | Retrieve task details |
| `create_task` | **Core** | Create tasks with hierarchy | | `manage_task` | **Core** | Create, update, delete, or move tasks |
| `update_task` | **Core** | Modify task properties | | `list_task_lists` | Complete | List task lists |
| `delete_task` | Extended | Remove tasks | | `get_task_list` | Complete | Get task list details |
| `move_task` | Complete | Reposition tasks | | `manage_task_list` | Complete | Create, update, delete task lists, or clear completed tasks |
| `clear_completed_tasks` | Complete | Hide completed tasks |
| `*_task_list` | Complete | List/get/create/update/delete task lists |
</td> </td>
</tr> </tr>
@@ -1044,14 +1042,11 @@ Saved files expire after 1 hour and are cleaned up automatically.
| `search_contacts` | **Core** | Search contacts by name, email, phone | | `search_contacts` | **Core** | Search contacts by name, email, phone |
| `get_contact` | **Core** | Retrieve detailed contact info | | `get_contact` | **Core** | Retrieve detailed contact info |
| `list_contacts` | **Core** | List contacts with pagination | | `list_contacts` | **Core** | List contacts with pagination |
| `create_contact` | **Core** | Create new contacts | | `manage_contact` | **Core** | Create, update, or delete contacts |
| `update_contact` | Extended | Update existing contacts |
| `delete_contact` | Extended | Delete contacts |
| `list_contact_groups` | Extended | List contact groups/labels | | `list_contact_groups` | Extended | List contact groups/labels |
| `get_contact_group` | Extended | Get group details with members | | `get_contact_group` | Extended | Get group details with members |
| `batch_*_contacts` | Complete | Batch create/update/delete contacts | | `manage_contacts_batch` | Complete | Batch create, update, or delete contacts |
| `*_contact_group` | Complete | Create/update/delete contact groups | | `manage_contact_group` | Complete | Create, update, delete groups, or modify membership |
| `modify_contact_group_members` | Complete | Add/remove contacts from groups |
</td> </td>
</tr> </tr>
@@ -1076,9 +1071,8 @@ Saved files expire after 1 hour and are cleaned up automatically.
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
| `search_custom` | **Core** | Perform web searches | | `search_custom` | **Core** | Perform web searches (supports site restrictions via sites parameter) |
| `get_search_engine_info` | Complete | Retrieve search engine metadata | | `get_search_engine_info` | Complete | Retrieve search engine metadata |
| `search_custom_siterestrict` | Extended | Search within specific domains |
</td> </td>
</tr> </tr>
@@ -1095,10 +1089,8 @@ Saved files expire after 1 hour and are cleaned up automatically.
| `create_script_project` | **Core** | Create new standalone or bound project | | `create_script_project` | **Core** | Create new standalone or bound project |
| `update_script_content` | **Core** | Update or create script files | | `update_script_content` | **Core** | Update or create script files |
| `run_script_function` | **Core** | Execute function with parameters | | `run_script_function` | **Core** | Execute function with parameters |
| `create_deployment` | Extended | Create new script deployment |
| `list_deployments` | Extended | List all project deployments | | `list_deployments` | Extended | List all project deployments |
| `update_deployment` | Extended | Update deployment configuration | | `manage_deployment` | Extended | Create, update, or delete script deployments |
| `delete_deployment` | Extended | Remove deployment |
| `list_script_processes` | Extended | View recent executions and status | | `list_script_processes` | Extended | View recent executions and status |
</td> </td>

87
README_NEW.md
View File

@@ -47,7 +47,7 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
## 🛠 Tools Reference ## 🛠 Tools Reference
### Gmail (11 tools) ### Gmail (10 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
@@ -60,39 +60,42 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `list_gmail_labels` | Extended | List all system and user labels | | `list_gmail_labels` | Extended | List all system and user labels |
| `manage_gmail_label` | Extended | Create, update, delete labels | | `manage_gmail_label` | Extended | Create, update, delete labels |
| `modify_gmail_message_labels` | Extended | Add/remove labels (archive, trash, etc.) | | `modify_gmail_message_labels` | Extended | Add/remove labels (archive, trash, etc.) |
| `manage_gmail_filter` | Extended | Create or delete Gmail filters |
| `get_gmail_threads_content_batch` | Complete | Batch retrieve threads | | `get_gmail_threads_content_batch` | Complete | Batch retrieve threads |
| `batch_modify_gmail_message_labels` | Complete | Bulk label operations | | `batch_modify_gmail_message_labels` | Complete | Bulk label operations |
**Also includes:** `get_gmail_attachment_content`, `list_gmail_filters`, `create_gmail_filter`, `delete_gmail_filter` **Also includes:** `get_gmail_attachment_content`, `list_gmail_filters`
### Google Drive (7 tools) ### Google Drive (10 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
| `search_drive_files` | Core | Search files with Drive query syntax or free text | | `search_drive_files` | Core | Search files with Drive query syntax or free text |
| `get_drive_file_content` | Core | Read content from Docs, Sheets, Office files (.docx, .xlsx, .pptx) | | `get_drive_file_content` | Core | Read content from Docs, Sheets, Office files (.docx, .xlsx, .pptx) |
| `get_drive_file_download_url` | Core | Download Drive files to local disk |
| `create_drive_file` | Core | Create files from content or URL (supports file://, http://, https://) | | `create_drive_file` | Core | Create files from content or URL (supports file://, http://, https://) |
| `create_drive_folder` | Core | Create empty folders in Drive or shared drives | | `create_drive_folder` | Core | Create empty folders in Drive or shared drives |
| `import_to_google_doc` | Core | Import files (MD, DOCX, HTML, etc.) as Google Docs |
| `get_drive_shareable_link` | Core | Get shareable links for a file |
| `list_drive_items` | Extended | List folder contents with shared drive support | | `list_drive_items` | Extended | List folder contents with shared drive support |
| `copy_drive_file` | Extended | Copy existing files (templates) with optional renaming |
| `update_drive_file` | Extended | Update metadata, move between folders, star, trash | | `update_drive_file` | Extended | Update metadata, move between folders, star, trash |
| `get_drive_file_permissions` | Complete | Check sharing status and permissions | | `manage_drive_access` | Extended | Grant, update, revoke permissions, and transfer ownership |
| `set_drive_file_permissions` | Extended | Set link sharing and file-level sharing settings |
| `get_drive_file_permissions` | Complete | Get detailed file permissions |
| `check_drive_file_public_access` | Complete | Verify public link sharing for Docs image insertion | | `check_drive_file_public_access` | Complete | Verify public link sharing for Docs image insertion |
**Also includes:** `get_drive_file_download_url` for generating download URLs ### Google Calendar (3 tools)
### Google Calendar (5 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
| `list_calendars` | Core | List all accessible calendars | | `list_calendars` | Core | List all accessible calendars |
| `get_events` | Core | Query events by time range, search, or specific ID | | `get_events` | Core | Query events by time range, search, or specific ID |
| `create_event` | Core | Create events with attendees, reminders, Google Meet, attachments | | `manage_event` | Core | Create, update, or delete calendar events |
| `modify_event` | Core | Update any event property including conferencing |
| `delete_event` | Extended | Remove events |
**Event features:** Timezone support, transparency (busy/free), visibility settings, up to 5 custom reminders **Event features:** Timezone support, transparency (busy/free), visibility settings, up to 5 custom reminders, Google Meet integration, attendees, attachments
### Google Docs (16 tools) ### Google Docs (14 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
@@ -103,6 +106,8 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `find_and_replace_doc` | Extended | Global find/replace with case matching | | `find_and_replace_doc` | Extended | Global find/replace with case matching |
| `list_docs_in_folder` | Extended | List Docs in a specific folder | | `list_docs_in_folder` | Extended | List Docs in a specific folder |
| `insert_doc_elements` | Extended | Add tables, lists, page breaks | | `insert_doc_elements` | Extended | Add tables, lists, page breaks |
| `update_paragraph_style` | Extended | Apply heading styles, lists (bulleted/numbered with nesting), and paragraph formatting |
| `get_doc_as_markdown` | Extended | Export document as formatted Markdown with optional comments |
| `export_doc_to_pdf` | Extended | Export to PDF and save to Drive | | `export_doc_to_pdf` | Extended | Export to PDF and save to Drive |
| `insert_doc_image` | Complete | Insert images from Drive or URLs | | `insert_doc_image` | Complete | Insert images from Drive or URLs |
| `update_doc_headers_footers` | Complete | Modify headers/footers | | `update_doc_headers_footers` | Complete | Modify headers/footers |
@@ -110,10 +115,10 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `inspect_doc_structure` | Complete | Analyze document structure for safe insertion points | | `inspect_doc_structure` | Complete | Analyze document structure for safe insertion points |
| `create_table_with_data` | Complete | Create and populate tables in one operation | | `create_table_with_data` | Complete | Create and populate tables in one operation |
| `debug_table_structure` | Complete | Debug table cell positions and content | | `debug_table_structure` | Complete | Debug table cell positions and content |
| `list_document_comments` | Complete | List all document comments |
| `manage_document_comment` | Complete | Create, reply to, or resolve comments |
**Comments:** `read_document_comments`, `create_document_comment`, `reply_to_document_comment`, `resolve_document_comment` ### Google Sheets (9 tools)
### Google Sheets (13 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
@@ -124,13 +129,11 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `get_spreadsheet_info` | Extended | Get metadata, sheets, conditional formats | | `get_spreadsheet_info` | Extended | Get metadata, sheets, conditional formats |
| `format_sheet_range` | Extended | Apply colors, number formats, text wrapping, alignment, bold/italic, font size | | `format_sheet_range` | Extended | Apply colors, number formats, text wrapping, alignment, bold/italic, font size |
| `create_sheet` | Complete | Add sheets to existing spreadsheets | | `create_sheet` | Complete | Add sheets to existing spreadsheets |
| `add_conditional_formatting` | Complete | Add boolean or gradient rules | | `list_spreadsheet_comments` | Complete | List all spreadsheet comments |
| `update_conditional_formatting` | Complete | Modify existing rules | | `manage_spreadsheet_comment` | Complete | Create, reply to, or resolve comments |
| `delete_conditional_formatting` | Complete | Remove formatting rules | | `manage_conditional_formatting` | Complete | Add, update, or delete conditional formatting rules |
**Comments:** `read_spreadsheet_comments`, `create_spreadsheet_comment`, `reply_to_spreadsheet_comment`, `resolve_spreadsheet_comment` ### Google Slides (7 tools)
### Google Slides (9 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
@@ -139,8 +142,8 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `batch_update_presentation` | Extended | Apply multiple updates (create slides, shapes, etc.) | | `batch_update_presentation` | Extended | Apply multiple updates (create slides, shapes, etc.) |
| `get_page` | Extended | Get specific slide details and elements | | `get_page` | Extended | Get specific slide details and elements |
| `get_page_thumbnail` | Extended | Generate PNG thumbnails | | `get_page_thumbnail` | Extended | Generate PNG thumbnails |
| `list_presentation_comments` | Complete | List all presentation comments |
**Comments:** `read_presentation_comments`, `create_presentation_comment`, `reply_to_presentation_comment`, `resolve_presentation_comment` | `manage_presentation_comment` | Complete | Create, reply to, or resolve comments |
### Google Forms (6 tools) ### Google Forms (6 tools)
@@ -153,24 +156,18 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `get_form_response` | Complete | Get individual response details | | `get_form_response` | Complete | Get individual response details |
| `batch_update_form` | Complete | Execute batch updates to forms (questions, items, settings) | | `batch_update_form` | Complete | Execute batch updates to forms (questions, items, settings) |
### Google Tasks (12 tools) ### Google Tasks (5 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
| `list_tasks` | Core | List tasks with filtering, subtask hierarchy preserved | | `list_tasks` | Core | List tasks with filtering, subtask hierarchy preserved |
| `get_task` | Core | Get task details | | `get_task` | Core | Get task details |
| `create_task` | Core | Create tasks with notes, due dates, parent/sibling positioning | | `manage_task` | Core | Create, update, delete, or move tasks |
| `update_task` | Core | Update task properties |
| `delete_task` | Extended | Remove tasks |
| `list_task_lists` | Complete | List all task lists | | `list_task_lists` | Complete | List all task lists |
| `get_task_list` | Complete | Get task list details | | `get_task_list` | Complete | Get task list details |
| `create_task_list` | Complete | Create new task lists | | `manage_task_list` | Complete | Create, update, delete task lists, or clear completed tasks |
| `update_task_list` | Complete | Rename task lists |
| `delete_task_list` | Complete | Delete task lists (and all tasks) |
| `move_task` | Complete | Reposition or move between lists |
| `clear_completed_tasks` | Complete | Hide completed tasks |
### Google Apps Script (11 tools) ### Google Apps Script (9 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
@@ -180,16 +177,27 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `create_script_project` | Core | Create new standalone or bound project | | `create_script_project` | Core | Create new standalone or bound project |
| `update_script_content` | Core | Update or create script files | | `update_script_content` | Core | Update or create script files |
| `run_script_function` | Core | Execute function with parameters | | `run_script_function` | Core | Execute function with parameters |
| `create_deployment` | Extended | Create new script deployment |
| `list_deployments` | Extended | List all project deployments | | `list_deployments` | Extended | List all project deployments |
| `update_deployment` | Extended | Update deployment configuration | | `manage_deployment` | Extended | Create, update, or delete script deployments |
| `delete_deployment` | Extended | Remove deployment |
| `list_script_processes` | Extended | View recent executions and status | | `list_script_processes` | Extended | View recent executions and status |
**Enables:** Cross-app automation, persistent workflows, custom business logic execution, script development and debugging **Enables:** Cross-app automation, persistent workflows, custom business logic execution, script development and debugging
**Note:** Trigger management is not currently supported via MCP tools. **Note:** Trigger management is not currently supported via MCP tools.
### Google Contacts (7 tools)
| Tool | Tier | Description |
|------|------|-------------|
| `search_contacts` | Core | Search contacts by name, email, phone |
| `get_contact` | Core | Retrieve detailed contact info |
| `list_contacts` | Core | List contacts with pagination |
| `manage_contact` | Core | Create, update, or delete contacts |
| `list_contact_groups` | Extended | List contact groups/labels |
| `get_contact_group` | Extended | Get group details with members |
| `manage_contacts_batch` | Complete | Batch create, update, or delete contacts |
| `manage_contact_group` | Complete | Create, update, delete groups, or modify membership |
### Google Chat (4 tools) ### Google Chat (4 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
@@ -199,12 +207,11 @@ export OAUTHLIB_INSECURE_TRANSPORT=1 # Development only
| `search_messages` | Core | Search across chat history | | `search_messages` | Core | Search across chat history |
| `list_spaces` | Extended | List rooms and DMs | | `list_spaces` | Extended | List rooms and DMs |
### Google Custom Search (3 tools) ### Google Custom Search (2 tools)
| Tool | Tier | Description | | Tool | Tier | Description |
|------|------|-------------| |------|------|-------------|
| `search_custom` | Core | Web search with filters (date, file type, language, safe search) | | `search_custom` | Core | Web search with filters (date, file type, language, safe search, site restrictions via sites parameter) |
| `search_custom_siterestrict` | Extended | Search within specific domains |
| `get_search_engine_info` | Complete | Get search engine metadata | | `get_search_engine_info` | Complete | Get search engine metadata |
**Requires:** `GOOGLE_PSE_API_KEY` and `GOOGLE_PSE_ENGINE_ID` environment variables **Requires:** `GOOGLE_PSE_API_KEY` and `GOOGLE_PSE_ENGINE_ID` environment variables
@@ -219,7 +226,7 @@ Choose a tier based on your needs:
|------|-------|----------| |------|-------|----------|
| **Core** | ~30 | Essential operations: search, read, create, send | | **Core** | ~30 | Essential operations: search, read, create, send |
| **Extended** | ~50 | Core + management: labels, folders, batch ops | | **Extended** | ~50 | Core + management: labels, folders, batch ops |
| **Complete** | ~80 | Full API: comments, headers, admin functions | | **Complete** | 111 | Full API: comments, headers, admin functions |
```bash ```bash
uvx workspace-mcp --tool-tier core # Start minimal uvx workspace-mcp --tool-tier core # Start minimal

203
core/comments.py
View File

@@ -7,7 +7,7 @@ All Google Workspace apps (Docs, Sheets, Slides) use the Drive API for comment o
import logging import logging
import asyncio import asyncio
from typing import Optional
from auth.service_decorator import require_google_service from auth.service_decorator import require_google_service
from core.server import server from core.server import server
@@ -16,6 +16,38 @@ from core.utils import handle_http_errors
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
async def _manage_comment_dispatch(
service,
app_name: str,
file_id: str,
action: str,
comment_content: Optional[str] = None,
comment_id: Optional[str] = None,
) -> str:
"""Route comment management actions to the appropriate implementation."""
action_lower = action.lower().strip()
if action_lower == "create":
if not comment_content:
raise ValueError("comment_content is required for create action")
return await _create_comment_impl(service, app_name, file_id, comment_content)
elif action_lower == "reply":
if not comment_id or not comment_content:
raise ValueError(
"comment_id and comment_content are required for reply action"
)
return await _reply_to_comment_impl(
service, app_name, file_id, comment_id, comment_content
)
elif action_lower == "resolve":
if not comment_id:
raise ValueError("comment_id is required for resolve action")
return await _resolve_comment_impl(service, app_name, file_id, comment_id)
else:
raise ValueError(
f"Invalid action '{action_lower}'. Must be 'create', 'reply', or 'resolve'."
)
def create_comment_tools(app_name: str, file_id_param: str): def create_comment_tools(app_name: str, file_id_param: str):
""" """
Factory function to create comment management tools for a specific Google Workspace app. Factory function to create comment management tools for a specific Google Workspace app.
@@ -25,165 +57,114 @@ def create_comment_tools(app_name: str, file_id_param: str):
file_id_param: Parameter name for the file ID (e.g., "document_id", "spreadsheet_id", "presentation_id") file_id_param: Parameter name for the file ID (e.g., "document_id", "spreadsheet_id", "presentation_id")
Returns: Returns:
Dict containing the four comment management functions with unique names Dict containing the comment management functions with unique names
""" """
# Create unique function names based on the app type # --- Consolidated tools ---
read_func_name = f"read_{app_name}_comments" list_func_name = f"list_{app_name}_comments"
create_func_name = f"create_{app_name}_comment" manage_func_name = f"manage_{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": if file_id_param == "document_id":
@require_google_service("drive", "drive_read") @require_google_service("drive", "drive_read")
@handle_http_errors(read_func_name, service_type="drive") @handle_http_errors(list_func_name, service_type="drive")
async def read_comments( async def list_comments(
service, user_google_email: str, document_id: str service, user_google_email: str, document_id: str
) -> str: ) -> str:
"""Read all comments from a Google Document.""" """List all comments from a Google Document."""
return await _read_comments_impl(service, app_name, document_id) return await _read_comments_impl(service, app_name, document_id)
@require_google_service("drive", "drive_file") @require_google_service("drive", "drive_file")
@handle_http_errors(create_func_name, service_type="drive") @handle_http_errors(manage_func_name, service_type="drive")
async def create_comment( async def manage_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, service,
user_google_email: str, user_google_email: str,
document_id: str, document_id: str,
comment_id: str, action: str,
reply_content: str, comment_content: Optional[str] = None,
comment_id: Optional[str] = None,
) -> str: ) -> str:
"""Reply to a specific comment in a Google Document.""" """Manage comments on a Google Document.
return await _reply_to_comment_impl(
service, app_name, document_id, comment_id, reply_content
)
@require_google_service("drive", "drive_file") Actions:
@handle_http_errors(resolve_func_name, service_type="drive") - create: Create a new comment. Requires comment_content.
async def resolve_comment( - reply: Reply to a comment. Requires comment_id and comment_content.
service, user_google_email: str, document_id: str, comment_id: str - resolve: Resolve a comment. Requires comment_id.
) -> str: """
"""Resolve a comment in a Google Document.""" return await _manage_comment_dispatch(
return await _resolve_comment_impl( service, app_name, document_id, action, comment_content, comment_id
service, app_name, document_id, comment_id
) )
elif file_id_param == "spreadsheet_id": elif file_id_param == "spreadsheet_id":
@require_google_service("drive", "drive_read") @require_google_service("drive", "drive_read")
@handle_http_errors(read_func_name, service_type="drive") @handle_http_errors(list_func_name, service_type="drive")
async def read_comments( async def list_comments(
service, user_google_email: str, spreadsheet_id: str service, user_google_email: str, spreadsheet_id: str
) -> str: ) -> str:
"""Read all comments from a Google Spreadsheet.""" """List all comments from a Google Spreadsheet."""
return await _read_comments_impl(service, app_name, spreadsheet_id) return await _read_comments_impl(service, app_name, spreadsheet_id)
@require_google_service("drive", "drive_file") @require_google_service("drive", "drive_file")
@handle_http_errors(create_func_name, service_type="drive") @handle_http_errors(manage_func_name, service_type="drive")
async def create_comment( async def manage_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, service,
user_google_email: str, user_google_email: str,
spreadsheet_id: str, spreadsheet_id: str,
comment_id: str, action: str,
reply_content: str, comment_content: Optional[str] = None,
comment_id: Optional[str] = None,
) -> str: ) -> str:
"""Reply to a specific comment in a Google Spreadsheet.""" """Manage comments on a Google Spreadsheet.
return await _reply_to_comment_impl(
service, app_name, spreadsheet_id, comment_id, reply_content
)
@require_google_service("drive", "drive_file") Actions:
@handle_http_errors(resolve_func_name, service_type="drive") - create: Create a new comment. Requires comment_content.
async def resolve_comment( - reply: Reply to a comment. Requires comment_id and comment_content.
service, user_google_email: str, spreadsheet_id: str, comment_id: str - resolve: Resolve a comment. Requires comment_id.
) -> str: """
"""Resolve a comment in a Google Spreadsheet.""" return await _manage_comment_dispatch(
return await _resolve_comment_impl( service, app_name, spreadsheet_id, action, comment_content, comment_id
service, app_name, spreadsheet_id, comment_id
) )
elif file_id_param == "presentation_id": elif file_id_param == "presentation_id":
@require_google_service("drive", "drive_read") @require_google_service("drive", "drive_read")
@handle_http_errors(read_func_name, service_type="drive") @handle_http_errors(list_func_name, service_type="drive")
async def read_comments( async def list_comments(
service, user_google_email: str, presentation_id: str service, user_google_email: str, presentation_id: str
) -> str: ) -> str:
"""Read all comments from a Google Presentation.""" """List all comments from a Google Presentation."""
return await _read_comments_impl(service, app_name, presentation_id) return await _read_comments_impl(service, app_name, presentation_id)
@require_google_service("drive", "drive_file") @require_google_service("drive", "drive_file")
@handle_http_errors(create_func_name, service_type="drive") @handle_http_errors(manage_func_name, service_type="drive")
async def create_comment( async def manage_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, service,
user_google_email: str, user_google_email: str,
presentation_id: str, presentation_id: str,
comment_id: str, action: str,
reply_content: str, comment_content: Optional[str] = None,
comment_id: Optional[str] = None,
) -> str: ) -> str:
"""Reply to a specific comment in a Google Presentation.""" """Manage comments on a Google Presentation.
return await _reply_to_comment_impl(
service, app_name, presentation_id, comment_id, reply_content Actions:
- create: Create a new comment. Requires comment_content.
- reply: Reply to a comment. Requires comment_id and comment_content.
- resolve: Resolve a comment. Requires comment_id.
"""
return await _manage_comment_dispatch(
service, app_name, presentation_id, action, comment_content, comment_id
) )
@require_google_service("drive", "drive_file") list_comments.__name__ = list_func_name
@handle_http_errors(resolve_func_name, service_type="drive") manage_comment.__name__ = manage_func_name
async def resolve_comment( server.tool()(list_comments)
service, user_google_email: str, presentation_id: str, comment_id: str server.tool()(manage_comment)
) -> 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 { return {
"read_comments": read_comments, "list_comments": list_comments,
"create_comment": create_comment, "manage_comment": manage_comment,
"reply_to_comment": reply_to_comment,
"resolve_comment": resolve_comment,
} }

64
core/tool_tiers.yaml
View File

@@ -13,8 +13,7 @@ gmail:
- manage_gmail_label - manage_gmail_label
- draft_gmail_message - draft_gmail_message
- list_gmail_filters - list_gmail_filters
- create_gmail_filter - manage_gmail_filter
- delete_gmail_filter
complete: complete:
- get_gmail_threads_content_batch - get_gmail_threads_content_batch
@@ -29,16 +28,12 @@ drive:
- create_drive_file - create_drive_file
- create_drive_folder - create_drive_folder
- import_to_google_doc - import_to_google_doc
- share_drive_file
- get_drive_shareable_link - get_drive_shareable_link
extended: extended:
- list_drive_items - list_drive_items
- copy_drive_file - copy_drive_file
- update_drive_file - update_drive_file
- update_drive_permission - manage_drive_access
- remove_drive_permission
- transfer_drive_ownership
- batch_share_drive_file
- set_drive_file_permissions - set_drive_file_permissions
complete: complete:
- get_drive_file_permissions - get_drive_file_permissions
@@ -48,10 +43,8 @@ calendar:
core: core:
- list_calendars - list_calendars
- get_events - get_events
- create_event - manage_event
- modify_event
extended: extended:
- delete_event
- query_freebusy - query_freebusy
complete: [] complete: []
@@ -75,10 +68,8 @@ docs:
- inspect_doc_structure - inspect_doc_structure
- create_table_with_data - create_table_with_data
- debug_table_structure - debug_table_structure
- read_document_comments - list_document_comments
- create_document_comment - manage_document_comment
- reply_to_document_comment
- resolve_document_comment
sheets: sheets:
core: core:
@@ -91,10 +82,9 @@ sheets:
- format_sheet_range - format_sheet_range
complete: complete:
- create_sheet - create_sheet
- read_spreadsheet_comments - list_spreadsheet_comments
- create_spreadsheet_comment - manage_spreadsheet_comment
- reply_to_spreadsheet_comment - manage_conditional_formatting
- resolve_spreadsheet_comment
chat: chat:
core: core:
@@ -127,53 +117,37 @@ slides:
- get_page - get_page
- get_page_thumbnail - get_page_thumbnail
complete: complete:
- read_presentation_comments - list_presentation_comments
- create_presentation_comment - manage_presentation_comment
- reply_to_presentation_comment
- resolve_presentation_comment
tasks: tasks:
core: core:
- get_task - get_task
- list_tasks - list_tasks
- create_task - manage_task
- update_task extended: []
extended:
- delete_task
complete: complete:
- list_task_lists - list_task_lists
- get_task_list - get_task_list
- create_task_list - manage_task_list
- update_task_list
- delete_task_list
- move_task
- clear_completed_tasks
contacts: contacts:
core: core:
- search_contacts - search_contacts
- get_contact - get_contact
- list_contacts - list_contacts
- create_contact - manage_contact
extended: extended:
- update_contact
- delete_contact
- list_contact_groups - list_contact_groups
- get_contact_group - get_contact_group
complete: complete:
- batch_create_contacts - manage_contacts_batch
- batch_update_contacts - manage_contact_group
- batch_delete_contacts
- create_contact_group
- update_contact_group
- delete_contact_group
- modify_contact_group_members
search: search:
core: core:
- search_custom - search_custom
extended: extended: []
- search_custom_siterestrict
complete: complete:
- get_search_engine_info - get_search_engine_info
@@ -187,10 +161,8 @@ appscript:
- run_script_function - run_script_function
- generate_trigger_code - generate_trigger_code
extended: extended:
- create_deployment - manage_deployment
- list_deployments - list_deployments
- update_deployment
- delete_deployment
- delete_script_project - delete_script_project
- list_versions - list_versions
- create_version - create_version

5
core/utils.py
View File

@@ -2,7 +2,6 @@ import io
import logging import logging
import os import os
import zipfile import zipfile
import xml.etree.ElementTree as ET
import ssl import ssl
import asyncio import asyncio
import functools import functools
@@ -10,6 +9,8 @@ import functools
from pathlib import Path from pathlib import Path
from typing import List, Optional from typing import List, Optional
from defusedxml import ElementTree as ET
from googleapiclient.errors import HttpError from googleapiclient.errors import HttpError
from .api_enablement import get_api_enablement_message from .api_enablement import get_api_enablement_message
from auth.google_auth import GoogleAuthenticationError from auth.google_auth import GoogleAuthenticationError
@@ -226,7 +227,7 @@ def extract_office_xml_text(file_bytes: bytes, mime_type: str) -> Optional[str]:
""" """
Very light-weight XML scraper for Word, Excel, PowerPoint files. Very light-weight XML scraper for Word, Excel, PowerPoint files.
Returns plain-text if something readable is found, else None. Returns plain-text if something readable is found, else None.
No external deps – just std-lib zipfile + ElementTree. Uses zipfile + defusedxml.ElementTree.
""" """
shared_strings: List[str] = [] shared_strings: List[str] = []
ns_excel_main = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" ns_excel_main = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"

100
gappsscript/apps_script_tools.py
View File

@@ -464,31 +464,57 @@ async def _create_deployment_impl(
@server.tool() @server.tool()
@handle_http_errors("create_deployment", service_type="script") @handle_http_errors("manage_deployment", service_type="script")
@require_google_service("script", "script_deployments") @require_google_service("script", "script_deployments")
async def create_deployment( async def manage_deployment(
service: Any, service: Any,
user_google_email: str, user_google_email: str,
action: str,
script_id: str, script_id: str,
description: str, deployment_id: Optional[str] = None,
description: Optional[str] = None,
version_description: Optional[str] = None, version_description: Optional[str] = None,
) -> str: ) -> str:
""" """
Creates a new deployment of the script. Manages Apps Script deployments. Supports creating, updating, and deleting deployments.
Args: Args:
service: Injected Google API service client service: Injected Google API service client
user_google_email: User's email address user_google_email: User's email address
action: Action to perform - "create", "update", or "delete"
script_id: The script project ID script_id: The script project ID
description: Deployment description deployment_id: The deployment ID (required for update and delete)
version_description: Optional version description description: Deployment description (required for create and update)
version_description: Optional version description (for create only)
Returns: Returns:
str: Formatted string with deployment details str: Formatted string with deployment details or confirmation
""" """
return await _create_deployment_impl( action = action.lower().strip()
service, user_google_email, script_id, description, version_description if action == "create":
) if description is None or description.strip() == "":
raise ValueError("description is required for create action")
return await _create_deployment_impl(
service, user_google_email, script_id, description, version_description
)
elif action == "update":
if not deployment_id:
raise ValueError("deployment_id is required for update action")
if description is None or description.strip() == "":
raise ValueError("description is required for update action")
return await _update_deployment_impl(
service, user_google_email, script_id, deployment_id, description
)
elif action == "delete":
if not deployment_id:
raise ValueError("deployment_id is required for delete action")
return await _delete_deployment_impl(
service, user_google_email, script_id, deployment_id
)
else:
raise ValueError(
f"Invalid action '{action}'. Must be 'create', 'update', or 'delete'."
)
async def _list_deployments_impl( async def _list_deployments_impl(
@@ -578,34 +604,6 @@ async def _update_deployment_impl(
return "\n".join(output) return "\n".join(output)
@server.tool()
@handle_http_errors("update_deployment", service_type="script")
@require_google_service("script", "script_deployments")
async def update_deployment(
service: Any,
user_google_email: str,
script_id: str,
deployment_id: str,
description: Optional[str] = None,
) -> str:
"""
Updates an existing deployment configuration.
Args:
service: Injected Google API service client
user_google_email: User's email address
script_id: The script project ID
deployment_id: The deployment ID to update
description: Optional new description
Returns:
str: Formatted string confirming update
"""
return await _update_deployment_impl(
service, user_google_email, script_id, deployment_id, description
)
async def _delete_deployment_impl( async def _delete_deployment_impl(
service: Any, service: Any,
user_google_email: str, user_google_email: str,
@@ -630,32 +628,6 @@ async def _delete_deployment_impl(
return output return output
@server.tool()
@handle_http_errors("delete_deployment", service_type="script")
@require_google_service("script", "script_deployments")
async def delete_deployment(
service: Any,
user_google_email: str,
script_id: str,
deployment_id: str,
) -> str:
"""
Deletes a deployment.
Args:
service: Injected Google API service client
user_google_email: User's email address
script_id: The script project ID
deployment_id: The deployment ID to delete
Returns:
str: Confirmation message
"""
return await _delete_deployment_impl(
service, user_google_email, script_id, deployment_id
)
async def _list_script_processes_impl( async def _list_script_processes_impl(
service: Any, service: Any,
user_google_email: str, user_google_email: str,

229
gcalendar/calendar_tools.py
View File

@@ -534,10 +534,14 @@ async def get_events(
return text_output return text_output
@server.tool() # ---------------------------------------------------------------------------
@handle_http_errors("create_event", service_type="calendar") # Internal implementation functions for event create/modify/delete.
@require_google_service("calendar", "calendar_events") # These are called by both the consolidated ``manage_event`` tool and the
async def create_event( # legacy single-action tools.
# ---------------------------------------------------------------------------
async def _create_event_impl(
service, service,
user_google_email: str, user_google_email: str,
summary: str, summary: str,
@@ -558,32 +562,7 @@ async def create_event(
guests_can_invite_others: Optional[bool] = None, guests_can_invite_others: Optional[bool] = None,
guests_can_see_other_guests: Optional[bool] = None, guests_can_see_other_guests: Optional[bool] = None,
) -> str: ) -> str:
""" """Internal implementation for creating a calendar event."""
Creates a new event.
Args:
user_google_email (str): The user's Google email address. Required.
summary (str): Event title.
start_time (str): Start time (RFC3339, e.g., "2023-10-27T10:00:00-07:00" or "2023-10-27" for all-day).
end_time (str): End time (RFC3339, e.g., "2023-10-27T11:00:00-07:00" or "2023-10-28" for all-day).
calendar_id (str): Calendar ID (default: 'primary').
description (Optional[str]): Event description.
location (Optional[str]): Event location.
attendees (Optional[List[str]]): Attendee email addresses.
timezone (Optional[str]): Timezone (e.g., "America/New_York").
attachments (Optional[List[str]]): List of Google Drive file URLs or IDs to attach to the event.
add_google_meet (bool): Whether to add a Google Meet video conference to the event. Defaults to False.
reminders (Optional[Union[str, List[Dict[str, Any]]]]): JSON string or list of reminder objects. Each should have 'method' ("popup" or "email") and 'minutes' (0-40320). Max 5 reminders. Example: '[{"method": "popup", "minutes": 15}]' or [{"method": "popup", "minutes": 15}]
use_default_reminders (bool): Whether to use calendar's default reminders. If False, uses custom reminders. Defaults to True.
transparency (Optional[str]): Event transparency for busy/free status. "opaque" shows as Busy (default), "transparent" shows as Available/Free. Defaults to None (uses Google Calendar default).
visibility (Optional[str]): Event visibility. "default" uses calendar default, "public" is visible to all, "private" is visible only to attendees, "confidential" is same as private (legacy). Defaults to None (uses Google Calendar default).
guests_can_modify (Optional[bool]): Whether attendees other than the organizer can modify the event. Defaults to None (uses Google Calendar default of False).
guests_can_invite_others (Optional[bool]): Whether attendees other than the organizer can invite others to the event. Defaults to None (uses Google Calendar default of True).
guests_can_see_other_guests (Optional[bool]): Whether attendees other than the organizer can see who the event's attendees are. Defaults to None (uses Google Calendar default of True).
Returns:
str: Confirmation message of the successful event creation with event link.
"""
logger.info( logger.info(
f"[create_event] Invoked. Email: '{user_google_email}', Summary: {summary}" f"[create_event] Invoked. Email: '{user_google_email}', Summary: {summary}"
) )
@@ -809,10 +788,7 @@ def _normalize_attendees(
return normalized if normalized else None return normalized if normalized else None
@server.tool() async def _modify_event_impl(
@handle_http_errors("modify_event", service_type="calendar")
@require_google_service("calendar", "calendar_events")
async def modify_event(
service, service,
user_google_email: str, user_google_email: str,
event_id: str, event_id: str,
@@ -834,33 +810,7 @@ async def modify_event(
guests_can_invite_others: Optional[bool] = None, guests_can_invite_others: Optional[bool] = None,
guests_can_see_other_guests: Optional[bool] = None, guests_can_see_other_guests: Optional[bool] = None,
) -> str: ) -> str:
""" """Internal implementation for modifying a calendar event."""
Modifies an existing event.
Args:
user_google_email (str): The user's Google email address. Required.
event_id (str): The ID of the event to modify.
calendar_id (str): Calendar ID (default: 'primary').
summary (Optional[str]): New event title.
start_time (Optional[str]): New start time (RFC3339, e.g., "2023-10-27T10:00:00-07:00" or "2023-10-27" for all-day).
end_time (Optional[str]): New end time (RFC3339, e.g., "2023-10-27T11:00:00-07:00" or "2023-10-28" for all-day).
description (Optional[str]): New event description.
location (Optional[str]): New event location.
attendees (Optional[Union[List[str], List[Dict[str, Any]]]]): Attendees as email strings or objects with metadata. Supports: ["email@example.com"] or [{"email": "email@example.com", "responseStatus": "accepted", "organizer": true, "optional": true}]. When using objects, existing metadata (responseStatus, organizer, optional) is preserved. New attendees default to responseStatus="needsAction".
timezone (Optional[str]): New timezone (e.g., "America/New_York").
add_google_meet (Optional[bool]): Whether to add or remove Google Meet video conference. If True, adds Google Meet; if False, removes it; if None, leaves unchanged.
reminders (Optional[Union[str, List[Dict[str, Any]]]]): JSON string or list of reminder objects to replace existing reminders. Each should have 'method' ("popup" or "email") and 'minutes' (0-40320). Max 5 reminders. Example: '[{"method": "popup", "minutes": 15}]' or [{"method": "popup", "minutes": 15}]
use_default_reminders (Optional[bool]): Whether to use calendar's default reminders. If specified, overrides current reminder settings.
transparency (Optional[str]): Event transparency for busy/free status. "opaque" shows as Busy, "transparent" shows as Available/Free. If None, preserves existing transparency setting.
visibility (Optional[str]): Event visibility. "default" uses calendar default, "public" is visible to all, "private" is visible only to attendees, "confidential" is same as private (legacy). If None, preserves existing visibility setting.
color_id (Optional[str]): Event color ID (1-11). If None, preserves existing color.
guests_can_modify (Optional[bool]): Whether attendees other than the organizer can modify the event. If None, preserves existing setting.
guests_can_invite_others (Optional[bool]): Whether attendees other than the organizer can invite others to the event. If None, preserves existing setting.
guests_can_see_other_guests (Optional[bool]): Whether attendees other than the organizer can see who the event's attendees are. If None, preserves existing setting.
Returns:
str: Confirmation message of the successful event modification with event link.
"""
logger.info( logger.info(
f"[modify_event] Invoked. Email: '{user_google_email}', Event ID: {event_id}" f"[modify_event] Invoked. Email: '{user_google_email}', Event ID: {event_id}"
) )
@@ -1075,23 +1025,13 @@ async def modify_event(
return confirmation_message return confirmation_message
@server.tool() async def _delete_event_impl(
@handle_http_errors("delete_event", service_type="calendar") service,
@require_google_service("calendar", "calendar_events") user_google_email: str,
async def delete_event( event_id: str,
service, user_google_email: str, event_id: str, calendar_id: str = "primary" calendar_id: str = "primary",
) -> str: ) -> str:
""" """Internal implementation for deleting a calendar event."""
Deletes an existing event.
Args:
user_google_email (str): The user's Google email address. Required.
event_id (str): The ID of the event to delete.
calendar_id (str): Calendar ID (default: 'primary').
Returns:
str: Confirmation message of the successful event deletion.
"""
logger.info( logger.info(
f"[delete_event] Invoked. Email: '{user_google_email}', Event ID: {event_id}" f"[delete_event] Invoked. Email: '{user_google_email}', Event ID: {event_id}"
) )
@@ -1133,6 +1073,141 @@ async def delete_event(
return confirmation_message return confirmation_message
# ---------------------------------------------------------------------------
# Consolidated event management tool
# ---------------------------------------------------------------------------
@server.tool()
@handle_http_errors("manage_event", service_type="calendar")
@require_google_service("calendar", "calendar_events")
async def manage_event(
service,
user_google_email: str,
action: str,
summary: Optional[str] = None,
start_time: Optional[str] = None,
end_time: Optional[str] = None,
event_id: Optional[str] = None,
calendar_id: str = "primary",
description: Optional[str] = None,
location: Optional[str] = None,
attendees: Optional[Union[List[str], List[Dict[str, Any]]]] = None,
timezone: Optional[str] = None,
attachments: Optional[List[str]] = None,
add_google_meet: Optional[bool] = None,
reminders: Optional[Union[str, List[Dict[str, Any]]]] = None,
use_default_reminders: Optional[bool] = None,
transparency: Optional[str] = None,
visibility: Optional[str] = None,
color_id: Optional[str] = None,
guests_can_modify: Optional[bool] = None,
guests_can_invite_others: Optional[bool] = None,
guests_can_see_other_guests: Optional[bool] = None,
) -> str:
"""
Manages calendar events. Supports creating, updating, and deleting events.
Args:
user_google_email (str): The user's Google email address. Required.
action (str): Action to perform - "create", "update", or "delete".
summary (Optional[str]): Event title (required for create).
start_time (Optional[str]): Start time in RFC3339 format (required for create).
end_time (Optional[str]): End time in RFC3339 format (required for create).
event_id (Optional[str]): Event ID (required for update and delete).
calendar_id (str): Calendar ID (default: 'primary').
description (Optional[str]): Event description.
location (Optional[str]): Event location.
attendees (Optional[Union[List[str], List[Dict[str, Any]]]]): Attendee email addresses or objects.
timezone (Optional[str]): Timezone (e.g., "America/New_York").
attachments (Optional[List[str]]): List of Google Drive file URLs or IDs to attach.
add_google_meet (Optional[bool]): Whether to add/remove Google Meet.
reminders (Optional[Union[str, List[Dict[str, Any]]]]): Custom reminder objects.
use_default_reminders (Optional[bool]): Whether to use default reminders.
transparency (Optional[str]): "opaque" (busy) or "transparent" (free).
visibility (Optional[str]): "default", "public", "private", or "confidential".
color_id (Optional[str]): Event color ID (1-11, update only).
guests_can_modify (Optional[bool]): Whether attendees can modify.
guests_can_invite_others (Optional[bool]): Whether attendees can invite others.
guests_can_see_other_guests (Optional[bool]): Whether attendees can see other guests.
Returns:
str: Confirmation message with event details.
"""
action_lower = action.lower().strip()
if action_lower == "create":
if not summary or not start_time or not end_time:
raise ValueError(
"summary, start_time, and end_time are required for create action"
)
return await _create_event_impl(
service=service,
user_google_email=user_google_email,
summary=summary,
start_time=start_time,
end_time=end_time,
calendar_id=calendar_id,
description=description,
location=location,
attendees=attendees,
timezone=timezone,
attachments=attachments,
add_google_meet=add_google_meet or False,
reminders=reminders,
use_default_reminders=use_default_reminders
if use_default_reminders is not None
else True,
transparency=transparency,
visibility=visibility,
guests_can_modify=guests_can_modify,
guests_can_invite_others=guests_can_invite_others,
guests_can_see_other_guests=guests_can_see_other_guests,
)
elif action_lower == "update":
if not event_id:
raise ValueError("event_id is required for update action")
return await _modify_event_impl(
service=service,
user_google_email=user_google_email,
event_id=event_id,
calendar_id=calendar_id,
summary=summary,
start_time=start_time,
end_time=end_time,
description=description,
location=location,
attendees=attendees,
timezone=timezone,
add_google_meet=add_google_meet,
reminders=reminders,
use_default_reminders=use_default_reminders,
transparency=transparency,
visibility=visibility,
color_id=color_id,
guests_can_modify=guests_can_modify,
guests_can_invite_others=guests_can_invite_others,
guests_can_see_other_guests=guests_can_see_other_guests,
)
elif action_lower == "delete":
if not event_id:
raise ValueError("event_id is required for delete action")
return await _delete_event_impl(
service=service,
user_google_email=user_google_email,
event_id=event_id,
calendar_id=calendar_id,
)
else:
raise ValueError(
f"Invalid action '{action_lower}'. Must be 'create', 'update', or 'delete'."
)
# ---------------------------------------------------------------------------
# Legacy single-action tools (deprecated -- prefer ``manage_event``)
# ---------------------------------------------------------------------------
@server.tool() @server.tool()
@handle_http_errors("query_freebusy", is_read_only=True, service_type="calendar") @handle_http_errors("query_freebusy", is_read_only=True, service_type="calendar")
@require_google_service("calendar", "calendar_read") @require_google_service("calendar", "calendar_read")

950
gcontacts/contacts_tools.py
View File

File diff suppressed because it is too large Load Diff

6
gdocs/docs_tools.py
View File

@@ -1678,7 +1678,5 @@ async def get_doc_as_markdown(
_comment_tools = create_comment_tools("document", "document_id") _comment_tools = create_comment_tools("document", "document_id")
# Extract and register the functions # Extract and register the functions
read_doc_comments = _comment_tools["read_comments"] list_document_comments = _comment_tools["list_comments"]
create_doc_comment = _comment_tools["create_comment"] manage_document_comment = _comment_tools["manage_comment"]
reply_to_comment = _comment_tools["reply_to_comment"]
resolve_comment = _comment_tools["resolve_comment"]

656
gdrive/drive_tools.py
View File

@@ -1774,365 +1774,368 @@ async def get_drive_shareable_link(
@server.tool() @server.tool()
@handle_http_errors("share_drive_file", is_read_only=False, service_type="drive") @handle_http_errors("manage_drive_access", is_read_only=False, service_type="drive")
@require_google_service("drive", "drive_file") @require_google_service("drive", "drive_file")
async def share_drive_file( async def manage_drive_access(
service, service,
user_google_email: str, user_google_email: str,
file_id: str, file_id: str,
action: str,
share_with: Optional[str] = None, share_with: Optional[str] = None,
role: str = "reader", role: Optional[str] = None,
share_type: str = "user", share_type: str = "user",
permission_id: Optional[str] = None,
recipients: Optional[List[Dict[str, Any]]] = None,
send_notification: bool = True, send_notification: bool = True,
email_message: Optional[str] = None, email_message: Optional[str] = None,
expiration_time: Optional[str] = None, expiration_time: Optional[str] = None,
allow_file_discovery: Optional[bool] = None, allow_file_discovery: Optional[bool] = None,
new_owner_email: Optional[str] = None,
move_to_new_owners_root: bool = False,
) -> str: ) -> str:
""" """
Shares a Google Drive file or folder with a user, group, domain, or anyone with the link. Consolidated tool for managing Google Drive file and folder access permissions.
When sharing a folder, all files inside inherit the permission. Supports granting, batch-granting, updating, revoking permissions, and
transferring file ownership -- all through a single entry point.
Args: Args:
user_google_email (str): The user's Google email address. Required. user_google_email (str): The user's Google email address. Required.
file_id (str): The ID of the file or folder to share. Required. file_id (str): The ID of the file or folder. Required.
share_with (Optional[str]): Email address (for user/group), domain name (for domain), or omit for 'anyone'. action (str): The access management action to perform. Required. One of:
role (str): Permission role - 'reader', 'commenter', or 'writer'. Defaults to 'reader'. - "grant": Share with a single user, group, domain, or anyone.
share_type (str): Type of sharing - 'user', 'group', 'domain', or 'anyone'. Defaults to 'user'. - "grant_batch": Share with multiple recipients in one call.
send_notification (bool): Whether to send a notification email. Defaults to True. - "update": Modify an existing permission (role or expiration).
email_message (Optional[str]): Custom message for the notification email. - "revoke": Remove an existing permission.
expiration_time (Optional[str]): Expiration time in RFC 3339 format (e.g., "2025-01-15T00:00:00Z"). Permission auto-revokes after this time. - "transfer_owner": Transfer file ownership to another user.
allow_file_discovery (Optional[bool]): For 'domain' or 'anyone' shares - whether the file can be found via search. Defaults to None (API default). share_with (Optional[str]): Email address (user/group), domain name (domain),
or omit for 'anyone'. Used by "grant".
Returns: role (Optional[str]): Permission role -- 'reader', 'commenter', or 'writer'.
str: Confirmation with permission details and shareable link. Used by "grant" (defaults to 'reader') and "update".
""" share_type (str): Type of sharing -- 'user', 'group', 'domain', or 'anyone'.
logger.info( Used by "grant". Defaults to 'user'.
f"[share_drive_file] Invoked. Email: '{user_google_email}', File ID: '{file_id}', Share with: '{share_with}', Role: '{role}', Type: '{share_type}'" permission_id (Optional[str]): The permission ID to modify or remove.
) Required for "update" and "revoke" actions.
recipients (Optional[List[Dict[str, Any]]]): List of recipient objects for
validate_share_role(role) "grant_batch". Each should have: email (str), role (str, optional),
validate_share_type(share_type) share_type (str, optional), expiration_time (str, optional). For domain
shares use 'domain' field instead of 'email'.
if share_type in ("user", "group") and not share_with:
raise ValueError(f"share_with is required for share_type '{share_type}'")
if share_type == "domain" and not share_with:
raise ValueError("share_with (domain name) is required for share_type 'domain'")
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name, webViewLink"
)
file_id = resolved_file_id
permission_body = {
"type": share_type,
"role": role,
}
if share_type in ("user", "group"):
permission_body["emailAddress"] = share_with
elif share_type == "domain":
permission_body["domain"] = share_with
if expiration_time:
validate_expiration_time(expiration_time)
permission_body["expirationTime"] = expiration_time
if share_type in ("domain", "anyone") and allow_file_discovery is not None:
permission_body["allowFileDiscovery"] = allow_file_discovery
create_params = {
"fileId": file_id,
"body": permission_body,
"supportsAllDrives": True,
"fields": "id, type, role, emailAddress, domain, expirationTime",
}
if share_type in ("user", "group"):
create_params["sendNotificationEmail"] = send_notification
if email_message:
create_params["emailMessage"] = email_message
created_permission = await asyncio.to_thread(
service.permissions().create(**create_params).execute
)
output_parts = [
f"Successfully shared '{file_metadata.get('name', 'Unknown')}'",
"",
"Permission created:",
f" - {format_permission_info(created_permission)}",
"",
f"View link: {file_metadata.get('webViewLink', 'N/A')}",
]
return "\n".join(output_parts)
@server.tool()
@handle_http_errors("batch_share_drive_file", is_read_only=False, service_type="drive")
@require_google_service("drive", "drive_file")
async def batch_share_drive_file(
service,
user_google_email: str,
file_id: str,
recipients: List[Dict[str, Any]],
send_notification: bool = True,
email_message: Optional[str] = None,
) -> str:
"""
Shares a Google Drive file or folder with multiple users or groups in a single operation.
Each recipient can have a different role and optional expiration time.
Note: Each recipient is processed sequentially. For very large recipient lists,
consider splitting into multiple calls.
Args:
user_google_email (str): The user's Google email address. Required.
file_id (str): The ID of the file or folder to share. Required.
recipients (List[Dict]): List of recipient objects. Each should have:
- email (str): Recipient email address. Required for 'user' or 'group' share_type.
- role (str): Permission role - 'reader', 'commenter', or 'writer'. Defaults to 'reader'.
- share_type (str, optional): 'user', 'group', or 'domain'. Defaults to 'user'.
- expiration_time (str, optional): Expiration in RFC 3339 format (e.g., "2025-01-15T00:00:00Z").
For domain shares, use 'domain' field instead of 'email':
- domain (str): Domain name. Required when share_type is 'domain'.
send_notification (bool): Whether to send notification emails. Defaults to True. send_notification (bool): Whether to send notification emails. Defaults to True.
email_message (Optional[str]): Custom message for notification emails. Used by "grant" and "grant_batch".
email_message (Optional[str]): Custom notification email message.
Used by "grant" and "grant_batch".
expiration_time (Optional[str]): Expiration in RFC 3339 format
(e.g., "2025-01-15T00:00:00Z"). Used by "grant" and "update".
allow_file_discovery (Optional[bool]): For 'domain'/'anyone' shares, whether
the file appears in search. Used by "grant".
new_owner_email (Optional[str]): Email of the new owner.
Required for "transfer_owner".
move_to_new_owners_root (bool): Move file to the new owner's My Drive root.
Defaults to False. Used by "transfer_owner".
Returns: Returns:
str: Summary of created permissions with success/failure for each recipient. str: Confirmation with details of the permission change applied.
""" """
valid_actions = ("grant", "grant_batch", "update", "revoke", "transfer_owner")
if action not in valid_actions:
raise ValueError(
f"Invalid action '{action}'. Must be one of: {', '.join(valid_actions)}"
)
logger.info( logger.info(
f"[batch_share_drive_file] Invoked. Email: '{user_google_email}', File ID: '{file_id}', Recipients: {len(recipients)}" f"[manage_drive_access] Invoked. Email: '{user_google_email}', "
f"File ID: '{file_id}', Action: '{action}'"
) )
resolved_file_id, file_metadata = await resolve_drive_item( # --- grant: share with a single recipient ---
service, file_id, extra_fields="name, webViewLink" if action == "grant":
) effective_role = role or "reader"
file_id = resolved_file_id validate_share_role(effective_role)
validate_share_type(share_type)
if not recipients: if share_type in ("user", "group") and not share_with:
raise ValueError("recipients list cannot be empty") raise ValueError(f"share_with is required for share_type '{share_type}'")
if share_type == "domain" and not share_with:
raise ValueError(
"share_with (domain name) is required for share_type 'domain'"
)
results = [] resolved_file_id, file_metadata = await resolve_drive_item(
success_count = 0 service, file_id, extra_fields="name, webViewLink"
failure_count = 0 )
file_id = resolved_file_id
for recipient in recipients: permission_body: Dict[str, Any] = {
share_type = recipient.get("share_type", "user")
if share_type == "domain":
domain = recipient.get("domain")
if not domain:
results.append(" - Skipped: missing domain for domain share")
failure_count += 1
continue
identifier = domain
else:
email = recipient.get("email")
if not email:
results.append(" - Skipped: missing email address")
failure_count += 1
continue
identifier = email
role = recipient.get("role", "reader")
try:
validate_share_role(role)
except ValueError as e:
results.append(f" - {identifier}: Failed - {e}")
failure_count += 1
continue
try:
validate_share_type(share_type)
except ValueError as e:
results.append(f" - {identifier}: Failed - {e}")
failure_count += 1
continue
permission_body = {
"type": share_type, "type": share_type,
"role": role, "role": effective_role,
} }
if share_type in ("user", "group"):
permission_body["emailAddress"] = share_with
elif share_type == "domain":
permission_body["domain"] = share_with
if share_type == "domain": if expiration_time:
permission_body["domain"] = identifier validate_expiration_time(expiration_time)
else: permission_body["expirationTime"] = expiration_time
permission_body["emailAddress"] = identifier
if recipient.get("expiration_time"): if share_type in ("domain", "anyone") and allow_file_discovery is not None:
try: permission_body["allowFileDiscovery"] = allow_file_discovery
validate_expiration_time(recipient["expiration_time"])
permission_body["expirationTime"] = recipient["expiration_time"]
except ValueError as e:
results.append(f" - {identifier}: Failed - {e}")
failure_count += 1
continue
create_params = { create_params: Dict[str, Any] = {
"fileId": file_id, "fileId": file_id,
"body": permission_body, "body": permission_body,
"supportsAllDrives": True, "supportsAllDrives": True,
"fields": "id, type, role, emailAddress, domain, expirationTime", "fields": "id, type, role, emailAddress, domain, expirationTime",
} }
if share_type in ("user", "group"): if share_type in ("user", "group"):
create_params["sendNotificationEmail"] = send_notification create_params["sendNotificationEmail"] = send_notification
if email_message: if email_message:
create_params["emailMessage"] = email_message create_params["emailMessage"] = email_message
try: created_permission = await asyncio.to_thread(
created_permission = await asyncio.to_thread( service.permissions().create(**create_params).execute
service.permissions().create(**create_params).execute )
)
results.append(f" - {format_permission_info(created_permission)}")
success_count += 1
except HttpError as e:
results.append(f" - {identifier}: Failed - {str(e)}")
failure_count += 1
output_parts = [ return "\n".join(
f"Batch share results for '{file_metadata.get('name', 'Unknown')}'", [
"", f"Successfully shared '{file_metadata.get('name', 'Unknown')}'",
f"Summary: {success_count} succeeded, {failure_count} failed", "",
"", "Permission created:",
"Results:", f" - {format_permission_info(created_permission)}",
] "",
output_parts.extend(results) f"View link: {file_metadata.get('webViewLink', 'N/A')}",
output_parts.extend( ]
[ )
# --- grant_batch: share with multiple recipients ---
if action == "grant_batch":
if not recipients:
raise ValueError("recipients list is required for 'grant_batch' action")
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name, webViewLink"
)
file_id = resolved_file_id
results: List[str] = []
success_count = 0
failure_count = 0
for recipient in recipients:
r_share_type = recipient.get("share_type", "user")
if r_share_type == "domain":
domain = recipient.get("domain")
if not domain:
results.append(" - Skipped: missing domain for domain share")
failure_count += 1
continue
identifier = domain
else:
r_email = recipient.get("email")
if not r_email:
results.append(" - Skipped: missing email address")
failure_count += 1
continue
identifier = r_email
r_role = recipient.get("role", "reader")
try:
validate_share_role(r_role)
except ValueError as e:
results.append(f" - {identifier}: Failed - {e}")
failure_count += 1
continue
try:
validate_share_type(r_share_type)
except ValueError as e:
results.append(f" - {identifier}: Failed - {e}")
failure_count += 1
continue
r_perm_body: Dict[str, Any] = {
"type": r_share_type,
"role": r_role,
}
if r_share_type == "domain":
r_perm_body["domain"] = identifier
else:
r_perm_body["emailAddress"] = identifier
if recipient.get("expiration_time"):
try:
validate_expiration_time(recipient["expiration_time"])
r_perm_body["expirationTime"] = recipient["expiration_time"]
except ValueError as e:
results.append(f" - {identifier}: Failed - {e}")
failure_count += 1
continue
r_create_params: Dict[str, Any] = {
"fileId": file_id,
"body": r_perm_body,
"supportsAllDrives": True,
"fields": "id, type, role, emailAddress, domain, expirationTime",
}
if r_share_type in ("user", "group"):
r_create_params["sendNotificationEmail"] = send_notification
if email_message:
r_create_params["emailMessage"] = email_message
try:
created_perm = await asyncio.to_thread(
service.permissions().create(**r_create_params).execute
)
results.append(f" - {format_permission_info(created_perm)}")
success_count += 1
except HttpError as e:
results.append(f" - {identifier}: Failed - {str(e)}")
failure_count += 1
output_parts = [
f"Batch share results for '{file_metadata.get('name', 'Unknown')}'",
"", "",
f"View link: {file_metadata.get('webViewLink', 'N/A')}", f"Summary: {success_count} succeeded, {failure_count} failed",
"",
"Results:",
] ]
) output_parts.extend(results)
output_parts.extend(
[
"",
f"View link: {file_metadata.get('webViewLink', 'N/A')}",
]
)
return "\n".join(output_parts)
return "\n".join(output_parts) # --- update: modify an existing permission ---
if action == "update":
if not permission_id:
raise ValueError("permission_id is required for 'update' action")
if not role and not expiration_time:
raise ValueError(
"Must provide at least one of: role, expiration_time for 'update' action"
)
if role:
validate_share_role(role)
if expiration_time:
validate_expiration_time(expiration_time)
@server.tool() resolved_file_id, file_metadata = await resolve_drive_item(
@handle_http_errors("update_drive_permission", is_read_only=False, service_type="drive") service, file_id, extra_fields="name"
@require_google_service("drive", "drive_file") )
async def update_drive_permission( file_id = resolved_file_id
service,
user_google_email: str,
file_id: str,
permission_id: str,
role: Optional[str] = None,
expiration_time: Optional[str] = None,
) -> str:
"""
Updates an existing permission on a Google Drive file or folder.
Args: effective_role = role
user_google_email (str): The user's Google email address. Required. if not effective_role:
file_id (str): The ID of the file or folder. Required. current_permission = await asyncio.to_thread(
permission_id (str): The ID of the permission to update (from get_drive_file_permissions). Required. service.permissions()
role (Optional[str]): New role - 'reader', 'commenter', or 'writer'. If not provided, role unchanged. .get(
expiration_time (Optional[str]): Expiration time in RFC 3339 format (e.g., "2025-01-15T00:00:00Z"). Set or update when permission expires. fileId=file_id,
permissionId=permission_id,
supportsAllDrives=True,
fields="role",
)
.execute
)
effective_role = current_permission.get("role")
Returns: update_body: Dict[str, Any] = {"role": effective_role}
str: Confirmation with updated permission details. if expiration_time:
""" update_body["expirationTime"] = expiration_time
logger.info(
f"[update_drive_permission] Invoked. Email: '{user_google_email}', File ID: '{file_id}', Permission ID: '{permission_id}', Role: '{role}'"
)
if not role and not expiration_time: updated_permission = await asyncio.to_thread(
raise ValueError("Must provide at least one of: role, expiration_time")
if role:
validate_share_role(role)
if expiration_time:
validate_expiration_time(expiration_time)
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name"
)
file_id = resolved_file_id
# Google API requires role in update body, so fetch current if not provided
if not role:
current_permission = await asyncio.to_thread(
service.permissions() service.permissions()
.get( .update(
fileId=file_id, fileId=file_id,
permissionId=permission_id, permissionId=permission_id,
body=update_body,
supportsAllDrives=True, supportsAllDrives=True,
fields="role", fields="id, type, role, emailAddress, domain, expirationTime",
) )
.execute .execute
) )
role = current_permission.get("role")
update_body = {"role": role} return "\n".join(
if expiration_time: [
update_body["expirationTime"] = expiration_time f"Successfully updated permission on '{file_metadata.get('name', 'Unknown')}'",
"",
"Updated permission:",
f" - {format_permission_info(updated_permission)}",
]
)
updated_permission = await asyncio.to_thread( # --- revoke: remove an existing permission ---
if action == "revoke":
if not permission_id:
raise ValueError("permission_id is required for 'revoke' action")
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name"
)
file_id = resolved_file_id
await asyncio.to_thread(
service.permissions()
.delete(
fileId=file_id,
permissionId=permission_id,
supportsAllDrives=True,
)
.execute
)
return "\n".join(
[
f"Successfully removed permission from '{file_metadata.get('name', 'Unknown')}'",
"",
f"Permission ID '{permission_id}' has been revoked.",
]
)
# --- transfer_owner: transfer file ownership ---
# action == "transfer_owner"
if not new_owner_email:
raise ValueError("new_owner_email is required for 'transfer_owner' action")
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name, owners"
)
file_id = resolved_file_id
current_owners = file_metadata.get("owners", [])
current_owner_emails = [o.get("emailAddress", "") for o in current_owners]
transfer_body: Dict[str, Any] = {
"type": "user",
"role": "owner",
"emailAddress": new_owner_email,
}
await asyncio.to_thread(
service.permissions() service.permissions()
.update( .create(
fileId=file_id, fileId=file_id,
permissionId=permission_id, body=transfer_body,
body=update_body, transferOwnership=True,
moveToNewOwnersRoot=move_to_new_owners_root,
supportsAllDrives=True, supportsAllDrives=True,
fields="id, type, role, emailAddress, domain, expirationTime", fields="id, type, role, emailAddress",
) )
.execute .execute
) )
output_parts = [ output_parts = [
f"Successfully updated permission on '{file_metadata.get('name', 'Unknown')}'", f"Successfully transferred ownership of '{file_metadata.get('name', 'Unknown')}'",
"", "",
"Updated permission:", f"New owner: {new_owner_email}",
f" - {format_permission_info(updated_permission)}", f"Previous owner(s): {', '.join(current_owner_emails) or 'Unknown'}",
]
return "\n".join(output_parts)
@server.tool()
@handle_http_errors("remove_drive_permission", is_read_only=False, service_type="drive")
@require_google_service("drive", "drive_file")
async def remove_drive_permission(
service,
user_google_email: str,
file_id: str,
permission_id: str,
) -> str:
"""
Removes a permission from a Google Drive file or folder, revoking access.
Args:
user_google_email (str): The user's Google email address. Required.
file_id (str): The ID of the file or folder. Required.
permission_id (str): The ID of the permission to remove (from get_drive_file_permissions). Required.
Returns:
str: Confirmation of the removed permission.
"""
logger.info(
f"[remove_drive_permission] Invoked. Email: '{user_google_email}', File ID: '{file_id}', Permission ID: '{permission_id}'"
)
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name"
)
file_id = resolved_file_id
await asyncio.to_thread(
service.permissions()
.delete(fileId=file_id, permissionId=permission_id, supportsAllDrives=True)
.execute
)
output_parts = [
f"Successfully removed permission from '{file_metadata.get('name', 'Unknown')}'",
"",
f"Permission ID '{permission_id}' has been revoked.",
] ]
if move_to_new_owners_root:
output_parts.append(f"File moved to {new_owner_email}'s My Drive root.")
output_parts.extend(["", "Note: Previous owner now has editor access."])
return "\n".join(output_parts) return "\n".join(output_parts)
@@ -2209,79 +2212,6 @@ async def copy_drive_file(
return "\n".join(output_parts) return "\n".join(output_parts)
@server.tool()
@handle_http_errors(
"transfer_drive_ownership", is_read_only=False, service_type="drive"
)
@require_google_service("drive", "drive_file")
async def transfer_drive_ownership(
service,
user_google_email: str,
file_id: str,
new_owner_email: str,
move_to_new_owners_root: bool = False,
) -> str:
"""
Transfers ownership of a Google Drive file or folder to another user.
This is an irreversible operation. The current owner will become an editor.
Only works within the same Google Workspace domain or for personal accounts.
Args:
user_google_email (str): The user's Google email address. Required.
file_id (str): The ID of the file or folder to transfer. Required.
new_owner_email (str): Email address of the new owner. Required.
move_to_new_owners_root (bool): If True, moves the file to the new owner's My Drive root. Defaults to False.
Returns:
str: Confirmation of the ownership transfer.
"""
logger.info(
f"[transfer_drive_ownership] Invoked. Email: '{user_google_email}', File ID: '{file_id}', New owner: '{new_owner_email}'"
)
resolved_file_id, file_metadata = await resolve_drive_item(
service, file_id, extra_fields="name, owners"
)
file_id = resolved_file_id
current_owners = file_metadata.get("owners", [])
current_owner_emails = [o.get("emailAddress", "") for o in current_owners]
permission_body = {
"type": "user",
"role": "owner",
"emailAddress": new_owner_email,
}
await asyncio.to_thread(
service.permissions()
.create(
fileId=file_id,
body=permission_body,
transferOwnership=True,
moveToNewOwnersRoot=move_to_new_owners_root,
supportsAllDrives=True,
fields="id, type, role, emailAddress",
)
.execute
)
output_parts = [
f"Successfully transferred ownership of '{file_metadata.get('name', 'Unknown')}'",
"",
f"New owner: {new_owner_email}",
f"Previous owner(s): {', '.join(current_owner_emails) or 'Unknown'}",
]
if move_to_new_owners_root:
output_parts.append(f"File moved to {new_owner_email}'s My Drive root.")
output_parts.extend(["", "Note: Previous owner now has editor access."])
return "\n".join(output_parts)
@server.tool() @server.tool()
@handle_http_errors( @handle_http_errors(
"set_drive_file_permissions", is_read_only=False, service_type="drive" "set_drive_file_permissions", is_read_only=False, service_type="drive"

126
gmail/gmail_tools.py
View File

@@ -1925,88 +1925,72 @@ async def list_gmail_filters(service, user_google_email: str) -> str:
@server.tool() @server.tool()
@handle_http_errors("create_gmail_filter", service_type="gmail") @handle_http_errors("manage_gmail_filter", service_type="gmail")
@require_google_service("gmail", "gmail_settings_basic") @require_google_service("gmail", "gmail_settings_basic")
async def create_gmail_filter( async def manage_gmail_filter(
service, service,
user_google_email: str, user_google_email: str,
criteria: Annotated[ action: str,
Dict[str, Any], criteria: Optional[Dict[str, Any]] = None,
Field( filter_action: Optional[Dict[str, Any]] = None,
description="Filter criteria object as defined in the Gmail API.", filter_id: Optional[str] = None,
),
],
action: Annotated[
Dict[str, Any],
Field(
description="Filter action object as defined in the Gmail API.",
),
],
) -> str: ) -> str:
""" """
Creates a Gmail filter using the users.settings.filters API. Manages Gmail filters. Supports creating and deleting filters.
Args: Args:
user_google_email (str): The user's Google email address. Required. user_google_email (str): The user's Google email address. Required.
criteria (Dict[str, Any]): Criteria for matching messages. action (str): Action to perform - "create" or "delete".
action (Dict[str, Any]): Actions to apply to matched messages. criteria (Optional[Dict[str, Any]]): Filter criteria object (required for create).
filter_action (Optional[Dict[str, Any]]): Filter action object (required for create). Named 'filter_action' to avoid shadowing the 'action' parameter.
filter_id (Optional[str]): ID of the filter to delete (required for delete).
Returns: Returns:
str: Confirmation message with the created filter ID. str: Confirmation message with filter details.
""" """
logger.info("[create_gmail_filter] Invoked") action_lower = action.lower().strip()
if action_lower == "create":
filter_body = {"criteria": criteria, "action": action} if not criteria or not filter_action:
raise ValueError(
created_filter = await asyncio.to_thread( "criteria and filter_action are required for create action"
service.users() )
.settings() logger.info("[manage_gmail_filter] Creating filter")
.filters() filter_body = {"criteria": criteria, "action": filter_action}
.create(userId="me", body=filter_body) created_filter = await asyncio.to_thread(
.execute service.users()
) .settings()
.filters()
filter_id = created_filter.get("id", "(unknown)") .create(userId="me", body=filter_body)
return f"Filter created successfully!\nFilter ID: {filter_id}" .execute
)
fid = created_filter.get("id", "(unknown)")
@server.tool() return f"Filter created successfully!\nFilter ID: {fid}"
@handle_http_errors("delete_gmail_filter", service_type="gmail") elif action_lower == "delete":
@require_google_service("gmail", "gmail_settings_basic") if not filter_id:
async def delete_gmail_filter( raise ValueError("filter_id is required for delete action")
service, logger.info(f"[manage_gmail_filter] Deleting filter {filter_id}")
user_google_email: str, filter_details = await asyncio.to_thread(
filter_id: str = Field(..., description="ID of the filter to delete."), service.users().settings().filters().get(userId="me", id=filter_id).execute
) -> str: )
""" await asyncio.to_thread(
Deletes a Gmail filter by ID. service.users()
.settings()
Args: .filters()
user_google_email (str): The user's Google email address. Required. .delete(userId="me", id=filter_id)
filter_id (str): The ID of the filter to delete. .execute
)
Returns: criteria_info = filter_details.get("criteria", {})
str: Confirmation message for the deletion. action_info = filter_details.get("action", {})
""" return (
logger.info(f"[delete_gmail_filter] Invoked. Filter ID: '{filter_id}'") "Filter deleted successfully!\n"
f"Filter ID: {filter_id}\n"
filter_details = await asyncio.to_thread( f"Criteria: {criteria_info or '(none)'}\n"
service.users().settings().filters().get(userId="me", id=filter_id).execute f"Action: {action_info or '(none)'}"
) )
else:
await asyncio.to_thread( raise ValueError(
service.users().settings().filters().delete(userId="me", id=filter_id).execute f"Invalid action '{action_lower}'. Must be 'create' or 'delete'."
) )
criteria = filter_details.get("criteria", {})
action = filter_details.get("action", {})
return (
"Filter deleted successfully!\n"
f"Filter ID: {filter_id}\n"
f"Criteria: {criteria or '(none)'}\n"
f"Action: {action or '(none)'}"
)
@server.tool() @server.tool()

55
gsearch/search_tools.py
View File

@@ -33,6 +33,7 @@ async def search_custom(
file_type: Optional[str] = None, file_type: Optional[str] = None,
language: Optional[str] = None, language: Optional[str] = None,
country: Optional[str] = None, country: Optional[str] = None,
sites: Optional[List[str]] = None,
) -> str: ) -> str:
""" """
Performs a search using Google Custom Search JSON API. Performs a search using Google Custom Search JSON API.
@@ -50,6 +51,7 @@ async def search_custom(
file_type (Optional[str]): Filter by file type (e.g., "pdf", "doc"). file_type (Optional[str]): Filter by file type (e.g., "pdf", "doc").
language (Optional[str]): Language code for results (e.g., "lang_en"). language (Optional[str]): Language code for results (e.g., "lang_en").
country (Optional[str]): Country code for results (e.g., "countryUS"). country (Optional[str]): Country code for results (e.g., "countryUS").
sites (Optional[List[str]]): List of sites/domains to restrict search to (e.g., ["example.com", "docs.example.com"]). When provided, results are limited to these sites.
Returns: Returns:
str: Formatted search results including title, link, and snippet for each result. str: Formatted search results including title, link, and snippet for each result.
@@ -71,6 +73,12 @@ async def search_custom(
f"[search_custom] Invoked. Email: '{user_google_email}', Query: '{q}', CX: '{cx}'" f"[search_custom] Invoked. Email: '{user_google_email}', Query: '{q}', CX: '{cx}'"
) )
# Apply site restriction if sites are provided
if sites:
site_query = " OR ".join([f"site:{site}" for site in sites])
q = f"{q} ({site_query})"
logger.info(f"[search_custom] Applied site restriction: {sites}")
# Build the request parameters # Build the request parameters
params = { params = {
"key": api_key, "key": api_key,
@@ -224,50 +232,3 @@ async def get_search_engine_info(service, user_google_email: str) -> str:
logger.info(f"Search engine info retrieved successfully for {user_google_email}") logger.info(f"Search engine info retrieved successfully for {user_google_email}")
return confirmation_message return confirmation_message
@server.tool()
@handle_http_errors(
"search_custom_siterestrict", is_read_only=True, service_type="customsearch"
)
@require_google_service("customsearch", "customsearch")
async def search_custom_siterestrict(
service,
user_google_email: str,
q: str,
sites: List[str],
num: int = 10,
start: int = 1,
safe: Literal["active", "moderate", "off"] = "off",
) -> str:
"""
Performs a search restricted to specific sites using Google Custom Search.
Args:
user_google_email (str): The user's Google email address. Required.
q (str): The search query. Required.
sites (List[str]): List of sites/domains to search within.
num (int): Number of results to return (1-10). Defaults to 10.
start (int): The index of the first result to return (1-based). Defaults to 1.
safe (Literal["active", "moderate", "off"]): Safe search level. Defaults to "off".
Returns:
str: Formatted search results from the specified sites.
"""
logger.info(
f"[search_custom_siterestrict] Invoked. Email: '{user_google_email}', Query: '{q}', Sites: {sites}"
)
# Build site restriction query
site_query = " OR ".join([f"site:{site}" for site in sites])
full_query = f"{q} ({site_query})"
# Use the main search function with the modified query
return await search_custom(
service=service,
user_google_email=user_google_email,
q=full_query,
num=num,
start=start,
safe=safe,
)

714
gsheets/sheets_tools.py
View File

@@ -729,405 +729,401 @@ async def format_sheet_range(
@server.tool() @server.tool()
@handle_http_errors("add_conditional_formatting", service_type="sheets") @handle_http_errors("manage_conditional_formatting", service_type="sheets")
@require_google_service("sheets", "sheets_write") @require_google_service("sheets", "sheets_write")
async def add_conditional_formatting( async def manage_conditional_formatting(
service, service,
user_google_email: str, user_google_email: str,
spreadsheet_id: str, spreadsheet_id: str,
range_name: str, action: str,
condition_type: str,
condition_values: Optional[Union[str, List[Union[str, int, float]]]] = None,
background_color: Optional[str] = None,
text_color: Optional[str] = None,
rule_index: Optional[int] = None,
gradient_points: Optional[Union[str, List[dict]]] = None,
) -> str:
"""
Adds a conditional formatting rule to a range.
Args:
user_google_email (str): The user's Google email address. Required.
spreadsheet_id (str): The ID of the spreadsheet. Required.
range_name (str): A1-style range (optionally with sheet name). Required.
condition_type (str): Sheets condition type (e.g., NUMBER_GREATER, TEXT_CONTAINS, DATE_BEFORE, CUSTOM_FORMULA).
condition_values (Optional[Union[str, List[Union[str, int, float]]]]): Values for the condition; accepts a list or a JSON string representing a list. Depends on condition_type.
background_color (Optional[str]): Hex background color to apply when condition matches.
text_color (Optional[str]): Hex text color to apply when condition matches.
rule_index (Optional[int]): Optional position to insert the rule (0-based) within the sheet's rules.
gradient_points (Optional[Union[str, List[dict]]]): List (or JSON list) of gradient points for a color scale. If provided, a gradient rule is created and boolean parameters are ignored.
Returns:
str: Confirmation of the added rule.
"""
logger.info(
"[add_conditional_formatting] Invoked. Email: '%s', Spreadsheet: %s, Range: %s, Type: %s, Values: %s",
user_google_email,
spreadsheet_id,
range_name,
condition_type,
condition_values,
)
if rule_index is not None and (not isinstance(rule_index, int) or rule_index < 0):
raise UserInputError("rule_index must be a non-negative integer when provided.")
condition_values_list = _parse_condition_values(condition_values)
gradient_points_list = _parse_gradient_points(gradient_points)
sheets, sheet_titles = await _fetch_sheets_with_rules(service, spreadsheet_id)
grid_range = _parse_a1_range(range_name, sheets)
target_sheet = None
for sheet in sheets:
if sheet.get("properties", {}).get("sheetId") == grid_range.get("sheetId"):
target_sheet = sheet
break
if target_sheet is None:
raise UserInputError(
"Target sheet not found while adding conditional formatting."
)
current_rules = target_sheet.get("conditionalFormats", []) or []
insert_at = rule_index if rule_index is not None else len(current_rules)
if insert_at > len(current_rules):
raise UserInputError(
f"rule_index {insert_at} is out of range for sheet '{target_sheet.get('properties', {}).get('title', 'Unknown')}' "
f"(current count: {len(current_rules)})."
)
if gradient_points_list:
new_rule = _build_gradient_rule([grid_range], gradient_points_list)
rule_desc = "gradient"
values_desc = ""
applied_parts = [f"gradient points {len(gradient_points_list)}"]
else:
rule, cond_type_normalized = _build_boolean_rule(
[grid_range],
condition_type,
condition_values_list,
background_color,
text_color,
)
new_rule = rule
rule_desc = cond_type_normalized
values_desc = ""
if condition_values_list:
values_desc = f" with values {condition_values_list}"
applied_parts = []
if background_color:
applied_parts.append(f"background {background_color}")
if text_color:
applied_parts.append(f"text {text_color}")
new_rules_state = copy.deepcopy(current_rules)
new_rules_state.insert(insert_at, new_rule)
add_rule_request = {"rule": new_rule}
if rule_index is not None:
add_rule_request["index"] = rule_index
request_body = {"requests": [{"addConditionalFormatRule": add_rule_request}]}
await asyncio.to_thread(
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
.execute
)
format_desc = ", ".join(applied_parts) if applied_parts else "format applied"
sheet_title = target_sheet.get("properties", {}).get("title", "Unknown")
state_text = _format_conditional_rules_section(
sheet_title, new_rules_state, sheet_titles, indent=""
)
return "\n".join(
[
f"Added conditional format on '{range_name}' in spreadsheet {spreadsheet_id} "
f"for {user_google_email}: {rule_desc}{values_desc}; format: {format_desc}.",
state_text,
]
)
@server.tool()
@handle_http_errors("update_conditional_formatting", service_type="sheets")
@require_google_service("sheets", "sheets_write")
async def update_conditional_formatting(
service,
user_google_email: str,
spreadsheet_id: str,
rule_index: int,
range_name: Optional[str] = None, range_name: Optional[str] = None,
condition_type: Optional[str] = None, condition_type: Optional[str] = None,
condition_values: Optional[Union[str, List[Union[str, int, float]]]] = None, condition_values: Optional[Union[str, List[Union[str, int, float]]]] = None,
background_color: Optional[str] = None, background_color: Optional[str] = None,
text_color: Optional[str] = None, text_color: Optional[str] = None,
sheet_name: Optional[str] = None, rule_index: Optional[int] = None,
gradient_points: Optional[Union[str, List[dict]]] = None, gradient_points: Optional[Union[str, List[dict]]] = None,
sheet_name: Optional[str] = None,
) -> str: ) -> str:
""" """
Updates an existing conditional formatting rule by index on a sheet. Manages conditional formatting rules on a Google Sheet. Supports adding,
updating, and deleting conditional formatting rules via a single tool.
Args: Args:
user_google_email (str): The user's Google email address. Required. user_google_email (str): The user's Google email address. Required.
spreadsheet_id (str): The ID of the spreadsheet. Required. spreadsheet_id (str): The ID of the spreadsheet. Required.
range_name (Optional[str]): A1-style range to apply the updated rule (optionally with sheet name). If omitted, existing ranges are preserved. action (str): The operation to perform. Must be one of "add", "update",
rule_index (int): Index of the rule to update (0-based). or "delete".
condition_type (Optional[str]): Sheets condition type. If omitted, the existing rule's type is preserved. range_name (Optional[str]): A1-style range (optionally with sheet name).
condition_values (Optional[Union[str, List[Union[str, int, float]]]]): Values for the condition. Required for "add". Optional for "update" (preserves existing ranges
background_color (Optional[str]): Hex background color when condition matches. if omitted). Not used for "delete".
text_color (Optional[str]): Hex text color when condition matches. condition_type (Optional[str]): Sheets condition type (e.g., NUMBER_GREATER,
sheet_name (Optional[str]): Sheet name to locate the rule when range_name is omitted. Defaults to first sheet. TEXT_CONTAINS, DATE_BEFORE, CUSTOM_FORMULA). Required for "add".
gradient_points (Optional[Union[str, List[dict]]]): If provided, updates the rule to a gradient color scale using these points. Optional for "update" (preserves existing type if omitted).
condition_values (Optional[Union[str, List[Union[str, int, float]]]]): Values
for the condition; accepts a list or a JSON string representing a list.
Depends on condition_type. Used by "add" and "update".
background_color (Optional[str]): Hex background color to apply when
condition matches. Used by "add" and "update".
text_color (Optional[str]): Hex text color to apply when condition matches.
Used by "add" and "update".
rule_index (Optional[int]): 0-based index of the rule. For "add", optionally
specifies insertion position. Required for "update" and "delete".
gradient_points (Optional[Union[str, List[dict]]]): List (or JSON list) of
gradient points for a color scale. If provided, a gradient rule is created
and boolean parameters are ignored. Used by "add" and "update".
sheet_name (Optional[str]): Sheet name to locate the rule when range_name is
omitted. Defaults to the first sheet. Used by "update" and "delete".
Returns: Returns:
str: Confirmation of the updated rule and the current rule state. str: Confirmation of the operation and the current rule state.
""" """
allowed_actions = {"add", "update", "delete"}
action_normalized = action.strip().lower()
if action_normalized not in allowed_actions:
raise UserInputError(
f"action must be one of {sorted(allowed_actions)}, got '{action}'."
)
logger.info( logger.info(
"[update_conditional_formatting] Invoked. Email: '%s', Spreadsheet: %s, Range: %s, Rule Index: %s", "[manage_conditional_formatting] Invoked. Action: '%s', Email: '%s', Spreadsheet: %s",
action_normalized,
user_google_email, user_google_email,
spreadsheet_id, spreadsheet_id,
range_name,
rule_index,
) )
if not isinstance(rule_index, int) or rule_index < 0: if action_normalized == "add":
raise UserInputError("rule_index must be a non-negative integer.") if not range_name:
raise UserInputError("range_name is required for action 'add'.")
if not condition_type and not gradient_points:
raise UserInputError(
"condition_type (or gradient_points) is required for action 'add'."
)
condition_values_list = _parse_condition_values(condition_values) if rule_index is not None and (
gradient_points_list = _parse_gradient_points(gradient_points) not isinstance(rule_index, int) or rule_index < 0
):
raise UserInputError(
"rule_index must be a non-negative integer when provided."
)
sheets, sheet_titles = await _fetch_sheets_with_rules(service, spreadsheet_id) gradient_points_list = _parse_gradient_points(gradient_points)
condition_values_list = (
None if gradient_points_list else _parse_condition_values(condition_values)
)
target_sheet = None sheets, sheet_titles = await _fetch_sheets_with_rules(service, spreadsheet_id)
grid_range = None
if range_name:
grid_range = _parse_a1_range(range_name, sheets) grid_range = _parse_a1_range(range_name, sheets)
target_sheet = None
for sheet in sheets: for sheet in sheets:
if sheet.get("properties", {}).get("sheetId") == grid_range.get("sheetId"): if sheet.get("properties", {}).get("sheetId") == grid_range.get("sheetId"):
target_sheet = sheet target_sheet = sheet
break break
else: if target_sheet is None:
raise UserInputError(
"Target sheet not found while adding conditional formatting."
)
current_rules = target_sheet.get("conditionalFormats", []) or []
insert_at = rule_index if rule_index is not None else len(current_rules)
if insert_at > len(current_rules):
raise UserInputError(
f"rule_index {insert_at} is out of range for sheet "
f"'{target_sheet.get('properties', {}).get('title', 'Unknown')}' "
f"(current count: {len(current_rules)})."
)
if gradient_points_list:
new_rule = _build_gradient_rule([grid_range], gradient_points_list)
rule_desc = "gradient"
values_desc = ""
applied_parts = [f"gradient points {len(gradient_points_list)}"]
else:
rule, cond_type_normalized = _build_boolean_rule(
[grid_range],
condition_type,
condition_values_list,
background_color,
text_color,
)
new_rule = rule
rule_desc = cond_type_normalized
values_desc = ""
if condition_values_list:
values_desc = f" with values {condition_values_list}"
applied_parts = []
if background_color:
applied_parts.append(f"background {background_color}")
if text_color:
applied_parts.append(f"text {text_color}")
new_rules_state = copy.deepcopy(current_rules)
new_rules_state.insert(insert_at, new_rule)
add_rule_request = {"rule": new_rule}
if rule_index is not None:
add_rule_request["index"] = rule_index
request_body = {"requests": [{"addConditionalFormatRule": add_rule_request}]}
await asyncio.to_thread(
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
.execute
)
format_desc = ", ".join(applied_parts) if applied_parts else "format applied"
sheet_title = target_sheet.get("properties", {}).get("title", "Unknown")
state_text = _format_conditional_rules_section(
sheet_title, new_rules_state, sheet_titles, indent=""
)
return "\n".join(
[
f"Added conditional format on '{range_name}' in spreadsheet "
f"{spreadsheet_id} for {user_google_email}: "
f"{rule_desc}{values_desc}; format: {format_desc}.",
state_text,
]
)
elif action_normalized == "update":
if rule_index is None:
raise UserInputError("rule_index is required for action 'update'.")
if not isinstance(rule_index, int) or rule_index < 0:
raise UserInputError("rule_index must be a non-negative integer.")
gradient_points_list = _parse_gradient_points(gradient_points)
condition_values_list = (
None
if gradient_points_list is not None
else _parse_condition_values(condition_values)
)
sheets, sheet_titles = await _fetch_sheets_with_rules(service, spreadsheet_id)
target_sheet = None
grid_range = None
if range_name:
grid_range = _parse_a1_range(range_name, sheets)
for sheet in sheets:
if sheet.get("properties", {}).get("sheetId") == grid_range.get(
"sheetId"
):
target_sheet = sheet
break
else:
target_sheet = _select_sheet(sheets, sheet_name)
if target_sheet is None:
raise UserInputError(
"Target sheet not found while updating conditional formatting."
)
sheet_props = target_sheet.get("properties", {})
sheet_id = sheet_props.get("sheetId")
sheet_title = sheet_props.get("title", f"Sheet {sheet_id}")
rules = target_sheet.get("conditionalFormats", []) or []
if rule_index >= len(rules):
raise UserInputError(
f"rule_index {rule_index} is out of range for sheet "
f"'{sheet_title}' (current count: {len(rules)})."
)
existing_rule = rules[rule_index]
ranges_to_use = existing_rule.get("ranges", [])
if range_name:
ranges_to_use = [grid_range]
if not ranges_to_use:
ranges_to_use = [{"sheetId": sheet_id}]
new_rule = None
rule_desc = ""
values_desc = ""
format_desc = ""
if gradient_points_list is not None:
new_rule = _build_gradient_rule(ranges_to_use, gradient_points_list)
rule_desc = "gradient"
format_desc = f"gradient points {len(gradient_points_list)}"
elif "gradientRule" in existing_rule:
if any(
[
background_color,
text_color,
condition_type,
condition_values_list,
]
):
raise UserInputError(
"Existing rule is a gradient rule. Provide gradient_points "
"to update it, or omit formatting/condition parameters to "
"keep it unchanged."
)
new_rule = {
"ranges": ranges_to_use,
"gradientRule": existing_rule.get("gradientRule", {}),
}
rule_desc = "gradient"
format_desc = "gradient (unchanged)"
else:
existing_boolean = existing_rule.get("booleanRule", {})
existing_condition = existing_boolean.get("condition", {})
existing_format = copy.deepcopy(existing_boolean.get("format", {}))
cond_type = (condition_type or existing_condition.get("type", "")).upper()
if not cond_type:
raise UserInputError("condition_type is required for boolean rules.")
if cond_type not in CONDITION_TYPES:
raise UserInputError(
f"condition_type must be one of {sorted(CONDITION_TYPES)}."
)
if condition_values_list is not None:
cond_values = [
{"userEnteredValue": str(val)} for val in condition_values_list
]
else:
cond_values = existing_condition.get("values")
new_format = copy.deepcopy(existing_format) if existing_format else {}
if background_color is not None:
bg_color_parsed = _parse_hex_color(background_color)
if bg_color_parsed:
new_format["backgroundColor"] = bg_color_parsed
elif "backgroundColor" in new_format:
del new_format["backgroundColor"]
if text_color is not None:
text_color_parsed = _parse_hex_color(text_color)
text_format = copy.deepcopy(new_format.get("textFormat", {}))
if text_color_parsed:
text_format["foregroundColor"] = text_color_parsed
elif "foregroundColor" in text_format:
del text_format["foregroundColor"]
if text_format:
new_format["textFormat"] = text_format
elif "textFormat" in new_format:
del new_format["textFormat"]
if not new_format:
raise UserInputError(
"At least one format option must remain on the rule."
)
new_rule = {
"ranges": ranges_to_use,
"booleanRule": {
"condition": {"type": cond_type},
"format": new_format,
},
}
if cond_values:
new_rule["booleanRule"]["condition"]["values"] = cond_values
rule_desc = cond_type
if condition_values_list:
values_desc = f" with values {condition_values_list}"
format_parts = []
if "backgroundColor" in new_format:
format_parts.append("background updated")
if "textFormat" in new_format and new_format["textFormat"].get(
"foregroundColor"
):
format_parts.append("text color updated")
format_desc = (
", ".join(format_parts) if format_parts else "format preserved"
)
new_rules_state = copy.deepcopy(rules)
new_rules_state[rule_index] = new_rule
request_body = {
"requests": [
{
"updateConditionalFormatRule": {
"index": rule_index,
"sheetId": sheet_id,
"rule": new_rule,
}
}
]
}
await asyncio.to_thread(
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
.execute
)
state_text = _format_conditional_rules_section(
sheet_title, new_rules_state, sheet_titles, indent=""
)
return "\n".join(
[
f"Updated conditional format at index {rule_index} on sheet "
f"'{sheet_title}' in spreadsheet {spreadsheet_id} "
f"for {user_google_email}: "
f"{rule_desc}{values_desc}; format: {format_desc}.",
state_text,
]
)
else: # action_normalized == "delete"
if rule_index is None:
raise UserInputError("rule_index is required for action 'delete'.")
if not isinstance(rule_index, int) or rule_index < 0:
raise UserInputError("rule_index must be a non-negative integer.")
sheets, sheet_titles = await _fetch_sheets_with_rules(service, spreadsheet_id)
target_sheet = _select_sheet(sheets, sheet_name) target_sheet = _select_sheet(sheets, sheet_name)
if target_sheet is None: sheet_props = target_sheet.get("properties", {})
raise UserInputError( sheet_id = sheet_props.get("sheetId")
"Target sheet not found while updating conditional formatting." target_sheet_name = sheet_props.get("title", f"Sheet {sheet_id}")
) rules = target_sheet.get("conditionalFormats", []) or []
if rule_index >= len(rules):
sheet_props = target_sheet.get("properties", {})
sheet_id = sheet_props.get("sheetId")
sheet_title = sheet_props.get("title", f"Sheet {sheet_id}")
rules = target_sheet.get("conditionalFormats", []) or []
if rule_index >= len(rules):
raise UserInputError(
f"rule_index {rule_index} is out of range for sheet '{sheet_title}' (current count: {len(rules)})."
)
existing_rule = rules[rule_index]
ranges_to_use = existing_rule.get("ranges", [])
if range_name:
ranges_to_use = [grid_range]
if not ranges_to_use:
ranges_to_use = [{"sheetId": sheet_id}]
new_rule = None
rule_desc = ""
values_desc = ""
format_desc = ""
if gradient_points_list is not None:
new_rule = _build_gradient_rule(ranges_to_use, gradient_points_list)
rule_desc = "gradient"
format_desc = f"gradient points {len(gradient_points_list)}"
elif "gradientRule" in existing_rule:
if any([background_color, text_color, condition_type, condition_values_list]):
raise UserInputError( raise UserInputError(
"Existing rule is a gradient rule. Provide gradient_points to update it, or omit formatting/condition parameters to keep it unchanged." f"rule_index {rule_index} is out of range for sheet "
) f"'{target_sheet_name}' (current count: {len(rules)})."
new_rule = {
"ranges": ranges_to_use,
"gradientRule": existing_rule.get("gradientRule", {}),
}
rule_desc = "gradient"
format_desc = "gradient (unchanged)"
else:
existing_boolean = existing_rule.get("booleanRule", {})
existing_condition = existing_boolean.get("condition", {})
existing_format = copy.deepcopy(existing_boolean.get("format", {}))
cond_type = (condition_type or existing_condition.get("type", "")).upper()
if not cond_type:
raise UserInputError("condition_type is required for boolean rules.")
if cond_type not in CONDITION_TYPES:
raise UserInputError(
f"condition_type must be one of {sorted(CONDITION_TYPES)}."
) )
if condition_values_list is not None: new_rules_state = copy.deepcopy(rules)
cond_values = [ del new_rules_state[rule_index]
{"userEnteredValue": str(val)} for val in condition_values_list
request_body = {
"requests": [
{
"deleteConditionalFormatRule": {
"index": rule_index,
"sheetId": sheet_id,
}
}
] ]
else:
cond_values = existing_condition.get("values")
new_format = copy.deepcopy(existing_format) if existing_format else {}
if background_color is not None:
bg_color_parsed = _parse_hex_color(background_color)
if bg_color_parsed:
new_format["backgroundColor"] = bg_color_parsed
elif "backgroundColor" in new_format:
del new_format["backgroundColor"]
if text_color is not None:
text_color_parsed = _parse_hex_color(text_color)
text_format = copy.deepcopy(new_format.get("textFormat", {}))
if text_color_parsed:
text_format["foregroundColor"] = text_color_parsed
elif "foregroundColor" in text_format:
del text_format["foregroundColor"]
if text_format:
new_format["textFormat"] = text_format
elif "textFormat" in new_format:
del new_format["textFormat"]
if not new_format:
raise UserInputError("At least one format option must remain on the rule.")
new_rule = {
"ranges": ranges_to_use,
"booleanRule": {
"condition": {"type": cond_type},
"format": new_format,
},
} }
if cond_values:
new_rule["booleanRule"]["condition"]["values"] = cond_values
rule_desc = cond_type await asyncio.to_thread(
if condition_values_list: service.spreadsheets()
values_desc = f" with values {condition_values_list}" .batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
format_parts = [] .execute
if "backgroundColor" in new_format:
format_parts.append("background updated")
if "textFormat" in new_format and new_format["textFormat"].get(
"foregroundColor"
):
format_parts.append("text color updated")
format_desc = ", ".join(format_parts) if format_parts else "format preserved"
new_rules_state = copy.deepcopy(rules)
new_rules_state[rule_index] = new_rule
request_body = {
"requests": [
{
"updateConditionalFormatRule": {
"index": rule_index,
"sheetId": sheet_id,
"rule": new_rule,
}
}
]
}
await asyncio.to_thread(
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
.execute
)
state_text = _format_conditional_rules_section(
sheet_title, new_rules_state, sheet_titles, indent=""
)
return "\n".join(
[
f"Updated conditional format at index {rule_index} on sheet '{sheet_title}' in spreadsheet {spreadsheet_id} "
f"for {user_google_email}: {rule_desc}{values_desc}; format: {format_desc}.",
state_text,
]
)
@server.tool()
@handle_http_errors("delete_conditional_formatting", service_type="sheets")
@require_google_service("sheets", "sheets_write")
async def delete_conditional_formatting(
service,
user_google_email: str,
spreadsheet_id: str,
rule_index: int,
sheet_name: Optional[str] = None,
) -> str:
"""
Deletes an existing conditional formatting rule by index on a sheet.
Args:
user_google_email (str): The user's Google email address. Required.
spreadsheet_id (str): The ID of the spreadsheet. Required.
rule_index (int): Index of the rule to delete (0-based).
sheet_name (Optional[str]): Name of the sheet that contains the rule. Defaults to the first sheet if not provided.
Returns:
str: Confirmation of the deletion and the current rule state.
"""
logger.info(
"[delete_conditional_formatting] Invoked. Email: '%s', Spreadsheet: %s, Sheet: %s, Rule Index: %s",
user_google_email,
spreadsheet_id,
sheet_name,
rule_index,
)
if not isinstance(rule_index, int) or rule_index < 0:
raise UserInputError("rule_index must be a non-negative integer.")
sheets, sheet_titles = await _fetch_sheets_with_rules(service, spreadsheet_id)
target_sheet = _select_sheet(sheets, sheet_name)
sheet_props = target_sheet.get("properties", {})
sheet_id = sheet_props.get("sheetId")
target_sheet_name = sheet_props.get("title", f"Sheet {sheet_id}")
rules = target_sheet.get("conditionalFormats", []) or []
if rule_index >= len(rules):
raise UserInputError(
f"rule_index {rule_index} is out of range for sheet '{target_sheet_name}' (current count: {len(rules)})."
) )
new_rules_state = copy.deepcopy(rules) state_text = _format_conditional_rules_section(
del new_rules_state[rule_index] target_sheet_name, new_rules_state, sheet_titles, indent=""
)
request_body = { return "\n".join(
"requests": [ [
{ f"Deleted conditional format at index {rule_index} on sheet "
"deleteConditionalFormatRule": { f"'{target_sheet_name}' in spreadsheet {spreadsheet_id} "
"index": rule_index, f"for {user_google_email}.",
"sheetId": sheet_id, state_text,
} ]
} )
]
}
await asyncio.to_thread(
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
.execute
)
state_text = _format_conditional_rules_section(
target_sheet_name, new_rules_state, sheet_titles, indent=""
)
return "\n".join(
[
f"Deleted conditional format at index {rule_index} on sheet '{target_sheet_name}' in spreadsheet {spreadsheet_id} for {user_google_email}.",
state_text,
]
)
@server.tool() @server.tool()
@@ -1232,7 +1228,5 @@ async def create_sheet(
_comment_tools = create_comment_tools("spreadsheet", "spreadsheet_id") _comment_tools = create_comment_tools("spreadsheet", "spreadsheet_id")
# Extract and register the functions # Extract and register the functions
read_sheet_comments = _comment_tools["read_comments"] list_spreadsheet_comments = _comment_tools["list_comments"]
create_sheet_comment = _comment_tools["create_comment"] manage_spreadsheet_comment = _comment_tools["manage_comment"]
reply_to_sheet_comment = _comment_tools["reply_to_comment"]
resolve_sheet_comment = _comment_tools["resolve_comment"]

12
gslides/slides_tools.py
View File

@@ -322,13 +322,9 @@ You can view or download the thumbnail using the provided URL."""
# Create comment management tools for slides # Create comment management tools for slides
_comment_tools = create_comment_tools("presentation", "presentation_id") _comment_tools = create_comment_tools("presentation", "presentation_id")
read_presentation_comments = _comment_tools["read_comments"] list_presentation_comments = _comment_tools["list_comments"]
create_presentation_comment = _comment_tools["create_comment"] manage_presentation_comment = _comment_tools["manage_comment"]
reply_to_presentation_comment = _comment_tools["reply_to_comment"]
resolve_presentation_comment = _comment_tools["resolve_comment"]
# Aliases for backwards compatibility and intuitive naming # Aliases for backwards compatibility and intuitive naming
read_slide_comments = read_presentation_comments list_slide_comments = list_presentation_comments
create_slide_comment = create_presentation_comment manage_slide_comment = manage_presentation_comment
reply_to_slide_comment = reply_to_presentation_comment
resolve_slide_comment = resolve_presentation_comment

603
gtasks/tasks_tools.py
View File

@@ -15,7 +15,7 @@ from mcp import Resource
from auth.oauth_config import is_oauth21_enabled, is_external_oauth21_provider from auth.oauth_config import is_oauth21_enabled, is_external_oauth21_provider
from auth.service_decorator import require_google_service from auth.service_decorator import require_google_service
from core.server import server from core.server import server
from core.utils import handle_http_errors from core.utils import UserInputError, handle_http_errors
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -193,138 +193,155 @@ async def get_task_list(
raise Exception(message) raise Exception(message)
@server.tool() # type: ignore # --- Task list _impl functions ---
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("create_task_list", service_type="tasks") # type: ignore
async def create_task_list( async def _create_task_list_impl(
service: Resource, user_google_email: str, title: str service: Resource, user_google_email: str, title: str
) -> str: ) -> str:
""" """Implementation for creating a new task list."""
Create a new task list.
Args:
user_google_email (str): The user's Google email address. Required.
title (str): The title of the new task list.
Returns:
str: Confirmation message with the new task list ID and details.
"""
logger.info( logger.info(
f"[create_task_list] Invoked. Email: '{user_google_email}', Title: '{title}'" f"[create_task_list] Invoked. Email: '{user_google_email}', Title: '{title}'"
) )
try: body = {"title": title}
body = {"title": title}
result = await asyncio.to_thread(service.tasklists().insert(body=body).execute) result = await asyncio.to_thread(service.tasklists().insert(body=body).execute)
response = f"""Task List Created for {user_google_email}: response = f"""Task List Created for {user_google_email}:
- Title: {result["title"]} - Title: {result["title"]}
- ID: {result["id"]} - ID: {result["id"]}
- Created: {result.get("updated", "N/A")} - Created: {result.get("updated", "N/A")}
- Self Link: {result.get("selfLink", "N/A")}""" - Self Link: {result.get("selfLink", "N/A")}"""
logger.info( logger.info(
f"Created task list '{title}' with ID {result['id']} for {user_google_email}" f"Created task list '{title}' with ID {result['id']} for {user_google_email}"
) )
return response return response
except HttpError as error:
message = _format_reauth_message(error, user_google_email)
logger.error(message, exc_info=True)
raise Exception(message)
except Exception as e:
message = f"Unexpected error: {e}."
logger.exception(message)
raise Exception(message)
@server.tool() # type: ignore async def _update_task_list_impl(
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("update_task_list", service_type="tasks") # type: ignore
async def update_task_list(
service: Resource, user_google_email: str, task_list_id: str, title: str service: Resource, user_google_email: str, task_list_id: str, title: str
) -> str: ) -> str:
""" """Implementation for updating an existing task list."""
Update an existing task list.
Args:
user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the task list to update.
title (str): The new title for the task list.
Returns:
str: Confirmation message with updated task list details.
"""
logger.info( logger.info(
f"[update_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, New Title: '{title}'" f"[update_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, New Title: '{title}'"
) )
try: body = {"id": task_list_id, "title": title}
body = {"id": task_list_id, "title": title}
result = await asyncio.to_thread( result = await asyncio.to_thread(
service.tasklists().update(tasklist=task_list_id, body=body).execute service.tasklists().update(tasklist=task_list_id, body=body).execute
) )
response = f"""Task List Updated for {user_google_email}: response = f"""Task List Updated for {user_google_email}:
- Title: {result["title"]} - Title: {result["title"]}
- ID: {result["id"]} - ID: {result["id"]}
- Updated: {result.get("updated", "N/A")}""" - Updated: {result.get("updated", "N/A")}"""
logger.info( logger.info(
f"Updated task list {task_list_id} with new title '{title}' for {user_google_email}" f"Updated task list {task_list_id} with new title '{title}' for {user_google_email}"
) )
return response return response
except HttpError as error:
message = _format_reauth_message(error, user_google_email)
logger.error(message, exc_info=True)
raise Exception(message)
except Exception as e:
message = f"Unexpected error: {e}."
logger.exception(message)
raise Exception(message)
@server.tool() # type: ignore async def _delete_task_list_impl(
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("delete_task_list", service_type="tasks") # type: ignore
async def delete_task_list(
service: Resource, user_google_email: str, task_list_id: str service: Resource, user_google_email: str, task_list_id: str
) -> str: ) -> str:
""" """Implementation for deleting a task list."""
Delete a task list. Note: This will also delete all tasks in the list.
Args:
user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the task list to delete.
Returns:
str: Confirmation message.
"""
logger.info( logger.info(
f"[delete_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}" f"[delete_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}"
) )
try: await asyncio.to_thread(service.tasklists().delete(tasklist=task_list_id).execute)
await asyncio.to_thread(
service.tasklists().delete(tasklist=task_list_id).execute response = f"Task list {task_list_id} has been deleted for {user_google_email}. All tasks in this list have also been deleted."
logger.info(f"Deleted task list {task_list_id} for {user_google_email}")
return response
async def _clear_completed_tasks_impl(
service: Resource, user_google_email: str, task_list_id: str
) -> str:
"""Implementation for clearing completed tasks from a task list."""
logger.info(
f"[clear_completed_tasks] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}"
)
await asyncio.to_thread(service.tasks().clear(tasklist=task_list_id).execute)
response = f"All completed tasks have been cleared from task list {task_list_id} for {user_google_email}. The tasks are now hidden and won't appear in default task list views."
logger.info(
f"Cleared completed tasks from list {task_list_id} for {user_google_email}"
)
return response
# --- Consolidated manage_task_list tool ---
@server.tool() # type: ignore
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("manage_task_list", service_type="tasks") # type: ignore
async def manage_task_list(
service: Resource,
user_google_email: str,
action: str,
task_list_id: Optional[str] = None,
title: Optional[str] = None,
) -> str:
"""
Manage task lists: create, update, delete, or clear completed tasks.
Args:
user_google_email (str): The user's Google email address. Required.
action (str): The action to perform. Must be one of: "create", "update", "delete", "clear_completed".
task_list_id (Optional[str]): The ID of the task list. Required for "update", "delete", and "clear_completed" actions.
title (Optional[str]): The title for the task list. Required for "create" and "update" actions.
Returns:
str: Result of the requested action.
"""
logger.info(
f"[manage_task_list] Invoked. Email: '{user_google_email}', Action: '{action}'"
)
valid_actions = ("create", "update", "delete", "clear_completed")
if action not in valid_actions:
raise UserInputError(
f"Invalid action '{action}'. Must be one of: {', '.join(valid_actions)}"
) )
response = f"Task list {task_list_id} has been deleted for {user_google_email}. All tasks in this list have also been deleted." if action == "create":
if not title:
raise UserInputError("'title' is required for the 'create' action.")
return await _create_task_list_impl(service, user_google_email, title)
logger.info(f"Deleted task list {task_list_id} for {user_google_email}") if action == "update":
return response if not task_list_id:
raise UserInputError("'task_list_id' is required for the 'update' action.")
if not title:
raise UserInputError("'title' is required for the 'update' action.")
return await _update_task_list_impl(
service, user_google_email, task_list_id, title
)
except HttpError as error: if action == "delete":
message = _format_reauth_message(error, user_google_email) if not task_list_id:
logger.error(message, exc_info=True) raise UserInputError("'task_list_id' is required for the 'delete' action.")
raise Exception(message) return await _delete_task_list_impl(service, user_google_email, task_list_id)
except Exception as e:
message = f"Unexpected error: {e}." # action == "clear_completed"
logger.exception(message) if not task_list_id:
raise Exception(message) raise UserInputError(
"'task_list_id' is required for the 'clear_completed' action."
)
return await _clear_completed_tasks_impl(service, user_google_email, task_list_id)
# --- Task tools ---
@server.tool() # type: ignore @server.tool() # type: ignore
@@ -633,10 +650,10 @@ async def get_task(
raise Exception(message) raise Exception(message)
@server.tool() # type: ignore # --- Task _impl functions ---
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("create_task", service_type="tasks") # type: ignore
async def create_task( async def _create_task_impl(
service: Resource, service: Resource,
user_google_email: str, user_google_email: str,
task_list_id: str, task_list_id: str,
@@ -646,72 +663,45 @@ async def create_task(
parent: Optional[str] = None, parent: Optional[str] = None,
previous: Optional[str] = None, previous: Optional[str] = None,
) -> str: ) -> str:
""" """Implementation for creating a new task in a task list."""
Create a new task in a task list.
Args:
user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the task list to create the task in.
title (str): The title of the task.
notes (Optional[str]): Notes/description for the task.
due (Optional[str]): Due date in RFC 3339 format (e.g., "2024-12-31T23:59:59Z").
parent (Optional[str]): Parent task ID (for subtasks).
previous (Optional[str]): Previous sibling task ID (for positioning).
Returns:
str: Confirmation message with the new task ID and details.
"""
logger.info( logger.info(
f"[create_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Title: '{title}'" f"[create_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Title: '{title}'"
) )
try: body = {"title": title}
body = {"title": title} if notes:
if notes: body["notes"] = notes
body["notes"] = notes if due:
if due: body["due"] = due
body["due"] = due
params = {"tasklist": task_list_id, "body": body} params = {"tasklist": task_list_id, "body": body}
if parent: if parent:
params["parent"] = parent params["parent"] = parent
if previous: if previous:
params["previous"] = previous params["previous"] = previous
result = await asyncio.to_thread(service.tasks().insert(**params).execute) result = await asyncio.to_thread(service.tasks().insert(**params).execute)
response = f"""Task Created for {user_google_email}: response = f"""Task Created for {user_google_email}:
- Title: {result["title"]} - Title: {result["title"]}
- ID: {result["id"]} - ID: {result["id"]}
- Status: {result.get("status", "N/A")} - Status: {result.get("status", "N/A")}
- Updated: {result.get("updated", "N/A")}""" - Updated: {result.get("updated", "N/A")}"""
if result.get("due"): if result.get("due"):
response += f"\n- Due Date: {result['due']}" response += f"\n- Due Date: {result['due']}"
if result.get("notes"): if result.get("notes"):
response += f"\n- Notes: {result['notes']}" response += f"\n- Notes: {result['notes']}"
if result.get("webViewLink"): if result.get("webViewLink"):
response += f"\n- Web View Link: {result['webViewLink']}" response += f"\n- Web View Link: {result['webViewLink']}"
logger.info( logger.info(
f"Created task '{title}' with ID {result['id']} for {user_google_email}" f"Created task '{title}' with ID {result['id']} for {user_google_email}"
) )
return response return response
except HttpError as error:
message = _format_reauth_message(error, user_google_email)
logger.error(message, exc_info=True)
raise Exception(message)
except Exception as e:
message = f"Unexpected error: {e}."
logger.exception(message)
raise Exception(message)
@server.tool() # type: ignore async def _update_task_impl(
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("update_task", service_type="tasks") # type: ignore
async def update_task(
service: Resource, service: Resource,
user_google_email: str, user_google_email: str,
task_list_id: str, task_list_id: str,
@@ -721,126 +711,74 @@ async def update_task(
status: Optional[str] = None, status: Optional[str] = None,
due: Optional[str] = None, due: Optional[str] = None,
) -> str: ) -> str:
""" """Implementation for updating an existing task."""
Update an existing task.
Args:
user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the task list containing the task.
task_id (str): The ID of the task to update.
title (Optional[str]): New title for the task.
notes (Optional[str]): New notes/description for the task.
status (Optional[str]): New status ("needsAction" or "completed").
due (Optional[str]): New due date in RFC 3339 format.
Returns:
str: Confirmation message with updated task details.
"""
logger.info( logger.info(
f"[update_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}" f"[update_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}"
) )
try: # First get the current task to build the update body
# First get the current task to build the update body current_task = await asyncio.to_thread(
current_task = await asyncio.to_thread( service.tasks().get(tasklist=task_list_id, task=task_id).execute
service.tasks().get(tasklist=task_list_id, task=task_id).execute )
)
body = { body = {
"id": task_id, "id": task_id,
"title": title if title is not None else current_task.get("title", ""), "title": title if title is not None else current_task.get("title", ""),
"status": status "status": status
if status is not None if status is not None
else current_task.get("status", "needsAction"), else current_task.get("status", "needsAction"),
} }
if notes is not None: if notes is not None:
body["notes"] = notes body["notes"] = notes
elif current_task.get("notes"): elif current_task.get("notes"):
body["notes"] = current_task["notes"] body["notes"] = current_task["notes"]
if due is not None: if due is not None:
body["due"] = due body["due"] = due
elif current_task.get("due"): elif current_task.get("due"):
body["due"] = current_task["due"] body["due"] = current_task["due"]
result = await asyncio.to_thread( result = await asyncio.to_thread(
service.tasks() service.tasks().update(tasklist=task_list_id, task=task_id, body=body).execute
.update(tasklist=task_list_id, task=task_id, body=body) )
.execute
)
response = f"""Task Updated for {user_google_email}: response = f"""Task Updated for {user_google_email}:
- Title: {result["title"]} - Title: {result["title"]}
- ID: {result["id"]} - ID: {result["id"]}
- Status: {result.get("status", "N/A")} - Status: {result.get("status", "N/A")}
- Updated: {result.get("updated", "N/A")}""" - Updated: {result.get("updated", "N/A")}"""
if result.get("due"): if result.get("due"):
response += f"\n- Due Date: {result['due']}" response += f"\n- Due Date: {result['due']}"
if result.get("notes"): if result.get("notes"):
response += f"\n- Notes: {result['notes']}" response += f"\n- Notes: {result['notes']}"
if result.get("completed"): if result.get("completed"):
response += f"\n- Completed: {result['completed']}" response += f"\n- Completed: {result['completed']}"
logger.info(f"Updated task {task_id} for {user_google_email}") logger.info(f"Updated task {task_id} for {user_google_email}")
return response return response
except HttpError as error:
message = _format_reauth_message(error, user_google_email)
logger.error(message, exc_info=True)
raise Exception(message)
except Exception as e:
message = f"Unexpected error: {e}."
logger.exception(message)
raise Exception(message)
@server.tool() # type: ignore async def _delete_task_impl(
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("delete_task", service_type="tasks") # type: ignore
async def delete_task(
service: Resource, user_google_email: str, task_list_id: str, task_id: str service: Resource, user_google_email: str, task_list_id: str, task_id: str
) -> str: ) -> str:
""" """Implementation for deleting a task from a task list."""
Delete a task from a task list.
Args:
user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the task list containing the task.
task_id (str): The ID of the task to delete.
Returns:
str: Confirmation message.
"""
logger.info( logger.info(
f"[delete_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}" f"[delete_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}"
) )
try: await asyncio.to_thread(
await asyncio.to_thread( service.tasks().delete(tasklist=task_list_id, task=task_id).execute
service.tasks().delete(tasklist=task_list_id, task=task_id).execute )
)
response = f"Task {task_id} has been deleted from task list {task_list_id} for {user_google_email}." response = f"Task {task_id} has been deleted from task list {task_list_id} for {user_google_email}."
logger.info(f"Deleted task {task_id} for {user_google_email}") logger.info(f"Deleted task {task_id} for {user_google_email}")
return response return response
except HttpError as error:
message = _format_reauth_message(error, user_google_email)
logger.error(message, exc_info=True)
raise Exception(message)
except Exception as e:
message = f"Unexpected error: {e}."
logger.exception(message)
raise Exception(message)
@server.tool() # type: ignore async def _move_task_impl(
@require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("move_task", service_type="tasks") # type: ignore
async def move_task(
service: Resource, service: Resource,
user_google_email: str, user_google_email: str,
task_list_id: str, task_list_id: str,
@@ -849,105 +787,148 @@ async def move_task(
previous: Optional[str] = None, previous: Optional[str] = None,
destination_task_list: Optional[str] = None, destination_task_list: Optional[str] = None,
) -> str: ) -> str:
""" """Implementation for moving a task to a different position, parent, or list."""
Move a task to a different position or parent within the same list, or to a different list.
Args:
user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the current task list containing the task.
task_id (str): The ID of the task to move.
parent (Optional[str]): New parent task ID (for making it a subtask).
previous (Optional[str]): Previous sibling task ID (for positioning).
destination_task_list (Optional[str]): Destination task list ID (for moving between lists).
Returns:
str: Confirmation message with updated task details.
"""
logger.info( logger.info(
f"[move_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}" f"[move_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}"
) )
try: params = {"tasklist": task_list_id, "task": task_id}
params = {"tasklist": task_list_id, "task": task_id} if parent:
if parent: params["parent"] = parent
params["parent"] = parent if previous:
if previous: params["previous"] = previous
params["previous"] = previous if destination_task_list:
if destination_task_list: params["destinationTasklist"] = destination_task_list
params["destinationTasklist"] = destination_task_list
result = await asyncio.to_thread(service.tasks().move(**params).execute) result = await asyncio.to_thread(service.tasks().move(**params).execute)
response = f"""Task Moved for {user_google_email}: response = f"""Task Moved for {user_google_email}:
- Title: {result["title"]} - Title: {result["title"]}
- ID: {result["id"]} - ID: {result["id"]}
- Status: {result.get("status", "N/A")} - Status: {result.get("status", "N/A")}
- Updated: {result.get("updated", "N/A")}""" - Updated: {result.get("updated", "N/A")}"""
if result.get("parent"): if result.get("parent"):
response += f"\n- Parent Task ID: {result['parent']}" response += f"\n- Parent Task ID: {result['parent']}"
if result.get("position"): if result.get("position"):
response += f"\n- Position: {result['position']}" response += f"\n- Position: {result['position']}"
move_details = [] move_details = []
if destination_task_list: if destination_task_list:
move_details.append(f"moved to task list {destination_task_list}") move_details.append(f"moved to task list {destination_task_list}")
if parent: if parent:
move_details.append(f"made a subtask of {parent}") move_details.append(f"made a subtask of {parent}")
if previous: if previous:
move_details.append(f"positioned after {previous}") move_details.append(f"positioned after {previous}")
if move_details: if move_details:
response += f"\n- Move Details: {', '.join(move_details)}" response += f"\n- Move Details: {', '.join(move_details)}"
logger.info(f"Moved task {task_id} for {user_google_email}") logger.info(f"Moved task {task_id} for {user_google_email}")
return response return response
except HttpError as error:
message = _format_reauth_message(error, user_google_email) # --- Consolidated manage_task tool ---
logger.error(message, exc_info=True)
raise Exception(message)
except Exception as e:
message = f"Unexpected error: {e}."
logger.exception(message)
raise Exception(message)
@server.tool() # type: ignore @server.tool() # type: ignore
@require_google_service("tasks", "tasks") # type: ignore @require_google_service("tasks", "tasks") # type: ignore
@handle_http_errors("clear_completed_tasks", service_type="tasks") # type: ignore @handle_http_errors("manage_task", service_type="tasks") # type: ignore
async def clear_completed_tasks( async def manage_task(
service: Resource, user_google_email: str, task_list_id: str service: Resource,
user_google_email: str,
action: str,
task_list_id: str,
task_id: Optional[str] = None,
title: Optional[str] = None,
notes: Optional[str] = None,
status: Optional[str] = None,
due: Optional[str] = None,
parent: Optional[str] = None,
previous: Optional[str] = None,
destination_task_list: Optional[str] = None,
) -> str: ) -> str:
""" """
Clear all completed tasks from a task list. The tasks will be marked as hidden. Manage tasks: create, update, delete, or move tasks within task lists.
Args: Args:
user_google_email (str): The user's Google email address. Required. user_google_email (str): The user's Google email address. Required.
task_list_id (str): The ID of the task list to clear completed tasks from. action (str): The action to perform. Must be one of: "create", "update", "delete", "move".
task_list_id (str): The ID of the task list. Required for all actions.
task_id (Optional[str]): The ID of the task. Required for "update", "delete", and "move" actions.
title (Optional[str]): The title of the task. Required for "create", optional for "update".
notes (Optional[str]): Notes/description for the task. Used by "create" and "update" actions.
status (Optional[str]): Task status ("needsAction" or "completed"). Used by "update" action.
due (Optional[str]): Due date in RFC 3339 format (e.g., "2024-12-31T23:59:59Z"). Used by "create" and "update" actions.
parent (Optional[str]): Parent task ID (for subtasks). Used by "create" and "move" actions.
previous (Optional[str]): Previous sibling task ID (for positioning). Used by "create" and "move" actions.
destination_task_list (Optional[str]): Destination task list ID (for moving between lists). Used by "move" action.
Returns: Returns:
str: Confirmation message. str: Result of the requested action.
""" """
logger.info( logger.info(
f"[clear_completed_tasks] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}" f"[manage_task] Invoked. Email: '{user_google_email}', Action: '{action}', Task List ID: {task_list_id}"
) )
try: allowed_statuses = {"needsAction", "completed"}
await asyncio.to_thread(service.tasks().clear(tasklist=task_list_id).execute) if status is not None and status not in allowed_statuses:
raise UserInputError("invalid status: must be 'needsAction' or 'completed'")
response = f"All completed tasks have been cleared from task list {task_list_id} for {user_google_email}. The tasks are now hidden and won't appear in default task list views." valid_actions = ("create", "update", "delete", "move")
if action not in valid_actions:
logger.info( raise UserInputError(
f"Cleared completed tasks from list {task_list_id} for {user_google_email}" f"Invalid action '{action}'. Must be one of: {', '.join(valid_actions)}"
) )
return response
except HttpError as error: if action == "create":
message = _format_reauth_message(error, user_google_email) if status is not None:
logger.error(message, exc_info=True) raise UserInputError("'status' is only supported for the 'update' action.")
raise Exception(message) if not title:
except Exception as e: raise UserInputError("'title' is required for the 'create' action.")
message = f"Unexpected error: {e}." return await _create_task_impl(
logger.exception(message) service,
raise Exception(message) user_google_email,
task_list_id,
title,
notes=notes,
due=due,
parent=parent,
previous=previous,
)
if action == "update":
if status is not None and status not in allowed_statuses:
raise UserInputError("invalid status: must be 'needsAction' or 'completed'")
if not task_id:
raise UserInputError("'task_id' is required for the 'update' action.")
return await _update_task_impl(
service,
user_google_email,
task_list_id,
task_id,
title=title,
notes=notes,
status=status,
due=due,
)
if action == "delete":
if not task_id:
raise UserInputError("'task_id' is required for the 'delete' action.")
return await _delete_task_impl(
service, user_google_email, task_list_id, task_id
)
# action == "move"
if not task_id:
raise UserInputError("'task_id' is required for the 'move' action.")
return await _move_task_impl(
service,
user_google_email,
task_list_id,
task_id,
parent=parent,
previous=previous,
destination_task_list=destination_task_list,
)

2
manifest.json
View File

@@ -2,7 +2,7 @@
"dxt_version": "0.1", "dxt_version": "0.1",
"name": "workspace-mcp", "name": "workspace-mcp",
"display_name": "Google Workspace MCP", "display_name": "Google Workspace MCP",
"version": "1.13.1", "version": "1.14.0",
"description": "Full natural language control over Google Calendar, Drive, Gmail, Docs, Sheets, Slides, Forms, Tasks, Chat and Custom Search through all MCP clients, AI assistants and developer tools", "description": "Full natural language control over Google Calendar, Drive, Gmail, Docs, Sheets, Slides, Forms, Tasks, Chat and Custom Search through all MCP clients, AI assistants and developer tools",
"long_description": "A production-ready MCP server that integrates all major Google Workspace services with AI assistants. Includes Google PSE integration for custom web searches.", "long_description": "A production-ready MCP server that integrates all major Google Workspace services with AI assistants. Includes Google PSE integration for custom web searches.",
"author": { "author": {

3
pyproject.toml
View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "workspace-mcp" name = "workspace-mcp"
version = "1.13.1" version = "1.14.0"
description = "Comprehensive, highly performant Google Workspace Streamable HTTP & SSE MCP Server for Calendar, Gmail, Docs, Sheets, Slides & Drive" description = "Comprehensive, highly performant Google Workspace Streamable HTTP & SSE MCP Server for Calendar, Gmail, Docs, Sheets, Slides & Drive"
readme = "README.md" readme = "README.md"
keywords = [ "mcp", "google", "workspace", "llm", "ai", "claude", "model", "context", "protocol", "server"] keywords = [ "mcp", "google", "workspace", "llm", "ai", "claude", "model", "context", "protocol", "server"]
@@ -22,6 +22,7 @@ dependencies = [
"python-dotenv>=1.1.0", "python-dotenv>=1.1.0",
"pyyaml>=6.0.2", "pyyaml>=6.0.2",
"cryptography>=45.0.0", "cryptography>=45.0.0",
"defusedxml>=0.7.1",
] ]
classifiers = [ classifiers = [
"Development Status :: 4 - Beta", "Development Status :: 4 - Beta",

4
server.json
View File

@@ -3,7 +3,7 @@
"name": "io.github.taylorwilsdon/workspace-mcp", "name": "io.github.taylorwilsdon/workspace-mcp",
"description": "Google Workspace MCP server for Gmail, Drive, Calendar, Docs, Sheets, Slides, Forms, Tasks, Chat.", "description": "Google Workspace MCP server for Gmail, Drive, Calendar, Docs, Sheets, Slides, Forms, Tasks, Chat.",
"status": "active", "status": "active",
"version": "1.13.1", "version": "1.14.0",
"packages": [ "packages": [
{ {
"registryType": "pypi", "registryType": "pypi",
@@ -11,7 +11,7 @@
"transport": { "transport": {
"type": "stdio" "type": "stdio"
}, },
"version": "1.13.1" "version": "1.14.0"
} }
] ]
} }

13
tests/gcontacts/test_contacts_tools.py
View File

@@ -291,9 +291,7 @@ class TestImports:
assert hasattr(contacts_tools, "list_contacts") assert hasattr(contacts_tools, "list_contacts")
assert hasattr(contacts_tools, "get_contact") assert hasattr(contacts_tools, "get_contact")
assert hasattr(contacts_tools, "search_contacts") assert hasattr(contacts_tools, "search_contacts")
assert hasattr(contacts_tools, "create_contact") assert hasattr(contacts_tools, "manage_contact")
assert hasattr(contacts_tools, "update_contact")
assert hasattr(contacts_tools, "delete_contact")
def test_import_group_tools(self): def test_import_group_tools(self):
"""Test that group tools can be imported.""" """Test that group tools can be imported."""
@@ -301,18 +299,13 @@ class TestImports:
assert hasattr(contacts_tools, "list_contact_groups") assert hasattr(contacts_tools, "list_contact_groups")
assert hasattr(contacts_tools, "get_contact_group") assert hasattr(contacts_tools, "get_contact_group")
assert hasattr(contacts_tools, "create_contact_group") assert hasattr(contacts_tools, "manage_contact_group")
assert hasattr(contacts_tools, "update_contact_group")
assert hasattr(contacts_tools, "delete_contact_group")
assert hasattr(contacts_tools, "modify_contact_group_members")
def test_import_batch_tools(self): def test_import_batch_tools(self):
"""Test that batch tools can be imported.""" """Test that batch tools can be imported."""
from gcontacts import contacts_tools from gcontacts import contacts_tools
assert hasattr(contacts_tools, "batch_create_contacts") assert hasattr(contacts_tools, "manage_contacts_batch")
assert hasattr(contacts_tools, "batch_update_contacts")
assert hasattr(contacts_tools, "batch_delete_contacts")
class TestConstants: class TestConstants:

11
uv.lock generated
View File

@@ -404,6 +404,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1c/7c/996760c30f1302704af57c66ff2d723f7d656d0d0b93563b5528a51484bb/cyclopts-4.5.1-py3-none-any.whl", hash = "sha256:0642c93601e554ca6b7b9abd81093847ea4448b2616280f2a0952416574e8c7a", size = 199807 }, { url = "https://files.pythonhosted.org/packages/1c/7c/996760c30f1302704af57c66ff2d723f7d656d0d0b93563b5528a51484bb/cyclopts-4.5.1-py3-none-any.whl", hash = "sha256:0642c93601e554ca6b7b9abd81093847ea4448b2616280f2a0952416574e8c7a", size = 199807 },
] ]
[[package]]
name = "defusedxml"
version = "0.7.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/0f/d5/c66da9b79e5bdb124974bfe172b4daf3c984ebd9c2a06e2b8a4dc7331c72/defusedxml-0.7.1.tar.gz", hash = "sha256:1bb3032db185915b62d7c6209c5a8792be6a32ab2fedacc84e01b52c51aa3e69", size = 75520 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604 },
]
[[package]] [[package]]
name = "dnspython" name = "dnspython"
version = "2.8.0" version = "2.8.0"
@@ -2039,6 +2048,7 @@ version = "1.13.1"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "cryptography" }, { name = "cryptography" },
{ name = "defusedxml" },
{ name = "fastapi" }, { name = "fastapi" },
{ name = "fastmcp" }, { name = "fastmcp" },
{ name = "google-api-python-client" }, { name = "google-api-python-client" },
@@ -2098,6 +2108,7 @@ valkey = [
[package.metadata] [package.metadata]
requires-dist = [ requires-dist = [
{ name = "cryptography", specifier = ">=45.0.0" }, { name = "cryptography", specifier = ">=45.0.0" },
{ name = "defusedxml", specifier = ">=0.7.1" },
{ name = "fastapi", specifier = ">=0.115.12" }, { name = "fastapi", specifier = ">=0.115.12" },
{ name = "fastmcp", specifier = ">=3.0.2" }, { name = "fastmcp", specifier = ">=3.0.2" },
{ name = "google-api-python-client", specifier = ">=2.168.0" }, { name = "google-api-python-client", specifier = ">=2.168.0" },