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

add instructions for llm if api not enabled

Browse Source
This commit is contained in:
Taylor Wilsdon
2025-07-28 11:49:01 -04:00
parent 18bf5c5ebf
commit b4743d2f16
13 changed files with 545 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
core/api_enablement.py Normal file
View File

@@ -0,0 +1,100 @@
import re
from typing import Dict, Optional, Tuple
API_ENABLEMENT_LINKS: Dict[str, str] = {
"calendar-json.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=calendar-json.googleapis.com",
"drive.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=drive.googleapis.com",
"gmail.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=gmail.googleapis.com",
"docs.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=docs.googleapis.com",
"sheets.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=sheets.googleapis.com",
"slides.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=slides.googleapis.com",
"forms.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=forms.googleapis.com",
"tasks.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=tasks.googleapis.com",
"chat.googleapis.com": "https://console.cloud.google.com/flows/enableapi?apiid=chat.googleapis.com",
}
SERVICE_NAME_TO_API: Dict[str, str] = {
"Google Calendar": "calendar-json.googleapis.com",
"Google Drive": "drive.googleapis.com",
"Gmail": "gmail.googleapis.com",
"Google Docs": "docs.googleapis.com",
"Google Sheets": "sheets.googleapis.com",
"Google Slides": "slides.googleapis.com",
"Google Forms": "forms.googleapis.com",
"Google Tasks": "tasks.googleapis.com",
"Google Chat": "chat.googleapis.com",
}
INTERNAL_SERVICE_TO_API: Dict[str, str] = {
"calendar": "calendar-json.googleapis.com",
"drive": "drive.googleapis.com",
"gmail": "gmail.googleapis.com",
"docs": "docs.googleapis.com",
"sheets": "sheets.googleapis.com",
"slides": "slides.googleapis.com",
"forms": "forms.googleapis.com",
"tasks": "tasks.googleapis.com",
"chat": "chat.googleapis.com",
}
def extract_api_info_from_error(error_details: str) -> Tuple[Optional[str], Optional[str]]:
"""
Extract API service and project ID from error details.
Returns:
Tuple of (api_service, project_id) or (None, None) if not found
"""
api_pattern = r'https://console\.developers\.google\.com/apis/api/([^/]+)/overview'
project_pattern = r'project[=\s]+([a-zA-Z0-9-]+)'
api_match = re.search(api_pattern, error_details)
project_match = re.search(project_pattern, error_details)
api_service = api_match.group(1) if api_match else None
project_id = project_match.group(1) if project_match else None
return api_service, project_id
def get_api_enablement_message(error_details: str, service_type: Optional[str] = None) -> str:
"""
Generate a helpful error message with direct API enablement link.
Args:
error_details: The error details string from the HttpError
service_type: Optional service type (e.g., "calendar", "gmail", or "Google Calendar")
Returns:
Formatted error message with enablement link
"""
api_service, project_id = extract_api_info_from_error(error_details)
if not api_service and service_type:
# Check internal service names first (e.g., "calendar", "gmail")
api_service = INTERNAL_SERVICE_TO_API.get(service_type)
if not api_service:
# Check display names (e.g., "Google Calendar")
api_service = SERVICE_NAME_TO_API.get(service_type)
if api_service and api_service in API_ENABLEMENT_LINKS:
enable_link = API_ENABLEMENT_LINKS[api_service]
service_display_name = next(
(name for name, api in SERVICE_NAME_TO_API.items() if api == api_service),
api_service
)
message = (
f"{service_display_name} API is not enabled for your project"
f"{f' ({project_id})' if project_id else ''}.\n\n"
f"Enable it here: {enable_link}\n\n"
f"After enabling, wait a few moments for the change to propagate, then try again."
f"IMPORTANT - LLM: share the link provided as a clickable hyperlink and instruct the user to enable the required API"
)
return message
return ""

24
core/comments.py
View File

