gmail_search_personal

Instructions

Implementation Reference

• tools.py:30-36 (handler)

  The tool handler for gmail_search_personal. Decorated with @mcp.tool(). Loads personal account credentials, then calls gmail_search() from api.py.

  @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:30-30 (registration)

  Tool registration via FastMCP @mcp.tool() decorator on the gmail_search_personal function.

  @mcp.tool()

• tools.py:16-21 (helper)

  Helper that loads credentials for the given account and wraps RuntimeError into an error envelope.

  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-95 (helper)

  Core Gmail search logic called by gmail_search_personal. Builds Gmail API service, lists messages matching query, fetches metadata for each, and returns results.

  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:11-12 (schema)

  Type definition for Account (Literal['personal', 'work']) used in the tool signature and throughout the codebase.

  Account = Literal["personal", "work"]