| #!/usr/bin/env python3 | | | """ | | | Gmail CLI tool for Claude Code skill. | | | Supports multiple Gmail accounts via separate credential files. | | | Uses the Google Gmail API directly (no third-party wrappers). | | | """ | | | import argparse | | | import base64 | | | import json | | | import os | | | import sys | | | from pathlib import Path | | | # Config directory for storing credentials | | | CONFIG_DIR = Path.home() / ".claude" / "skills" / "gmail" / "config" | | | # Account aliases mapping to credential files | | | # Customize these with your own email addresses | | | ACCOUNTS = { | | | "personal": "you@gmail.com", | | | "work": "you@company.com", | | | "business": "you@business.io", | | | } | | | def _resolve_account(account: str): | | | """Resolve account alias to email and safe filename.""" | |
| email = ACCOUNTS.get(account, account) | |
| safe_name = email.replace("@", "_at_").replace(".", "_") | |
| return email, safe_name | | | def _run_oauth_flow(client_secret_file: str, token_file: str, email: str): | | | """Perform OAuth flow and save token as plain JSON.""" | | | from google_auth_oauthlib.flow import InstalledAppFlow | |
| SCOPES = ['https://mail.google.com/'] | |
| print(f"\nAuthenticating: {email}", file=sys.stderr) | |
| print(f"A browser window will open — sign in with: {email}\n", file=sys.stderr) | |
| flow = InstalledAppFlow.from_client_secrets_file(client_secret_file, SCOPES) | | | creds = flow.run_local_server(port=0) | | | # Save as plain JSON (compatible with get_service) | | | token_data = { | | | "access_token": creds.token, | | | "refresh_token": creds.refresh_token, | | | "token_uri": creds.token_uri or "https://oauth2.googleapis.com/token", | | | "client_id": creds.client_id, | | | "client_secret": creds.client_secret, | | | } | |
| with open(token_file, "w") as f: | |
| json.dump(token_data, f, indent=2) | |
| print(f"Token saved for {email}\n", file=sys.stderr) | |
| def get_service(account: str): | |
| """Get a Gmail API service client for the given account. | | | Triggers OAuth flow if no token exists yet.""" | | | from google.oauth2.credentials import Credentials | | | from googleapiclient.discovery import build | | | email, safe_name = resolve_account(account) | | | client_secret = CONFIG_DIR / f"client_secret{safe_name}.json" | |
| token_path = CONFIG_DIR / f"token_{safe_name}.json" | |
| if not client_secret.exists(): | |
| print(f"ERROR: No credentials found for {email}", file=sys.stderr) | |
| print(f"Expected: {client_secret}", file=sys.stderr) | |
| print(f"\nRun 'gmail_cli.py setup' for instructions.", file=sys.stderr) | |
| sys.exit(1) | |
| if not token_path.exists(): | |
| _run_oauth_flow(str(client_secret), str(token_path), email) | |
| with open(token_path) as f: | |
| td = json.load(f) | |
| creds = Credentials( | |
| token=td.get("access_token"), | |
| refresh_token=td.get("refresh_token"), | |
| token_uri=td.get("token_uri", "https://oauth2.googleapis.com/token"), | |
| client_id=td.get("client_id"), | |
| client_secret=td.get("client_secret"), | |
| ) | |
| return build("gmail", "v1", credentials=creds, cache_discovery=False) | |
| def _extract_body(payload): | |
| """Extract text/plain body from a Gmail API message payload.""" | |
| if payload.get("mimeType") == "text/plain": | |
| data = payload["body"].get("data", "") | |
| if data: | |
| return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace") | |
| for part in payload.get("parts", []): | |
| result = _extract_body(part) | |
| if result: | | | return result | | | return "" | | | def _extract_html(payload): | | | """Extract text/html body from a Gmail API message payload.""" | |
| if payload.get("mimeType") == "text/html": | |
| data = payload["body"].get("data", "") | |
| if data: | |
| return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace") | |
| for part in payload.get("parts", []): | |
| result = _extract_html(part) | |
| if result: | | | return result | | | return "" | | | def _extract_attachments(payload, attachments=None): | | | """Extract attachment filenames from a Gmail API message payload.""" | | | if attachments is None: | |
| attachments = [] | |
| filename = payload.get("filename") | |
| if filename: | |
| attachments.append(filename) | |
| for part in payload.get("parts", []): | |
| _extract_attachments(part, attachments) | | | return attachments | | | def cmd_list(args): | | | """List recent emails.""" | | | service = get_service(args.account) | | | # Build query | | | query_parts = ["in:inbox"] | | | if args.unread: | | | query_parts.append("is:unread") | | | if args.starred: | | | query_parts = ["is:starred"] | | | if args.label: | |
| query_parts = [f"label:{args.label}"] | |
| query_parts.append("newer_than:30d") | |
| query = " ".join(query_parts) | |
| try: | | | result = service.users().messages().list(userId="me", q=query, maxResults=args.limit).execute() | | | except Exception as e: | |
| print(f"ERROR: List failed: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| msg_refs = result.get("messages", []) | |
| if args.json: | | | output = [] | | | for ref in msg_refs: | |
| msg = service.users().messages().get(userId="me", id=ref["id"], format="metadata", | |
| metadataHeaders=["From", "To", "Subject", "Date"]).execute() | |
| headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])} | |
| labels = msg.get("labelIds", []) | |
| output.append({ | |
| "id": msg["id"], | |
| "thread_id": msg["threadId"], | |
| "from": headers.get("From", ""), | |
| "to": headers.get("To", ""), | |
| "subject": headers.get("Subject", ""), | |
| "date": headers.get("Date", ""), | |
| "snippet": msg.get("snippet", ""), | |
| "labels": labels, | | | "unread": "UNREAD" in labels, | |
| }) | |
| print(json.dumps(output, indent=2, default=str)) | |
| else: | | | for ref in msg_refs: | |
| msg = service.users().messages().get(userId="me", id=ref["id"], format="metadata", | |
| metadataHeaders=["From", "Subject", "Date"]).execute() | |
| headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])} | |
| labels = msg.get("labelIds", []) | |
| status = "[UNREAD]" if "UNREAD" in labels else "" | |
| print(f"{status} {headers.get('Date', '?')} | From: {headers.get('From', '?')}") | |
| print(f" Subject: {headers.get('Subject', '?')}") | |
| print(f" ID: {msg['id']}") | |
| print() | |
| def cmd_read(args): | |
| """Read a specific email by ID.""" | | | service = get_service(args.account) | | | try: | | | msg = service.users().messages().get(userId="me", id=args.message_id, format="full").execute() | | | except Exception as e: | |
| print(f"ERROR: Could not read message {args.message_id}: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]} | |
| body_plain = _extract_body(msg["payload"]) | |
| body_html = _extract_html(msg["payload"]) | |
| attachments = _extract_attachments(msg["payload"]) | |
| if args.json: | |
| output = { | |
| "id": msg["id"], | |
| "thread_id": msg["threadId"], | |
| "from": headers.get("From", ""), | |
| "to": headers.get("To", ""), | |
| "cc": headers.get("Cc", ""), | |
| "subject": headers.get("Subject", ""), | |
| "date": headers.get("Date", ""), | |
| "body_plain": body_plain, | | | "body_html": body_html, | | | "attachments": attachments, | | | } | | | print(json.dumps(output, indent=2, default=str)) | | | else: | |
| print(f"From: {headers.get('From', '?')}") | |
| print(f"To: {headers.get('To', '?')}") | |
| cc = headers.get("Cc", "") | |
| if cc: | |
| print(f"CC: {cc}") | |
| print(f"Date: {headers.get('Date', '?')}") | |
| print(f"Subject: {headers.get('Subject', '?')}") | |
| print("-" * 60) | |
| print(body_plain or body_html or "(no body)") | |
| if attachments: | |
| print("-" * 60) | |
| print(f"Attachments: {', '.join(attachments)}") | |
| def cmd_search(args): | |
| """Search emails with Gmail query syntax.""" | | | service = get_service(args.account) | | | try: | | | result = service.users().messages().list(userId="me", q=args.query, maxResults=args.limit).execute() | | | except Exception as e: | |
| print(f"ERROR: Search failed: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| msg_refs = result.get("messages", []) | |
| if args.json: | | | output = [] | | | for ref in msg_refs: | |
| msg = service.users().messages().get(userId="me", id=ref["id"], format="metadata", | |
| metadataHeaders=["From", "Subject", "Date"]).execute() | |
| headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])} | |
| output.append({ | |
| "id": msg["id"], | |
| "from": headers.get("From", ""), | |
| "subject": headers.get("Subject", ""), | |
| "date": headers.get("Date", ""), | |
| "snippet": msg.get("snippet", ""), | |
| }) | |
| print(json.dumps(output, indent=2, default=str)) | |
| else: | | | print(f"Found {len(msg_refs)} messages matching: {args.query}\n") | | | for ref in msg_refs: | |
| msg = service.users().messages().get(userId="me", id=ref["id"], format="metadata", | |
| metadataHeaders=["From", "Subject", "Date"]).execute() | |
| headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])} | |
| print(f"{headers.get('Date', '?')} | From: {headers.get('From', '?')}") | |
| print(f" Subject: {headers.get('Subject', '?')}") | |
| print(f" ID: {msg['id']}") | |
| print() | |
| def cmd_send(args): | |
| """Send an email.""" | | | import email.mime.text | | | import email.mime.multipart | | | import email.mime.base | | | import mimetypes | |
| service = get_service(args.account) | |
| sender = ACCOUNTS.get(args.account, args.account) | |
| # Read body from stdin if needed | | | body = args.body | |
| if body == "-": | |
| body = sys.stdin.read() | |
| if args.attachments: | | | msg = email.mime.multipart.MIMEMultipart() | | | if args.html: | | | msg.attach(email.mime.text.MIMEText(body, "html")) | | | else: | | | msg.attach(email.mime.text.MIMEText(body, "plain")) | | | for filepath in args.attachments: | | | content_type, _ = mimetypes.guess_type(filepath) | | | if content_type is None: | | | content_type = "application/octet-stream" | |
| main_type, sub_type = content_type.split("/", 1) | |
| with open(filepath, "rb") as f: | |
| attachment = email.mime.base.MIMEBase(main_type, sub_type) | | | attachment.set_payload(f.read()) | | | import email.encoders | | | email.encoders.encode_base64(attachment) | |
| attachment.add_header("Content-Disposition", "attachment", filename=os.path.basename(filepath)) | |
| msg.attach(attachment) | |
| else: | | | if args.html: | | | msg = email.mime.text.MIMEText(body, "html") | | | else: | |
| msg = email.mime.text.MIMEText(body, "plain") | |
| msg["to"] = args.to | |
| msg["from"] = sender | |
| msg["subject"] = args.subject | |
| if args.cc: | | | msg["cc"] = args.cc | | | if args.bcc: | |
| msg["bcc"] = args.bcc | |
| raw = base64.urlsafe_b64encode(msg.as_bytes()).decode() | |
| try: | | | sent = service.users().messages().send(userId="me", body={"raw": raw}).execute() | | | except Exception as e: | |
| print(f"ERROR: Send failed: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| if args.json: | | | print(json.dumps({"status": "sent", "id": sent.get("id", "unknown")}, indent=2)) | | | else: | |
| print(f"Email sent successfully. ID: {sent.get('id', '?')}") | |
| def cmd_reply(args): | |
| """Reply to an email.""" | | | import email.mime.text | | | service = get_service(args.account) | | | try: | |
| original = service.users().messages().get(userId="me", id=args.message_id, format="metadata", | |
| metadataHeaders=["From", "To", "Subject", "Message-ID", "References", "In-Reply-To"]).execute() | |
| except Exception as e: | |
| print(f"ERROR: Could not find message {args.message_id}: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| headers = {h["name"]: h["value"] for h in original.get("payload", {}).get("headers", [])} | |
| thread_id = original["threadId"] | |
| body = args.body | |
| if body == "-": | |
| body = sys.stdin.read() | |
| reply_to = headers.get("From", "") | |
| subject = headers.get("Subject", "") | |
| if not subject.lower().startswith("re:"): | |
| subject = f"Re: {subject}" | |
| message_id = headers.get("Message-ID", "") | |
| references = headers.get("References", "") | |
| if message_id: | |
| references = f"{references} {message_id}".strip() | |
| sender = ACCOUNTS.get(args.account, args.account) | |
| msg = email.mime.text.MIMEText(body) | |
| msg["to"] = reply_to | |
| msg["from"] = sender | |
| msg["subject"] = subject | |
| if message_id: | | | msg["In-Reply-To"] = message_id | | | if references: | |
| msg["References"] = references | |
| raw = base64.urlsafe_b64encode(msg.as_bytes()).decode() | |
| try: | | | sent = service.users().messages().send(userId="me", body={"raw": raw, "threadId": thread_id}).execute() | | | except Exception as e: | |
| print(f"ERROR: Reply failed: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| if args.json: | | | print(json.dumps({"status": "sent", "id": sent.get("id", "unknown")}, indent=2)) | | | else: | |
| print(f"Reply sent successfully. ID: {sent.get('id', '?')}") | |
| def cmd_labels(args): | |
| """List all labels.""" | | | service = get_service(args.account) | | | try: | | | result = service.users().labels().list(userId="me").execute() | | | except Exception as e: | |
| print(f"ERROR: Failed to list labels: {e}", file=sys.stderr) | |
| sys.exit(1) | |
| labels = result.get("labels", []) | |
| if args.json: | |
| output = [{"id": l["id"], "name": l["name"]} for l in labels] | |
| print(json.dumps(output, indent=2)) | |
| else: | |
| for label in sorted(labels, key=lambda l: l["name"]): | |
| print(f"{label['name']} (ID: {label['id']})") | |
| def cmd_mark_read(args): | |
| """Mark message(s) as read.""" | |
| service = get_service(args.account) | |
| for msg_id in args.message_ids: | | | try: | |
| service.users().messages().modify( | |
| userId='me', id=msg_id, | |
| body={'removeLabelIds': ['UNREAD']} | |
| ).execute() | |
| print(f"Marked as read: {msg_id}") | |
| except Exception as e: | |
| print(f"ERROR marking {msg_id} as read: {e}", file=sys.stderr) | |
| def cmd_mark_unread(args): | |
| """Mark message(s) as unread.""" | |
| service = get_service(args.account) | |
| for msg_id in args.message_ids: | | | try: | |
| service.users().messages().modify( | |
| userId='me', id=msg_id, | |
| body={'addLabelIds': ['UNREAD']} | |
| ).execute() | |
| print(f"Marked as unread: {msg_id}") | |
| except Exception as e: | |
| print(f"ERROR marking {msg_id} as unread: {e}", file=sys.stderr) | |
| def cmd_archive(args): | |
| """Archive message(s) by removing from inbox.""" | | | service = get_service(args.account) | | | for msg_id in args.message_ids: | | | try: | |
| service.users().messages().modify( | |
| userId='me', id=msg_id, | |
| body={'removeLabelIds': ['INBOX']} | |
| ).execute() | |
| print(f"Archived: {msg_id}") | |
| except Exception as e: | |
| print(f"ERROR archiving {msg_id}: {e}", file=sys.stderr) | |
| def cmd_star(args): | |
| """Star message(s).""" | |
| service = get_service(args.account) | |
| for msg_id in args.message_ids: | | | try: | |
| service.users().messages().modify( | |
| userId='me', id=msg_id, | |
| body={'addLabelIds': ['STARRED']} | |
| ).execute() | |
| print(f"Starred: {msg_id}") | |
| except Exception as e: | |
| print(f"ERROR starring {msg_id}: {e}", file=sys.stderr) | |
| def cmd_unstar(args): | |
| """Remove star from message(s).""" | | | service = get_service(args.account) | | | for msg_id in args.message_ids: | | | try: | |
| service.users().messages().modify( | |
| userId='me', id=msg_id, | |
| body={'removeLabelIds': ['STARRED']} | |
| ).execute() | |
| print(f"Unstarred: {msg_id}") | |
| except Exception as e: | |
| print(f"ERROR unstarring {msg_id}: {e}", file=sys.stderr) | |
| def cmd_trash(args): | |
| """Move message(s) to trash.""" | |
| service = get_service(args.account) | |
| for msg_id in args.message_ids: | | | try: | |
| service.users().messages().trash(userId='me', id=msg_id).execute() | |
| print(f"Trashed: {msg_id}") | |
| except Exception as e: | |
| print(f"ERROR trashing {msg_id}: {e}", file=sys.stderr) | |
| def cmd_accounts(args): | |
| """List configured accounts.""" | |
| print("Configured account aliases:") | |
| for alias, email in ACCOUNTS.items(): | |
| _, safe_name = _resolve_account(alias) | |
| token_file = CONFIG_DIR / f"token_{safe_name}.json" | |
| status = "authenticated" if token_file.exists() else "NOT SET UP" | |
| print(f" {alias}: {email} [{status}]") | |
| def cmd_setup(args): | |
| """Print setup instructions.""" | | | print(""" | | | Gmail Skill Setup | | | ================= | | | You only need ONE Google Cloud project for all accounts. | | | 1. Go to https://console.cloud.google.com/ | | | 2. Create a project (e.g. "Claude Gmail Access") | | | 3. Enable the Gmail API (APIs & Services > Library) | | | 4. Configure OAuth consent screen: | | | - External type | | | - Add ALL email addresses as test users | | | - Add scope: https://mail.google.com/ | | | 5. Create OAuth credentials: | | | - APIs & Services > Credentials > Create > OAuth client ID > Desktop app | | | - Download the JSON file | | | 6. Copy the SAME JSON file for each account (just rename): | | | """) | |
| for alias, email in ACCOUNTS.items(): | |
| _, safe_name = _resolve_account(alias) | |
| print(f" cp client_secret.json ~/.claude/skills/gmail/config/client_secret_{safe_name}.json") | | | print(""" | |
| 7. Authorize each account (browser opens for OAuth): | |
| python3 gmail_cli.py --account personal list --limit 1 | |
| python3 gmail_cli.py --account work list --limit 1 | |
| (etc.) | |
| 8. IMPORTANT: Publish your OAuth app (APIs & Services > OAuth consent screen | | | > Publish App) or refresh tokens expire after 7 days! | | | """) | |
| def main(): | |
| parser = argparse.ArgumentParser( | |
| description="Gmail CLI for Claude Code", | | | formatter_class=argparse.RawDescriptionHelpFormatter | | | ) | | | parser.add_argument( | | | "--account", "-a", | | | default="personal", | | | help="Account to use: personal, work, business (or full email)" | | | ) | | | parser.add_argument( | | | "--json", "-j", | | | action="store_true", | | | help="Output in JSON format" | | | ) | | | subparsers = parser.add_subparsers(dest="command", help="Commands") | | | # List command | |
| list_parser = subparsers.add_parser("list", help="List recent emails") | |
| list_parser.add_argument("--limit", "-n", type=int, default=10, help="Number of emails") | |
| list_parser.add_argument("--unread", "-u", action="store_true", help="Only unread") | |
| list_parser.add_argument("--starred", "-s", action="store_true", help="Only starred") | |
| list_parser.add_argument("--label", "-l", help="Filter by label") | |
| list_parser.set_defaults(func=cmd_list) | |
| # Read command | | | read_parser = subparsers.add_parser("read", help="Read a specific email") | | | read_parser.add_argument("message_id", help="Message ID") | | | read_parser.set_defaults(func=cmd_read) | | | # Search command | | | search_parser = subparsers.add_parser("search", help="Search emails") | | | search_parser.add_argument("query", help="Gmail search query") | |
| search_parser.add_argument("--limit", "-n", type=int, default=20, help="Max results") | |
| search_parser.set_defaults(func=cmd_search) | |
| # Send command | |
| send_parser = subparsers.add_parser("send", help="Send an email") | |
| send_parser.add_argument("--to", "-t", required=True, help="Recipient") | |
| send_parser.add_argument("--subject", "-s", required=True, help="Subject") | |
| send_parser.add_argument("--body", "-b", default="-", help="Body (use - for stdin)") | |
| send_parser.add_argument("--cc", help="CC recipients") | |
| send_parser.add_argument("--bcc", help="BCC recipients") | |
| send_parser.add_argument("--html", action="store_true", help="Body is HTML") | |
| send_parser.add_argument("--attachments", nargs="*", help="File paths to attach") | |
| send_parser.add_argument("--no-signature", action="store_true", help="Omit signature") | |
| send_parser.set_defaults(func=cmd_send) | |
| # Reply command | | | reply_parser = subparsers.add_parser("reply", help="Reply to an email") | | | reply_parser.add_argument("message_id", help="Message ID to reply to") | |
| reply_parser.add_argument("--body", "-b", default="-", help="Body (use - for stdin)") | |
| reply_parser.set_defaults(func=cmd_reply) | |
| # Labels command | |
| labels_parser = subparsers.add_parser("labels", help="List all labels") | |
| labels_parser.set_defaults(func=cmd_labels) | |
| # Mark read/unread commands | | | mark_read_parser = subparsers.add_parser("mark-read", help="Mark messages as read") | | | mark_read_parser.add_argument("message_ids", nargs="+", help="Message IDs") | |
| mark_read_parser.set_defaults(func=cmd_mark_read) | |
| mark_unread_parser = subparsers.add_parser("mark-unread", help="Mark messages as unread") | |
| mark_unread_parser.add_argument("message_ids", nargs="+", help="Message IDs") | | | mark_unread_parser.set_defaults(func=cmd_mark_unread) | | | # Archive command | |
| archive_parser = subparsers.add_parser("archive", help="Archive messages (remove from inbox)") | |
| archive_parser.add_argument("message_ids", nargs="+", help="Message IDs") | |
| archive_parser.set_defaults(func=cmd_archive) | |
| # Star command | |
| star_parser = subparsers.add_parser("star", help="Star messages") | |
| star_parser.add_argument("message_ids", nargs="+", help="Message IDs") | |
| star_parser.set_defaults(func=cmd_star) | |
| # Unstar command | | | unstar_parser = subparsers.add_parser("unstar", help="Remove star from messages") | |
| unstar_parser.add_argument("message_ids", nargs="+", help="Message IDs") | |
| unstar_parser.set_defaults(func=cmd_unstar) | |
| # Trash command | | | trash_parser = subparsers.add_parser("trash", help="Move messages to trash") | |
| trash_parser.add_argument("message_ids", nargs="+", help="Message IDs") | |
| trash_parser.set_defaults(func=cmd_trash) | |
| # Accounts command | | | accounts_parser = subparsers.add_parser("accounts", help="List configured accounts") | | | accounts_parser.set_defaults(func=cmd_accounts) | | | # Setup command | | | setup_parser = subparsers.add_parser("setup", help="Show setup instructions") | |
| setup_parser.set_defaults(func=cmd_setup) | |
| args = parser.parse_args() | |
| if not args.command: | |
| parser.print_help() | |
| sys.exit(1) | |
| args.func(args) | |
| if __name__ == "__main__": | |
| main() |