gmail_search_work

Instructions

Implementation Reference

• tools.py:39-45 (handler)

  The gmail_search_work MCP tool handler. It loads credentials for the 'work' account via _load_or_error, then delegates to the gmail_search function in api.py.

  @mcp.tool()
  def gmail_search_work(query: str, max_results: int = 20) -> dict[str, Any]:
      """Search Gmail for the configured work account."""
      result = _load_or_error("work")
      if isinstance(result, dict):
          return result
      return gmail_search(result, query, max_results)

• tools.py:30-36 (registration)

  Tool registration via @mcp.tool() decorator (line 39). Also shows gmail_search_personal as a sibling for context.

  @mcp.tool()
  def gmail_search_personal(query: str, max_results: int = 20) -> dict[str, Any]:
      """Search Gmail for the configured personal account."""
      result = _load_or_error("personal")
      if isinstance(result, dict):
          return result
      return gmail_search(result, query, max_results)

• tools.py:16-21 (helper)

  The _load_or_error helper used by gmail_search_work to load credentials and handle errors.

  def _load_or_error(account: Account):
      """Return credentials or an error envelope if Keychain lookup fails."""
      try:
          return load_credentials(account)
      except RuntimeError as e:
          return {"ok": False, "error": str(e), "code": 401}

• api.py:35-96 (helper)

  The gmail_search function that performs the actual Gmail API search. Called by gmail_search_work after credentials are loaded.

  def gmail_search(
      creds: Credentials,
      query: str,
      max_results: int = 20,
  ) -> dict[str, Any]:
      """Search Gmail and return message summaries."""
      from google.auth.exceptions import RefreshError
      from googleapiclient.discovery import build
  
      max_results = min(max_results, 50)
  
      try:
          service = build("gmail", "v1", credentials=creds)
  
          list_response = (
              service.users()
              .messages()
              .list(userId="me", q=query, maxResults=max_results)
              .execute(num_retries=3)
          )
  
          messages = list_response.get("messages", [])
          results: list[dict[str, Any]] = []
  
          for msg in messages:
              detail = (
                  service.users()
                  .messages()
                  .get(
                      userId="me",
                      id=msg["id"],
                      format="metadata",
                      metadataHeaders=["From", "Subject", "Date"],
                  )
                  .execute(num_retries=3)
              )
  
              headers = {
                  h["name"]: h["value"] for h in detail.get("payload", {}).get("headers", [])
              }
  
              results.append(
                  {
                      "id": detail["id"],
                      "thread_id": detail["threadId"],
                      "from": headers.get("From", ""),
                      "subject": headers.get("Subject", ""),
                      "date": headers.get("Date", ""),
                      "snippet": detail.get("snippet", ""),
                      "labels": detail.get("labelIds", []),
                  }
              )
  
          return {"ok": True, "data": results}
  
      except HttpError as e:
          return _parse_http_error(e)
      except RefreshError as e:
          return {"ok": False, "error": f"token refresh failed: {e}", "code": 401}
      except Exception as e:
          return {"ok": False, "error": f"upstream failure: {type(e).__name__}", "code": 503}

• config.py:1-39 (schema)

  Configuration including Account type literal ('personal'|'work') and KEYCHAIN_SERVICES mapping used to resolve 'work' credentials.

  """Shared configuration for the google-mcp server."""
  from __future__ import annotations
  
  import os
  from pathlib import Path
  from typing import Literal
  
  _HERE = Path(__file__).parent
  CREDENTIALS_PATH = _HERE / "credentials.json"
  
  Account = Literal["personal", "work"]
  
  KEYCHAIN_SERVICES: dict[Account, str] = {
      "personal": "google-mcp-personal",
      "work": "google-mcp-work",
  }
  
  # Narrowest scopes covering the four tools: Gmail metadata/snippet search,
  # calendar listing (to filter by accessRole), and event reads.
  SCOPES = [
      "https://www.googleapis.com/auth/gmail.readonly",
      "https://www.googleapis.com/auth/calendar.calendarlist.readonly",
      "https://www.googleapis.com/auth/calendar.events.readonly",
  ]
  
  ACCOUNT_EMAIL_ENVS: dict[Account, str] = {
      "personal": "GOOGLE_MCP_PERSONAL_EMAIL",
      "work": "GOOGLE_MCP_WORK_EMAIL",
  }
  
  WORK_CALENDAR_FILTER_ENV = "GOOGLE_MCP_WORK_CALENDAR"
  
  
  def get_expected_email(account: Account) -> str | None:
      """Return expected email for account from environment, if configured."""
      env_name = ACCOUNT_EMAIL_ENVS[account]
      value = os.getenv(env_name, "").strip()
      return value or None