@@ -37,75 +37,75 @@ def create_comment_tools(app_name: str, file_id_param: str):
# Create functions without decorators first, then apply decorators with proper names
if file_id_param == "document_id":
@require_google_service("drive", "drive_read")
@handle_http_errors(read_func_name)
@handle_http_errors(read_func_name, service_type="drive")
async def read_comments(service, user_google_email: str, document_id: str) -> str:
"""Read all comments from a Google Document."""
return await _read_comments_impl(service, app_name, document_id)
@require_google_service("drive", "drive_file")
@handle_http_errors(create_func_name)
@handle_http_errors(create_func_name, service_type="drive")
async def create_comment(service, user_google_email: str, document_id: str, comment_content: str) -> str:
"""Create a new comment on a Google Document."""
return await _create_comment_impl(service, app_name, document_id, comment_content)
@require_google_service("drive", "drive_file")
@handle_http_errors(reply_func_name)
@handle_http_errors(reply_func_name, service_type="drive")
async def reply_to_comment(service, user_google_email: str, document_id: str, comment_id: str, reply_content: str) -> str:
"""Reply to a specific comment in a Google Document."""
return await _reply_to_comment_impl(service, app_name, document_id, comment_id, reply_content)
@require_google_service("drive", "drive_file")
@handle_http_errors(resolve_func_name)
@handle_http_errors(resolve_func_name, service_type="drive")
async def resolve_comment(service, user_google_email: str, document_id: str, comment_id: str) -> str:
"""Resolve a comment in a Google Document."""
return await _resolve_comment_impl(service, app_name, document_id, comment_id)
elif file_id_param == "spreadsheet_id":
@require_google_service("drive", "drive_read")
@handle_http_errors(read_func_name)
@handle_http_errors(read_func_name, service_type="drive")
async def read_comments(service, user_google_email: str, spreadsheet_id: str) -> str:
"""Read all comments from a Google Spreadsheet."""
return await _read_comments_impl(service, app_name, spreadsheet_id)
@require_google_service("drive", "drive_file")
@handle_http_errors(create_func_name)
@handle_http_errors(create_func_name, service_type="drive")
async def create_comment(service, user_google_email: str, spreadsheet_id: str, comment_content: str) -> str:
"""Create a new comment on a Google Spreadsheet."""
return await _create_comment_impl(service, app_name, spreadsheet_id, comment_content)
@require_google_service("drive", "drive_file")
@handle_http_errors(reply_func_name)
@handle_http_errors(reply_func_name, service_type="drive")
async def reply_to_comment(service, user_google_email: str, spreadsheet_id: str, comment_id: str, reply_content: str) -> str:
"""Reply to a specific comment in a Google Spreadsheet."""
return await _reply_to_comment_impl(service, app_name, spreadsheet_id, comment_id, reply_content)
@require_google_service("drive", "drive_file")
@handle_http_errors(resolve_func_name)
@handle_http_errors(resolve_func_name, service_type="drive")
async def resolve_comment(service, user_google_email: str, spreadsheet_id: str, comment_id: str) -> str:
"""Resolve a comment in a Google Spreadsheet."""
return await _resolve_comment_impl(service, app_name, spreadsheet_id, comment_id)
elif file_id_param == "presentation_id":
@require_google_service("drive", "drive_read")
@handle_http_errors(read_func_name)
@handle_http_errors(read_func_name, service_type="drive")
async def read_comments(service, user_google_email: str, presentation_id: str) -> str:
"""Read all comments from a Google Presentation."""
return await _read_comments_impl(service, app_name, presentation_id)
@require_google_service("drive", "drive_file")
@handle_http_errors(create_func_name)
@handle_http_errors(create_func_name, service_type="drive")
async def create_comment(service, user_google_email: str, presentation_id: str, comment_content: str) -> str:
"""Create a new comment on a Google Presentation."""
return await _create_comment_impl(service, app_name, presentation_id, comment_content)
@require_google_service("drive", "drive_file")
@handle_http_errors(reply_func_name)
@handle_http_errors(reply_func_name, service_type="drive")
async def reply_to_comment(service, user_google_email: str, presentation_id: str, comment_id: str, reply_content: str) -> str:
"""Reply to a specific comment in a Google Presentation."""
return await _reply_to_comment_impl(service, app_name, presentation_id, comment_id, reply_content)
@require_google_service("drive", "drive_file")
@handle_http_errors(resolve_func_name)
@handle_http_errors(resolve_func_name, service_type="drive")
async def resolve_comment(service, user_google_email: str, presentation_id: str, comment_id: str) -> str:
"""Resolve a comment in a Google Presentation."""
return await _resolve_comment_impl(service, app_name, presentation_id, comment_id)

35
core/utils.py
View File

@@ -10,6 +10,7 @@ import functools
from typing import List, Optional
from googleapiclient.errors import HttpError
from .api_enablement import get_api_enablement_message
logger = logging.getLogger(__name__)
@@ -233,7 +234,7 @@ def extract_office_xml_text(file_bytes: bytes, mime_type: str) -> Optional[str]:
return None
def handle_http_errors(tool_name: str, is_read_only: bool = False):
def handle_http_errors(tool_name: str, is_read_only: bool = False, service_type: Optional[str] = None):
"""
A decorator to handle Google API HttpErrors and transient SSL errors in a standardized way.
@@ -247,6 +248,7 @@ def handle_http_errors(tool_name: str, is_read_only: bool = False):
tool_name (str): The name of the tool being decorated (e.g., 'list_calendars').
is_read_only (bool): If True, the operation is considered safe to retry on
transient network errors. Defaults to False.
service_type (str): Optional. The Google service type (e.g., 'calendar', 'gmail').
"""
def decorator(func):
@@ -275,12 +277,31 @@ def handle_http_errors(tool_name: str, is_read_only: bool = False):
) from e
except HttpError as error:
user_google_email = kwargs.get("user_google_email", "N/A")
message = (
f"API error in {tool_name}: {error}. "
f"You might need to re-authenticate for user '{user_google_email}'. "
f"LLM: Try 'start_google_auth' with the user's email and the appropriate service_name."
)
logger.error(message, exc_info=True)
error_details = str(error)
# Check if this is an API not enabled error
if error.resp.status == 403 and "accessNotConfigured" in error_details:
enablement_msg = get_api_enablement_message(error_details, service_type)
if enablement_msg:
message = (
f"API error in {tool_name}: {enablement_msg}\n\n"
f"User: {user_google_email}"
)
else:
message = (
f"API error in {tool_name}: {error}. "
f"The required API is not enabled for your project. "
f"Please check the Google Cloud Console to enable it."
)
else:
message = (
f"API error in {tool_name}: {error}. "
f"You might need to re-authenticate for user '{user_google_email}'. "
f"LLM: Try 'start_google_auth' with the user's email and the appropriate service_name."
)
logger.error(f"API error in {tool_name}: {error}", exc_info=True)
raise Exception(message) from error
except TransientNetworkError:
# Re-raise without wrapping to preserve the specific error type