# Copyright (c) 2025 Dedalus Labs, Inc. and its contributors
# SPDX-License-Identifier: MIT
"""Gmail API operations for MCP server.
Read-only access to the Gmail API v1 using OAuth Bearer tokens.
Auth options:
- Preferred: run `uv run python -m src.gmail_auth` once to store tokens locally,
then gmail-mcp will auto-refresh tokens as needed.
- Advanced/manual: set `GMAIL_ACCESS_TOKEN` directly.
"""
from __future__ import annotations
from typing import Any
from urllib.parse import urlencode
from dotenv import load_dotenv
from pydantic import BaseModel
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, SecretKeys
load_dotenv()
# --- Connection --------------------------------------------------------------
def get_gmail_connection(*, interactive_auth: bool = False) -> Connection:
"""Create the Dedalus connection for Gmail."""
from .gmail_oauth import ensure_gmail_access_token
ensure_gmail_access_token(interactive=interactive_auth)
return Connection(
name="gmail",
secrets=SecretKeys(token="GMAIL_ACCESS_TOKEN"),
base_url="https://gmail.googleapis.com/gmail/v1",
auth_header_format="Bearer {api_key}",
)
# --- Response Models ---------------------------------------------------------
class GmailResult(BaseModel):
success: bool
data: Any = None
error: str | None = None
# --- Helper ------------------------------------------------------------------
def _encode_params(params: dict[str, Any] | None) -> str:
if not params:
return ""
# Remove None values and support list params via doseq=True.
cleaned: dict[str, Any] = {k: v for k, v in params.items() if v is not None}
return urlencode(cleaned, doseq=True)
async def _request(
method: HttpMethod,
path: str,
params: dict[str, Any] | None = None,
body: dict[str, Any] | None = None,
) -> GmailResult:
"""Make a Gmail API request via the enclave dispatch."""
ctx = get_context()
qs = _encode_params(params)
if qs:
path = f"{path}?{qs}"
request = HttpRequest(method=method, path=path, body=body)
response = await ctx.dispatch("gmail", request)
if response.success:
return GmailResult(success=True, data=response.response.body)
msg = response.error.message if response.error else "Request failed"
return GmailResult(success=False, error=msg)
# --- Tools -------------------------------------------------------------------
@tool(description="Get the authenticated user's Gmail profile")
async def gmail_get_profile() -> GmailResult:
return await _request(HttpMethod.GET, "/users/me/profile")
@tool(description="List labels in the user's mailbox")
async def gmail_list_labels() -> GmailResult:
return await _request(HttpMethod.GET, "/users/me/labels")
@tool(description="List messages in the user's mailbox")
async def gmail_list_messages(
query: str | None = None,
label_ids: list[str] | None = None,
max_results: int = 25,
page_token: str | None = None,
include_spam_trash: bool = False,
) -> GmailResult:
"""List messages.
Args:
query: Gmail search query (same as Gmail web search operators)
label_ids: Restrict to messages with these label IDs
max_results: 1-500 (default 25)
page_token: token returned by a previous call
include_spam_trash: include spam/trash in results
"""
max_results = max(1, min(500, max_results))
params: dict[str, Any] = {
"q": query,
"labelIds": label_ids,
"maxResults": str(max_results),
"pageToken": page_token,
"includeSpamTrash": str(include_spam_trash).lower(),
}
return await _request(HttpMethod.GET, "/users/me/messages", params=params)
@tool(description="Get a message by ID")
async def gmail_get_message(
message_id: str,
format: str = "metadata",
) -> GmailResult:
"""Get a message.
Args:
message_id: Gmail message ID
format: one of 'metadata', 'full', 'minimal', or 'raw'
"""
return await _request(
HttpMethod.GET,
f"/users/me/messages/{message_id}",
params={"format": format},
)
@tool(description="Get an entire thread by ID")
async def gmail_get_thread(thread_id: str, format: str = "metadata") -> GmailResult:
return await _request(
HttpMethod.GET,
f"/users/me/threads/{thread_id}",
params={"format": format},
)
@tool(description="Get an attachment by message ID and attachment ID")
async def gmail_get_attachment(message_id: str, attachment_id: str) -> GmailResult:
return await _request(
HttpMethod.GET,
f"/users/me/messages/{message_id}/attachments/{attachment_id}",
)
# --- Export ------------------------------------------------------------------
gmail_tools = [
gmail_get_profile,
gmail_list_labels,
gmail_list_messages,
gmail_get_message,
gmail_get_thread,
gmail_get_attachment,
]
