# NousResearch/hermes-agent · Google Workspace skill expansion and CLI migration
## Open now
- 4 issues, 0 PRs open on GitHub.
- #9189: \[Feature\]: Add Google Tasks tool to Google Workspace integration
- #15600: Feature: Add Google Slides support to google-workspace skill
- #17272: \[Feature\]: Service-account auth + Docs/Drive create endpoints in google-workspace skill (for autonomous deployments)
- #22872: productivity/google-workspace: CLI wrapper missing attachment download
These issues and pull requests concern migrating the Google Workspace skill to a CLI backend (gws) and adding new capabilities including Gmail attachments, Google Tasks, Google Slides, Docs/Drive creation, and service-account authentication.
## TL;DR
- 8 member(s): 4 open issue(s), 1 closed issue(s), 0 open PR(s), 1 merged PR(s), 2 PR(s) closed without merging.
- A fix was merged to main in PR #6788. Verify it addresses your case.
- **Recommended:** read PR #6788 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.
## Maintainer resolution verdict
- Internal-tracker verdict over this cluster's issue members: `unresolved`. This is distinct from the fix status above. Weigh it yourself.
## How this cluster was grouped
- Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text.
- Members outside that structural core may belong to a different underlying problem.
- Pairwise link density `0.16`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.
## Open related issues
- #9189: \[Feature\]: Add Google Tasks tool to Google Workspace integration
- #15600: Feature: Add Google Slides support to google-workspace skill
- #17272: \[Feature\]: Service-account auth + Docs/Drive create endpoints in google-workspace skill (for autonomous deployments)
- #22872: productivity/google-workspace: CLI wrapper missing attachment download
## Closed / duplicate issues
- #411: Feature: Migrate Google Workspace Skill to Official Google Workspace CLI (gws)
## Prior & competing PRs
- **#6788: feat(skills): migrate google-workspace to gws CLI backend**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `skills/productivity/google-workspace/SKILL.md`, `skills/productivity/google-workspace/scripts/google_api.py`, `skills/productivity/google-workspace/scripts/gws_bridge.py`, `skills/productivity/google-workspace/scripts/setup.py`, `tests/skills/test_google_oauth_setup.py`, `tests/skills/test_google_workspace_api.py`
- fix hunks:
- `skills/productivity/google-workspace/SKILL.md`
```diff
@@ -1,7 +1,7 @@
---
name: google-workspace
-description: Gmail, Calendar, Drive, Contacts, Sheets, and Docs integration via Python. Uses OAuth2 with automatic token refresh. No external binaries needed — runs entirely with Google's Python client libraries in the Hermes venv.
-version: 1.0.0
+description: Gmail, Calendar, Drive, Contacts, Sheets, and Docs integration via gws CLI (googleworkspace/cli). Uses OAuth2 with automatic token refresh via bridge script. Requires gws binary.
+version: 2.0.0
author: Nous Research
license: MIT
required_credential_files:
```
```diff
@@ -11,14 +11,25 @@ required_credential_files:
[redacted:secret-assignment] Google OAuth2 client credentials (downloaded from Google Cloud Console)
metadata:
hermes:
- tags: [Google, Gmail, Calendar, Drive, Sheets, Docs, Contacts, Email, OAuth]
+ tags: [Google, Gmail, Calendar, Drive, Sheets, Docs, Contacts, Email, OAuth, gws]
homepage: https://github.com/NousResearch/hermes-agent
related_skills: [himalaya]
---
# Google Workspace
-Gmail, Calendar, Drive, Contacts, Sheets, and Docs — all through Python scripts in this skill. No external binaries to install.
+Gmail, Calendar, Drive, Contacts, Sheets, and Docs — powered by `gws` (Google's official Rust CLI). The skill provides a backward-compatible Python wrapper that handles OAuth token refresh and delegates to `gws`.
+
+## Architecture
+
+```
+google_api.py → gws_bridge.py → gws CLI
+(argparse compat) (token refresh) (Google APIs)
+```
+
+- `setup.py` handles OAuth2 (headless-compatible, works on CLI/Telegram/Discord)
+- `gws_bridge.py` refreshes the Hermes token and injects it into `gws` via `GOOGLE_WORKSPACE_CLI_TOKEN`
+- `google_api.py` provides the same CLI interface as v1 but delegates to `gws`
## References
```
```diff
@@ -27,7 +38,22 @@ Gmail, Calendar, Drive, Contacts, Sheets, and Docs — all through Python script
## Scripts
- `scripts/setup.py` — OAuth2 setup (run once to authorize)
-- `scripts/google_api.py` — API wrapper CLI (agent uses this for all operations)
+- `scripts/gws_bridge.py` — Token refresh bridge to gws CLI
+- `scripts/google_api.py` — Backward-compatible API wrapper (delegates to gws)
+
+## Prerequisites
+
+Install `gws`:
+
+```bash
+cargo install google-workspace-cli
+# or via npm (recommended, downloads prebuilt binary):
+npm install -g @googleworkspace/cli
+# or via Homebrew:
+brew install googleworkspace-cli
+```
+
+Verify: `gws --version`
## First-Time Setup
```
```diff
@@ -56,42 +82,29 @@ If it prints `AUTHENTICATED`, skip to Usage — setup is already done.
### Step 1: Triage — ask the user what they need
-Before starting OAuth setup, ask the user TWO questions:
-
**Question 1: "What Google services do you need? Just email, or also
Calendar/Drive/Sheets/Docs?"**
-- **Email only** → They don't need this skill at all. Use the `himalaya` skill
- instead — it works with a Gmail App Password (Settings → Security → App
- Passwords) and takes 2 minutes to set up. No Google Cloud project needed.
- Load the himalaya skill and follow its setup instructions.
+- **Email only** → Use the `himalaya` skill instead — simpler setup.
+- **Calendar, Drive, Sheets, Docs (or email + these)** → Continue below.
-- **Calendar, Drive, Sheets, Docs (or email + these)** → Continue with this
- skill's OAuth setup below.
+**Partial scopes**: Users can authorize only a subset of services. The setup
+script accepts partial scopes and warns about missing ones.
-**Question 2: "Does your Google account use Advanced Protection (hardware
-security keys required to sign in)? If you're not sure, you probably don't
-— it's something you would have explicitly enrolled in."**
+**Question 2: "Does your Google account use Advanced Protection?"**
-- **No / Not sure** → Normal setup. Continue below.
-- **Yes** → Their Workspace admin must add the OAuth client ID to the org's
- allowed apps list before Step 4 will work. Let them know upfront.
+- **No / Not sure** → Normal setup.
+- **Yes** → Workspace admin must add the OAuth client ID to allowed apps first.
### Step 2: Create OAuth credentials (one-time, ~5 minutes)
Tell the user:
-> You need a Google Cloud OAuth client. This is a one-time setup:
->
> 1. Go to https://console.cloud.google.com/apis/credentials
> 2. Create a project (or use an existing one)
-> 3. Click "Enable APIs" and enable: Gmail API, Google Calendar API,
-> Google Drive API, Google Sheets API, Google Docs API, People API
-> 4. Go to Credentials → Create Credentials → OAuth 2.0 Client ID
-> 5. Application type: "Desktop app" → Create
-> 6. Click "Download JSON" and tell me the file path
-
-Once they provide the path:
+> 3. Enable the APIs you need (Gmail, Calendar, Drive, Sheets, Docs, People)
+> 4. Credentials → Create Credentials → OAuth 2.0 Client ID → Desktop app
+> 5. Download JSON and tell me the file path
```bash
$GSETUP --client-secret /path/to/client_secret.json
```
```diff
@@ -103,20 +116,10 @@ $GSETUP --client-secret /path/to/client_secret.json
$GSETUP --auth-url
```
-This prints a URL. **Send the URL to the user** and tell them:
-
-> Open this link in your browser, sign in with your Google account, and
-> authorize access. After authorizing, you'll be redirected to a page that
-> may show an error — that's expected. Copy the ENTIRE URL from your
-> browser's address bar and paste it back to me.
+Send the URL to the user. After authorizing, they paste back the redirect URL or code.
### Step 4: Exchange the code
-The user will paste back either a URL like `http://localhost:1/?code=4/0A...&scope=...`
-or just the code string. Either works. The `--auth-url` step stores a temporary
-pending OAuth session locally so `--auth-code` can complete the PKCE exchange
-later, even on headless systems:
-
```bash
$GSETUP --auth-code "THE_URL_OR_CODE_THE_USER_PASTED"
```
```
```diff
@@ -127,18 +130,11 @@ $GSETUP --auth-code "THE_URL_OR_CODE_THE_USER_PASTED"
$GSETUP --check
```
-Should print `AUTHENTICATED`. Setup is complete — token refreshes automatically from now on.
-
-### Notes
-
-- Token is stored at `google_token.json` under the active profile's `HERMES_HOME` and auto-refreshes.
-- Pending OAuth session state/verifier are stored temporarily at `google_oauth_pending.json` under the active profile's `HERMES_HOME` until exchange completes.
-- Hermes now refuses to overwrite a full Google Workspace token with a narrower re-auth token missing Gmail scopes, so one profile's partial consent cannot silently break email actions later.
-- To revoke: `$GSETUP --revoke`
+Should print `AUTHENTICATED`. Token refreshes automatically from now on.
## Usage
-All commands go through the API script. Set `GAPI` as a shorthand:
+All commands go through the API script:
```bash
HERMES_HOME="${HERMES_HOME:-$HOME/.hermes}"
```
```diff
@@ -153,40 +149,21 @@ GAPI="$PYTHON_BIN $GWORKSPACE_SKILL_DIR/scripts/google_api.py"
### Gmail
```bash
-# Search (returns JSON array with id, from, subject, date, snippet)
$GAPI gmail search "is:unread" --max 10
-$GAPI gmail search "from:[redacted:email] newer_than:1d"
-$GAPI gmail search "has:attachment filename:pdf newer_than:7d"
-
-# Read full message (returns JSON with body text)
$GAPI gmail get MESSAGE_ID
-
-# Send
$GAPI gmail send --to [redacted:email] --subject "Hello" --body "Message text"
-$GAPI gmail send --to [redacted:email] --subject "Report" --body "Q4
Details...
" --html
-
-# Reply (automatically threads and sets In-Reply-To)
+$GAPI gmail send --to [redacted:email] --subject "Report" --body "Q4
" --html
$GAPI gmail reply MESSAGE_ID --body "Thanks, that works for me."
-
-# Labels
$GAPI gmail labels
$GAPI gmail modify MESSAGE_ID --add-labels LABEL_ID
-$GAPI gmail modify MESSAGE_ID --remove-labels UNREAD
```
### Calendar
```bash
-# List events (defaults to next 7 days)
$GAPI calendar list
-$GAPI calendar list --start 2026-03-01T00:00:00Z --end 2026-03-07T23:59:59Z
-
-# Create event (ISO 8601 with timezone required)
-$GAPI calendar create --summary "Team Standup" --start 2026-03-01T10:00:00-06:00 --end 2026-03-01T10:30:00-06:00
-$GAPI calendar create --summary "Lunch" --start 2026-03-01T12:00:00Z --end 2026-03-01T13:00:00Z --location "Cafe"
-$GAPI calendar create --summary "Review" --start 2026-03-01T14:00:00Z --end 2026-03-01T15:00:00Z --attendees "[redacted:email],[redacted:email]"
-
-# Delete event
+$GAPI calendar create --summary "Standup" --start 2026-03-01T10:00:00+01:00 --end 2026-03-01T10:30:00+01:00
+$GAPI calendar create --summary "Review" --start ... --end ... --attendees "[redacted:email],[redacted:email]"
$GAPI calendar delete EVENT_ID
```
```
```diff
@@ -206,13 +183,8 @@ $GAPI contacts list --max 20
### Sheets
```bash
-# Read
$GAPI sheets get SHEET_ID "Sheet1!A1:D10"
-
-# Write
$GAPI sheets update SHEET_ID "Sheet1!A1:B2" --values '[["Name","Score"],["Alice","95"]]'
-
-# Append rows
$GAPI sheets append SHEET_ID "Sheet1!A:C" --values '[["new","row","data"]]'
```
```
```diff
@@ -222,37 +194,52 @@ $GAPI sheets append SHEET_ID "Sheet1!A:C" --values '[["new","row","data"]]'
$GAPI docs get DOC_ID
```
+### Direct gws access (advanced)
+
+For operations not covered by the wrapper, use `gws_bridge.py` directly:
+
+```bash
+GBRIDGE="$PYTHON_BIN $GWORKSPACE_SKILL_DIR/scripts/gws_bridge.py"
+$GBRIDGE calendar +agenda --today --format table
+$GBRIDGE gmail +triage --labels --format json
+$GBRIDGE drive +upload ./report.pdf
+$GBRIDGE sheets +read --spreadsheet SHEET_ID --range "Sheet1!A1:D10"
+```
+
## Output Format
-All commands return JSON. Parse with `jq` or read directly. Key fields:
+All commands return JSON via `gws --format json`. Key output shapes:
+
+- **Gmail search/triage**: Array of message summaries (sender, subject, date, snippet)
+- **Gmail get/read**: Message object with headers and body text
+- **Gmail send/reply**: Confirmation with message ID
+- **Calendar list/agenda**: Array of event objects (summary, start, end, location)
+- **Calendar create**: Confirmation with event ID and htmlLink
+- **Drive search**: Array of file objects (id, name, mimeType, webViewLink)
+- **Sheets get/read**: 2D array of cell values
+- **Docs get**: Full document JSON (use `body.content` for text extraction)
+- **Contacts list**: Array of person objects with names, emails, phones
-- **Gmail search**: `[{id, threadId, from, to, subject, date, snippet, labels}]`
-- **Gmail get**: `{id, threadId, from, to, subject, date, labels, body}`
-- **Gmail send/reply**: `{status: "sent", id, threadId}`
-- **Calendar list**: `[{id, summary, start, end, location, description, htmlLink}]`
-- **Calendar create**: `{status: "created", id, summary, htmlLink}`
-- **Drive search**: `[{id, name, mimeType, modifiedTime, webViewLink}]`
-- **Contacts list**: `[{name, emails: [...], phones: [...]}]`
-- **Sheets get**: `[[cell, cell, ...], ...]`
+Parse output with `jq` or read JSON directly.
## Rules
-1. **Never send email or create/delete events without confirming with the user first.** Show the draft content and ask for approval.
-2. **Check auth before first use** — run `setup.py --check`. If it fails, guide the user through setup.
-3. **Use the Gmail search syntax reference** for complex queries — load it with `skill_view("google-workspace", file_path="references/gmail-search-syntax.md")`.
-4. **Calendar times must include timezone** — always use ISO 8601 with offset (e.g., `2026-03-01T10:00:00-06:00`) or UTC (`Z`).
-5. **Respect rate limits** — avoid rapid-fire sequential API calls. Batch reads when possible.
+1. **Never send email or create/delete events without confirming with the user first.**
+2. **Check auth before first use** — run `setup.py --check`.
+3. **Use the Gmail search syntax reference** for complex queries.
+4. **Calendar times must include timezone** — ISO 8601 with offset or UTC.
+5. **Respect rate limits** — avoid rapid-fire sequential API calls.
## Troubleshooting
| Problem | Fix |
|---------|-----|
-| `NOT_AUTHENTICATED` | Run setup Steps 2-5 above |
-| `REFRESH_FAILED` | Token revoked or expired — redo Steps 3-5 |
-| `HttpError 403: Insufficient Permission` | Missing API scope — `$GSETUP --revoke` then redo Steps 3-5 |
-| `HttpError 403: Access Not Configured` | API not enabled — user needs to enable it in Google Cloud Console |
-| `ModuleNotFoundError` | Run `$GSETUP --install-deps` |
-| Advanced Protection blocks auth | Workspace admin must allowlist the OAuth client ID |
+| `NOT_AUTHENTICATED` | Run setup Steps 2-5 |
+| `REFRESH_FAILED` | Token revoked — redo Steps 3-5 |
+| `gws: command not found` | Install: `npm install -g @googleworkspace/cli` |
+| `HttpError 403` | Missing scope — `$GSETUP --revoke` then redo Steps 3-5 |
+| `HttpError 403: Access Not Configured` | Enable API in Google Cloud Console |
+| Advanced Protection blocks auth | Admin must allowlist the OAuth client ID |
## Revoking Access
```
- `skills/productivity/google-workspace/scripts/google_api.py`
```diff
@@ -1,16 +1,17 @@
#!/usr/bin/env python3
"""Google Workspace API CLI for Hermes Agent.
-A thin CLI wrapper around Google's Python client libraries.
-Authenticates using the token stored by setup.py.
+Thin wrapper that delegates to gws (googleworkspace/cli) via gws_bridge.py.
+Maintains the same CLI interface for backward compatibility with Hermes skills.
Usage:
python google_api.py gmail search "is:unread" [--max 10]
python google_api.py gmail get MESSAGE_ID
python google_api.py gmail send --to [redacted:email] --subject "Hi" --body "Hello"
python google_api.py gmail reply MESSAGE_ID --body "Thanks"
- python google_api.py calendar list [--from DATE] [--to DATE] [--calendar primary]
+ python google_api.py calendar list [--start DATE] [--end DATE] [--calendar primary]
python google_api.py calendar create --summary "Meeting" --start DATETIME --end DATETIME
+ python google_api.py calendar delete EVENT_ID
python google_api.py drive search "budget report" [--max 10]
python google_api.py contacts list [--max 20]
python google_api.py sheets get SHEET_ID RANGE
```
```diff
@@ -20,386 +21,193 @@
"""
import argparse
-import base64
import json
+import os
+import subprocess
import sys
-from datetime import datetime, timedelta, timezone
-from email.mime.text import MIMEText
from pathlib import Path
-try:
- from hermes_constants import display_hermes_home, get_hermes_home
-except ModuleNotFoundError:
- HERMES_AGENT_ROOT = Path(__file__).resolve().parents[4]
- if HERMES_AGENT_ROOT.exists():
- sys.path.insert(0, str(HERMES_AGENT_ROOT))
- from hermes_constants import display_hermes_home, get_hermes_home
-
-HERMES_HOME = get_hermes_home()
-TOKEN_PATH = [redacted:secret-assignment] / "google_token.json"
-
-SCOPES = [
- "https://www.googleapis.com/auth/gmail.readonly",
- "https://www.googleapis.com/auth/gmail.send",
- "https://www.googleapis.com/auth/gmail.modify",
- "https://www.googleapis.com/auth/calendar",
- "https://www.googleapis.com/auth/drive.readonly",
- "https://www.googleapis.com/auth/contacts.readonly",
- "https://www.googleapis.com/auth/spreadsheets",
- "https://www.googleapis.com/auth/documents.readonly",
-]
-
-
-def _missing_scopes() -> list[str]:
- try:
- payload = json.loads(TOKEN_PATH.read_text())
- except Exception:
- return []
- raw = payload.get("scopes") or payload.get("scope")
- if not raw:
- return []
- granted = {s.strip() for s in (raw.split() if isinstance(raw, str) else raw) if s.strip()}
- return sorted(scope for scope in SCOPES if scope not in granted)
-
-
-def get_credentials():
- """Load and refresh credentials from token file."""
- if not TOKEN_PATH.exists():
- print("Not authenticated. Run the setup script first:", file=sys.stderr)
- print(f" python {Path(__file__).parent / 'setup.py'}", file=sys.stderr)
- sys.exit(1)
-
- from google.oauth2.credentials import Credentials
- from google.auth.transport.requests import Request
-
- creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
- if creds.expired and creds.refresh_token:
- creds.refresh(Request())
- TOKEN_PATH.write_text(creds.to_json())
- if not creds.valid:
- print("Token is invalid. Re-run setup.", file=sys.stderr)
- sys.exit(1)
-
- missing_scopes = _missing_scopes()
- if missing_scopes:
- print(
- "Token is valid but missing Google Workspace scopes required by this skill.",
- file=sys.stderr,
- )
- for scope in missing_scopes:
- print(f" - {scope}", file=sys.stderr)
- print(
- f"Re-run setup.py from the active Hermes profile ({display_hermes_home()}) to restore full access.",
- file=sys.stderr,
- )
- sys.exit(1)
- return creds
+BRIDGE = Path(__file__).parent / "gws_bridge.py"
+PYTHON = sys.executable
-def build_service(api, version):
- from googleapiclient.discovery import build
- return build(api, version, credentials=[redacted:secret-assignment]
+def gws(*args: str) -> None:
+ """Call gws via the bridge and exit with its return code."""
+ result = subprocess.run(
+ [PYTHON, str(BRIDGE)] + list(args),
+ env={**os.environ, "HERMES_HOME": os.environ.get("HERMES_HOME", str(Path.home() / ".hermes"))},
+ )
+ sys.exit(result.returncode)
-# =========================================================================
-# Gmail
-# =========================================================================
+# -- Gmail --
def gmail_search(args):
- service = build_service("gmail", "v1")
- results = service.users().messages().list(
- userId="me", q=args.query, maxResults=args.max
- ).execute()
- messages = results.get("messages", [])
- if not messages:
- print("No messages found.")
- return
-
- output = []
- for msg_meta in messages:
- msg = service.users().messages().get(
- userId="me", id=msg_meta["id"], format="metadata",
- metadataHeaders=["From", "To", "Subject", "Date"],
- ).execute()
- headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
- output.append({
- "id": msg["id"],
- "threadId": msg["threadId"],
- "from": headers.get("From", ""),
- "to": headers.get("To", ""),
- "subject": headers.get("Subject", ""),
- "date": headers.get("Date", ""),
- "snippet": msg.get("snippet", ""),
- "labels": msg.get("labelIds", []),
- })
- print(json.dumps(output, indent=2, ensure_ascii=False))
-
+ cmd = ["gmail", "+triage", "--query", args.query, "--max", str(args.max), "--format", "json"]
+ gws(*cmd)
def gmail_get(args):
- service = build_service("gmail", "v1")
- msg = service.users().messages().get(
- userId="me", id=args.message_id, format="full"
- ).execute()
-
- headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
-
- # Extract body text
- body = ""
- payload = msg.get("payload", {})
- if payload.get("body", {}).get("data"):
- body = base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", errors="replace")
- elif payload.get("parts"):
- for part in payload["parts"]:
- if part.get("mimeType") == "text/plain" and part.get("body", {}).get("data"):
- body = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace")
- break
- if not body:
- for part in payload["parts"]:
- if part.get("mimeType") == "text/html" and part.get("body", {}).get("data"):
- body = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace")
- break
-
- result = {
- "id": msg["id"],
- "threadId": msg["threadId"],
- "from": headers.get("From", ""),
- "to": headers.get("To", ""),
- "subject": headers.get("Subject", ""),
- "date": headers.get("Date", ""),
- "labels": msg.get("labelIds", []),
- "body": body,
- }
- print(json.dumps(result, indent=2, ensure_ascii=False))
-
+ gws("gmail", "+read", "--id", args.message_id, "--headers", "--format", "json")
def gmail_send(args):
- service = build_service("gmail", "v1")
- message = MIMEText(args.body, "html" if args.html else "plain")
- message["to"] = args.to
- message["subject"] = args.subject
+ cmd = ["gmail", "+send", "--to", args.to, "--subject", args.subject, "--body", args.body, "--format", "json"]
if args.cc:
- message["cc"] = args.cc
-
- raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
- body = {"raw": raw}
-
- if args.thread_id:
- body["threadId"] = args.thread_id
-
- result = service.users().messages().send(userId="me", body=body).execute()
- print(json.dumps({"status": "sent", "id": result["id"], "threadId": result.get("threadId", "")}, indent=2))
-
+ cmd += ["--cc", args.cc]
+ if args.html:
+ cmd.append("--html")
+ gws(*cmd)
def gmail_reply(args):
- service = build_service("gmail", "v1")
- # Fetch original to get thread ID and headers
- original = service.users().messages().get(
- userId="me", id=args.message_id, format="metadata",
- metadataHeaders=["From", "Subject", "Message-ID"],
- ).execute()
- headers = {h["name"]: h["value"] for h in original.get("payload", {}).get("headers", [])}
-
- subject = headers.get("Subject", "")
- if not subject.startswith("Re:"):
- subject = f"Re: {subject}"
-
- message = MIMEText(args.body)
- message["to"] = headers.get("From", "")
- message["subject"] = subject
- if headers.get("Message-ID"):
- message["In-Reply-To"] = headers["Message-ID"]
- message["References"] = headers["Message-ID"]
-
- raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
- body = {"raw": raw, "threadId": original["threadId"]}
-
- result = service.users().messages().send(userId="me", body=body).execute()
- print(json.dumps({"status": "sent", "id": result["id"], "threadId": result.get("threadId", "")}, indent=2))
-
+ gws("gmail", "+reply", "--message-id", args.message_id, "--body", args.body, "--format", "json")
def gmail_labels(args):
- service = build_service("gmail", "v1")
- results = service.users().labels().list(userId="me").execute()
- labels = [{"id": l["id"], "name": l["name"], "type": l.get("type", "")} for l in results.get("labels", [])]
- print(json.dumps(labels, indent=2))
-
+ gws("gmail", "users", "labels", "list", "--params", json.dumps({"userId": "me"}), "--format", "json")
def gmail_modify(args):
- service = build_service("gmail", "v1")
body = {}
if args.add_labels:
body["addLabelIds"] = args.add_labels.split(",")
if args.remove_labels:
body["removeLabelIds"] = args.remove_labels.split(",")
- result = service.users().messages().modify(userId="me", id=args.message_id, body=body).execute()
- print(json.dumps({"id": result["id"], "labels": result.get("labelIds", [])}, indent=2))
+ gws(
+ "gmail", "users", "messages", "modify",
+ "--params", json.dumps({"userId": "me", "id": args.message_id}),
+ "--json", json.dumps(body),
+ "--format", "json",
+ )
-# =========================================================================
-# Calendar
-# =========================================================================
+# -- Calendar --
def calendar_list(args):
- service = build_service("calendar", "v3")
- now = datetime.now(timezone.utc)
- time_min = args.start or now.isoformat()
- time_max = args.end or (now + timedelta(days=7)).isoformat()
-
- # Ensure timezone info
- for val in [time_min, time_max]:
- if "T" in val and "Z" not in val and "+" not in val and "-" not in val[11:]:
- val += "Z"
-
- results = service.events().list(
- calendarId=args.calendar, timeMin=time_min, timeMax=time_max,
- maxResults=args.max, singleEvents=True, orderBy="startTime",
- ).execute()
-
- events = []
- for e in results.get("items", []):
- events.append({
- "id": e["id"],
- "summary": e.get("summary", "(no title)"),
- "start": e.get("start", {}).get("dateTime", e.get("start", {}).get("date", "")),
- "end": e.get("end", {}).get("dateTime", e.get("end", {}).get("date", "")),
- "location": e.get("location", ""),
- "description": e.get("description", ""),
- "status": e.get("status", ""),
- "htmlLink": e.get("htmlLink", ""),
- })
- print(json.dumps(events, indent=2, ensure_ascii=False))
-
+ if args.start or args.end:
+ # Specific date range — use raw Calendar API for precise timeMin/timeMax
+ from datetime import datetime, timedelta, timezone as tz
+ now = datetime.now(tz.utc)
+ time_min = args.start or now.isoformat()
+ time_max = args.end or (now + timedelta(days=7)).isoformat()
+ gws(
+ "calendar", "events", "list",
+ "--params", json.dumps({
+ "calendarId": args.calendar,
+ "timeMin": time_min,
+ "timeMax": time_max,
+ "maxResults": args.max,
+ "singleEvents": True,
+ "orderBy": "startTime",
+ }),
+ "--format", "json",
+ )
+ else:
+ # No date range — use +agenda helper (defaults to 7 days)
+ cmd = ["calendar", "+agenda", "--days", "7", "--format", "json"]
+ if args.calendar != "primary":
+ cmd += ["--calendar", args.calendar]
+ gws(*cmd)
def calendar_create(args):
- service = build_service("calendar", "v3")
- event = {
- "summary": args.summary,
- "start": {"dateTime": args.start},
- "end": {"dateTime": args.end},
- }
+ cmd = [
+ "calendar", "+insert",
+ "--summary", args.summary,
+ "--start", args.start,
+ "--end", args.end,
+ "--format", "json",
+ ]
if args.location:
- event["location"] = args.location
+ cmd += ["--location", args.location]
if args.description:
- event["description"] = args.description
+ cmd += ["--description", args.description]
if args.attendees:
- event["attendees"] = [{"email": e.strip()} for e in args.attendees.split(",")]
-
- result = service.events().insert(calendarId=args.calendar, body=event).execute()
- print(json.dumps({
- "status": "created",
- "id": result["id"],
- "summary": result.get("summary", ""),
- "htmlLink": result.get("htmlLink", ""),
- }, indent=2))
-
+ for email in args.attendees.split(","):
+ cmd += ["--attendee", email.strip()]
+ if args.calendar != "primary":
+ cmd += ["--calendar", args.calendar]
+ gws(*cmd)
def calendar_delete(args):
- service = build_service("calendar", "v3")
- service.events().delete(calendarId=args.calendar, eventId=args.event_id).execute()
- print(json.dumps({"status": "deleted", "eventId": args.event_id}))
+ gws(
+ "calendar", "events", "delete",
+ "--params", json.dumps({"calendarId": args.calendar, "eventId": args.event_id}),
+ "--format", "json",
+ )
-# =========================================================================
-# Drive
-# =========================================================================
+# -- Drive --
def drive_search(args):
- service = build_service("drive", "v3")
- query = f"fullText contains '{args.query}'" if not args.raw_query else args.query
- results = service.files().list(
- q=query, pageSize=args.max, fields="files(id, name, mimeType, modifiedTime, webViewLink)",
- ).execute()
- files = results.get("files", [])
- print(json.dumps(files, indent=2, ensure_ascii=False))
+ query = args.query if args.raw_query else f"fullText contains '{args.query}'"
+ gws(
+ "drive", "files", "list",
+ "--params", json.dumps({
+ "q": query,
+ "pageSize": args.max,
+ "fields": "files(id,name,mimeType,modifiedTime,webViewLink)",
+ }),
+ "--format", "json",
+ )
-# =========================================================================
-# Contacts
-# =========================================================================
+# -- Contacts --
def contacts_list(args):
- service = build_service("people", "v1")
- results = service.people().connections().list(
- resourceName="people/me",
- pageSize=args.max,
- personFields="names,emailAddresses,phoneNumbers",
- ).execute()
- contacts = []
- for person in results.get("connections", []):
- names = person.get("names", [{}])
- emails = person.get("emailAddresses", [])
- phones = person.get("phoneNumbers", [])
- contacts.append({
- "name": names[0].get("displayName", "") if names else "",
- "emails": [e.get("value", "") for e in emails],
- "phones": [p.get("value", "") for p in phones],
- })
- print(json.dumps(contacts, indent=2, ensure_ascii=False))
-
-
-# =========================================================================
-# Sheets
-# =========================================================================
+ gws(
+ "people", "people", "connections", "list",
+ "--params", json.dumps({
+ "resourceName": "people/me",
+ "pageSize": args.max,
+ "personFields": "names,emailAddresses,phoneNumbers",
+ }),
+ "--format", "json",
+ )
-def sheets_get(args):
- service = build_service("sheets", "v4")
- result = service.spreadsheets().values().get(
- spreadsheetId=args.sheet_id, range=args.range,
- ).execute()
- print(json.dumps(result.get("values", []), indent=2, ensure_ascii=False))
+# -- Sheets --
+
+def sheets_get(args):
+ gws(
+ "sheets", "+read",
+ "--spreadsheet", args.sheet_id,
+ "--range", args.range,
+ "--format", "json",
+ )
def sheets_update(args):
- service = build_service("sheets", "v4")
values = json.loads(args.values)
- body = {"values": values}
- result = service.spreadsheets().values().update(
- spreadsheetId=args.sheet_id, range=args.range,
- valueInputOption="USER_ENTERED", body=body,
- ).execute()
- print(json.dumps({"updatedCells": result.get("updatedCells", 0), "updatedRange": result.get("updatedRange", "")}, indent=2))
-
+ gws(
+ "sheets", "spreadsheets", "values", "update",
+ "--params", json.dumps({
+ "spreadsheetId": args.sheet_id,
+ "range": args.range,
+ "valueInputOption": "USER_ENTERED",
+ }),
+ "--json", json.dumps({"values": values}),
+ "--format", "json",
+ )
def sheets_append(args):
- service = build_service("sheets", "v4")
values = json.loads(args.values)
- body = {"values": values}
- result = service.spreadsheets().values().append(
- spreadsheetId=args.sheet_id, range=args.range,
- valueInputOption="USER_ENTERED", insertDataOption="INSERT_ROWS", body=body,
- ).execute()
- print(json.dumps({"updatedCells": result.get("updates", {}).get("updatedCells", 0)}, indent=2))
+ gws(
+ "sheets", "+append",
+ "--spreadsheet", args.sheet_id,
+ "--json-values", json.dumps(values),
+ "--format", "json",
+ )
-# =========================================================================
-# Docs
-# =========================================================================
+# -- Docs --
def docs_get(args):
- service = build_service("docs", "v1")
- doc = service.documents().get(documentId=args.doc_id).execute()
- # Extract plain text from the document structure
- text_parts = []
- for element in doc.get("body", {}).get("content", []):
- paragraph = element.get("paragraph", {})
- for pe in paragraph.get("elements", []):
- text_run = pe.get("textRun", {})
- if text_run.get("content"):
- text_parts.append(text_run["content"])
- result = {
- "title": doc.get("title", ""),
- "documentId": doc.get("documentId", ""),
- "body": "".join(text_parts),
- }
- print(json.dumps(result, indent=2, ensure_ascii=False))
-
-
-# =========================================================================
-# CLI parser
-# =========================================================================
+ gws(
+ "docs", "documents", "get",
+ "--params", json.dumps({"documentId": args.doc_id}),
+ "--format", "json",
+ )
+
+
+# -- CLI parser (backward-compatible interface) --
def main():
- parser = argparse.ArgumentParser(description="Google Workspace API for Hermes Agent")
+ parser = argparse.ArgumentParser(description="Google Workspace API for Hermes Agent (gws backend)")
sub = parser.add_subparsers(dest="service", required=True)
# --- Gmail ---
```
```diff
@@ -421,7 +229,7 @@ def main():
p.add_argument("--body", required=True)
p.add_argument("--cc", default="")
p.add_argument("--html", action="store_true", help="Send body as HTML")
- p.add_argument("--thread-id", default="", help="Thread ID for threading")
+ p.add_argument("--thread-id", default="", help="Thread ID (unused with gws, kept for compat)")
p.set_defaults(func=gmail_send)
p = gmail_sub.add_parser("reply")
```
- `skills/productivity/google-workspace/scripts/gws_bridge.py`
```diff
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Bridge between Hermes OAuth token and gws CLI.
+
+Refreshes the token if expired, then executes gws with the valid access token.
+"""
+import json
+import os
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def get_hermes_home() -> Path:
+ return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
+
+
+def get_token_path() -> Path:
+ return get_hermes_home() / "google_token.json"
+
+
+def refresh_token(token_data: dict) -> dict:
+ """Refresh the access token using the refresh token."""
+ import urllib.error
+ import urllib.parse
+ import urllib.request
+
+ params = urllib.parse.urlencode({
+ "client_id": token_data["client_id"],
+ "client_secret": token_data["client_secret"],
+ "refresh_token": token_data["refresh_token"],
+ "grant_type": "refresh_token",
+ }).encode()
+
+ req = urllib.request.Request(token_data["token_uri"], data=params)
+ try:
+ with urllib.request.urlopen(req) as resp:
+ result = json.loads(resp.read())
+ except urllib.error.HTTPError as e:
+ body = e.read().decode("utf-8", errors="replace")
+ print(f"ERROR: Token refresh failed (HTTP {e.code}): {body}", file=sys.stderr)
+ print("Re-run setup.py to re-authenticate.", file=sys.stderr)
+ sys.exit(1)
+
+ token_data["token"] = result["access_token"]
+ token_data["expiry"] = datetime.fromtimestamp(
+ datetime.now(timezone.utc).timestamp() + result["expires_in"],
+ tz=timezone.utc,
+ ).isoformat()
+
+ get_token_path().write_text(json.dumps(token_data, indent=2))
+ return token_data
+
+
+def get_valid_token() -> str:
+ """Return a valid access token, refreshing if needed."""
+ token_path = [redacted:secret-assignment]
+ if not token_path.exists():
+ print("ERROR: No Google token found. Run setup.py --auth-url first.", file=sys.stderr)
+ sys.exit(1)
+
+ token_data = [redacted:secret-assignment]
+
+ expiry = token_data.get("expiry", "")
+ if expiry:
+ exp_dt = datetime.fromisoformat(expiry.replace("Z", "+00:00"))
+ now = datetime.now(timezone.utc)
+ if now >= exp_dt:
+ token_data = [redacted:secret-assignment]
+
+ return token_data["token"]
+
+
+def main():
+ """Refresh token if needed, then exec gws with remaining args."""
+ if len(sys.argv) < 2:
+ print("Usage: gws_bridge.py ", file=sys.stderr)
+ sys.exit(1)
+
+ access_token = [redacted:secret-assignment]
+ env = os.environ.copy()
+ env["GOOGLE_WORKSPACE_CLI_TOKEN"] = access_token
+
+ result = subprocess.run(["gws"] + sys.argv[1:], env=env)
+ sys.exit(result.returncode)
+
+
+if __name__ == "__main__":
+ main()
```
- `skills/productivity/google-workspace/scripts/setup.py`
```diff
@@ -23,6 +23,7 @@
import argparse
import json
+import os
import subprocess
import sys
from pathlib import Path
```
```diff
@@ -128,7 +129,11 @@ def check_auth():
from google.auth.transport.requests import Request
try:
- creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
+ # Don't pass scopes — user may have authorized only a subset.
+ # Passing scopes forces google-auth to validate them on refresh,
+ # which fails with invalid_scope if the token has fewer scopes
+ # than requested.
+ creds = Credentials.from_authorized_user_file(str(TOKEN_PATH))
except Exception as e:
print(f"TOKEN_CORRUPT: {e}")
return False
```
```diff
@@ -137,8 +142,9 @@ def check_auth():
if creds.valid:
missing_scopes = _missing_scopes_from_payload(payload)
if missing_scopes:
- print(f"AUTH_SCOPE_MISMATCH: {_format_missing_scopes(missing_scopes)}")
- return False
+ print(f"AUTHENTICATED (partial): Token valid but missing {len(missing_scopes)} scopes:")
+ for s in missing_scopes:
+ print(f" - {s}")
print(f"AUTHENTICATED: Token valid at {TOKEN_PATH}")
return True
```
```diff
@@ -148,8 +154,9 @@ def check_auth():
TOKEN_PATH.write_text(creds.to_json())
missing_scopes = _missing_scopes_from_payload(_load_token_payload(TOKEN_PATH))
if missing_scopes:
- print(f"AUTH_SCOPE_MISMATCH: {_format_missing_scopes(missing_scopes)}")
- return False
+ print(f"AUTHENTICATED (partial): Token refreshed but missing {len(missing_scopes)} scopes:")
+ for s in missing_scopes:
+ print(f" - {s}")
print(f"AUTHENTICATED: Token refreshed at {TOKEN_PATH}")
return True
except Exception as e:
```
```diff
@@ -272,16 +279,33 @@ def exchange_auth_code(code: str):
_ensure_deps()
from google_auth_oauthlib.flow import Flow
+ from urllib.parse import parse_qs, urlparse
+
+ # Extract granted scopes from the callback URL if present
+ if returned_state and "scope" in parse_qs(urlparse(code).query if isinstance(code, str) and code.startswith("http") else {}):
+ granted_scopes = parse_qs(urlparse(code).query)["scope"][0].split()
+ else:
+ # Try to extract from code_or_url parameter
+ if isinstance(code, str) and code.startswith("http"):
+ params = parse_qs(urlparse(code).query)
+ if "scope" in params:
+ granted_scopes = params["scope"][0].split()
+ else:
+ granted_scopes = SCOPES
+ else:
+ granted_scopes = SCOPES
flow = Flow.from_client_secrets_file(
str(CLIENT_SECRET_PATH),
- scopes=SCOPES,
+ scopes=granted_scopes,
redirect_uri=pending_auth.get("redirect_uri", REDIRECT_URI),
state=pending_auth["state"],
code_verifier=pending_auth["code_verifier"],
)
try:
+ # Accept partial scopes — user may deselect some permissions in the consent screen
+ os.environ["OAUTHLIB_RELAX_TOKEN_SCOPE"] = "1"
flow.fetch_token(code=code)
except Exception as e:
print(f"ERROR: Token exchange failed: {e}")
```
```diff
@@ -290,11 +314,21 @@ def exchange_auth_code(code: str):
creds = flow.credentials
token_payload = [redacted:secret-assignment]
+
+ # Store only the scopes actually granted by the user, not what was requested.
+ # creds.to_json() writes the requested scopes, which causes refresh to fail
+ # with invalid_scope if the user only authorized a subset.
+ actually_granted = list(creds.granted_scopes or []) if hasattr(creds, "granted_scopes") and creds.granted_scopes else []
+ if actually_granted:
+ token_payload["scopes"] = actually_granted
+ elif granted_scopes != SCOPES:
+ # granted_scopes was extracted from the callback URL
+ token_payload["scopes"] = granted_scopes
+
missing_scopes = _missing_scopes_from_payload(token_payload)
if missing_scopes:
- print(f"ERROR: Refusing to save incomplete Google Workspace token. {_format_missing_scopes(missing_scopes)}")
- print(f"Existing token at {TOKEN_PATH} was left unchanged.")
- sys.exit(1)
+ print(f"WARNING: Token missing some Google Workspace scopes: {', '.join(missing_scopes)}")
+ print("Some services may not be available.")
TOKEN_PATH.write_text(json.dumps(token_payload, indent=2))
PENDING_AUTH_PATH.unlink(missing_ok=True)
```
- `tests/skills/test_google_oauth_setup.py`
```diff
@@ -211,14 +211,15 @@ def test_keeps_pending_auth_session_when_exchange_fails(self, setup_module, caps
assert setup_module.PENDING_AUTH_PATH.exists()
assert not setup_module.TOKEN_PATH.exists()
- def test_refuses_to_overwrite_existing_token_with_narrower_scopes(self, setup_module, capsys):
+ def test_accepts_narrower_scopes_with_warning(self, setup_module, capsys):
+ """Partial scopes are accepted with a warning (gws migration: v2.0)."""
setup_module.PENDING_AUTH_PATH.write_text(
json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"})
)
- setup_module.TOKEN_PATH.write_text(json.dumps({"token": "existing-token", "scopes": setup_module.SCOPES}))
+ setup_module.TOKEN_PATH.write_text(json.dumps({"token": "***", "scopes": setup_module.SCOPES}))
FakeFlow.credentials_payload = {
- "token": "narrow-token",
- "refresh_token": "refresh-token",
+ "token": "***",
+ "refresh_token": "***",
"token_uri": "https://oauth2.googleapis.com/token",
"client_id": "client-id",
"client_secret": "client-secret",
```
```diff
@@ -228,10 +229,12 @@ def test_refuses_to_overwrite_existing_token_with_narrower_scopes(self, setup_mo
],
}
- with pytest.raises(SystemExit):
- setup_module.exchange_auth_code("4/test-auth-code")
+ setup_module.exchange_auth_code("4/test-auth-code")
out = capsys.readouterr().out
- assert "refusing to save incomplete google workspace token" in out.lower()
- assert json.loads(setup_module.TOKEN_PATH.read_text())["token"] == "existing-token"
- assert setup_module.PENDING_AUTH_PATH.exists()
+ assert "warning" in out.lower()
+ assert "missing" in out.lower()
+ # Token is saved (partial scopes accepted)
+ assert setup_module.TOKEN_PATH.exists()
+ # Pending auth is cleaned up
+ assert not setup_module.PENDING_AUTH_PATH.exists()
```
- `tests/skills/test_google_workspace_api.py`
```diff
@@ -1,117 +1,175 @@
-"""Regression tests for Google Workspace API credential validation."""
+"""Tests for Google Workspace gws bridge and CLI wrapper."""
import importlib.util
import json
+import os
+import subprocess
import sys
import types
+from datetime import datetime, timedelta, timezone
from pathlib import Path
+from unittest.mock import MagicMock, patch
import pytest
-SCRIPT_PATH = (
+BRIDGE_PATH = (
+ Path(__file__).resolve().parents[2]
+ / "skills/productivity/google-workspace/scripts/gws_bridge.py"
+)
+API_PATH = (
Path(__file__).resolve().parents[2]
/ "skills/productivity/google-workspace/scripts/google_api.py"
)
-class FakeAuthorizedCredentials:
- def __init__(self, *, valid=True, expired=False, refresh_token="[redacted:secret-assignment]"):
- self.valid = valid
- self.expired = expired
- self.refresh_token = [redacted:secret-assignment]
- self.refresh_calls = 0
-
- def refresh(self, _request):
- self.refresh_calls += 1
- self.valid = True
- self.expired = False
-
- def to_json(self):
- return json.dumps({
- "token": "refreshed-token",
- "refresh_token": self.refresh_token,
- "token_uri": "https://oauth2.googleapis.com/token",
- "client_id": "client-id",
- "client_secret": "client-secret",
- "scopes": [
- "https://www.googleapis.com/auth/gmail.readonly",
- "https://www.googleapis.com/auth/gmail.send",
- "https://www.googleapis.com/auth/gmail.modify",
- "https://www.googleapis.com/auth/calendar",
- "https://www.googleapis.com/auth/drive.readonly",
- "https://www.googleapis.com/auth/contacts.readonly",
- "https://www.googleapis.com/auth/spreadsheets",
- "https://www.googleapis.com/auth/documents.readonly",
- ],
- })
-
-
-class FakeCredentialsFactory:
- creds = FakeAuthorizedCredentials()
-
- @classmethod
- def from_authorized_user_file(cls, _path, _scopes):
- return cls.creds
[redacted:email]
+def bridge_module(monkeypatch, tmp_path):
+ hermes_home = tmp_path / ".hermes"
+ hermes_home.mkdir()
+ monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+ spec = importlib.util.spec_from_file_location("gws_bridge_test", BRIDGE_PATH)
+ module = importlib.util.module_from_spec(spec)
+ assert spec.loader is not None
+ spec.loader.exec_module(module)
+ return module
@pytest.fixture
-def google_api_module(monkeypatch, tmp_path):
- google_module = types.ModuleType("google")
- oauth2_module = types.ModuleType("google.oauth2")
- credentials_module = [redacted:secret-assignment]"google.oauth2.credentials")
- credentials_module.Credentials = [redacted:secret-assignment]
- auth_module = types.ModuleType("google.auth")
- transport_module = types.ModuleType("google.auth.transport")
- requests_module = types.ModuleType("google.auth.transport.requests")
- requests_module.Request = object
-
- monkeypatch.setitem(sys.modules, "google", google_module)
- monkeypatch.setitem(sys.modules, "google.oauth2", oauth2_module)
- monkeypatch.setitem(sys.modules, "google.oauth2.credentials", credentials_module)
- monkeypatch.setitem(sys.modules, "google.auth", auth_module)
- monkeypatch.setitem(sys.modules, "google.auth.transport", transport_module)
- monkeypatch.setitem(sys.modules, "google.auth.transport.requests", requests_module)
-
- spec = importlib.util.spec_from_file_location("google_workspace_api_test", SCRIPT_PATH)
+def api_module(monkeypatch, tmp_path):
+ hermes_home = tmp_path / ".hermes"
+ hermes_home.mkdir()
+ monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+ spec = importlib.util.spec_from_file_location("gws_api_test", API_PATH)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
-
- monkeypatch.setattr(module, "TOKEN_PATH", tmp_path / "google_token.json")
return module
-def _write_token(path: Path, scopes):
- path.write_text(json.dumps({
- "token": "access-token",
- "refresh_token": "refresh-token",
+def _write_token(path: Path, *, token="[redacted:secret-assignment]", expiry=None, **extra):
+ data = {
+ "token": token,
+ "refresh_token": "1//refresh",
+ "client_id": "123.apps.googleusercontent.com",
+ "client_secret": "secret",
"token_uri": "https://oauth2.googleapis.com/token",
- "client_id": "client-id",
- "client_secret": "client-secret",
- "scopes": scopes,
- }))
+ **extra,
+ }
+ if expiry is not None:
+ data["expiry"] = expiry
+ path.write_text(json.dumps(data))
+
+
+def test_bridge_returns_valid_token(bridge_module, tmp_path):
+ """Non-expired token is returned without refresh."""
+ future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+ token_path = [redacted:secret-assignment]
+ _write_token(token_path, token="[redacted:secret-assignment]", expiry=future)
+
+ result = bridge_module.get_valid_token()
+ assert result == "ya29.valid"
+
+
+def test_bridge_refreshes_expired_token(bridge_module, tmp_path):
+ """Expired token triggers a refresh via token_uri."""
+ past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
+ token_path = [redacted:secret-assignment]
+ _write_token(token_path, token="[redacted:secret-assignment]", expiry=past)
+ mock_resp = MagicMock()
+ mock_resp.read.return_value = json.dumps({
+ "access_token": "ya29.refreshed",
+ "expires_in": 3600,
+ }).encode()
+ mock_resp.__enter__ = lambda s: s
+ mock_resp.__exit__ = MagicMock(return_value=False)
-def test_get_credentials_rejects_missing_scopes(google_api_module, capsys):
- FakeCredentialsFactory.creds = FakeAuthorizedCredentials(valid=True)
- _write_token(google_api_module.TOKEN_PATH, [
- "https://www.googleapis.com/auth/drive.readonly",
- "https://www.googleapis.com/auth/spreadsheets",
- ])
+ with patch("urllib.request.urlopen", return_value=mock_resp):
+ result = bridge_module.get_valid_token()
+ assert result == "ya29.refreshed"
+ # Verify persisted
+ saved = json.loads(token_path.read_text())
+ assert saved["token"] == "ya29.refreshed"
+
+
+def test_bridge_exits_on_missing_token(bridge_module):
+ """Missing token file causes exit with code 1."""
with pytest.raises(SystemExit):
- google_api_module.get_credentials()
+ bridge_module.get_valid_token()
+
+
+def test_bridge_main_injects_token_env(bridge_module, tmp_path):
+ """main() sets GOOGLE_WORKSPACE_CLI_TOKEN in subprocess env."""
+ future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+ token_path = [redacted:secret-assignment]
+ _write_token(token_path, token="[redacted:secret-assignment]", expiry=future)
+
+ captured = {}
+
+ def capture_run(cmd, **kwargs):
+ captured["cmd"] = cmd
+ captured["env"] = kwargs.get("env", {})
+ return MagicMock(returncode=0)
+
+ with patch.object(sys, "argv", ["gws_bridge.py", "gmail", "+triage"]):
+ with patch.object(subprocess, "run", side_effect=capture_run):
+ with pytest.raises(SystemExit):
+ bridge_module.main()
+
+ assert captured["env"]["GOOGLE_WORKSPACE_CLI_TOKEN"] == "ya29.injected"
+ assert captured["cmd"] == ["gws", "gmail", "+triage"]
+
+
+def test_api_calendar_list_uses_agenda_by_default(api_module):
+ """calendar list without dates uses +agenda helper."""
+ captured = {}
+
+ def capture_run(cmd, **kwargs):
+ captured["cmd"] = cmd
+ return MagicMock(returncode=0)
+
+ args = api_module.argparse.Namespace(
+ start="", end="", max=25, calendar="primary", func=api_module.calendar_list,
+ )
+
+ with patch.object(subprocess, "run", side_effect=capture_run):
+ with pytest.raises(SystemExit):
+ api_module.calendar_list(args)
+
+ gws_args = captured["cmd"][2:] # skip python + bridge path
+ assert "calendar" in gws_args
+ assert "+agenda" in gws_args
+ assert "--days" in gws_args
+
- err = capsys.readouterr().err
- assert "missing google workspace scopes" in err.lower()
- assert "gmail.send" in err
+def test_api_calendar_list_respects_date_range(api_module):
+ """calendar list with --start/--end uses raw events list API."""
+ captured = {}
+ def capture_run(cmd, **kwargs):
+ captured["cmd"] = cmd
+ return MagicMock(returncode=0)
-def test_get_credentials_accepts_full_scope_token(google_api_module):
- FakeCredentialsFactory.creds = FakeAuthorizedCredentials(valid=True)
- _write_token(google_api_module.TOKEN_PATH, list(google_api_module.SCOPES))
+ args = api_module.argparse.Namespace(
+ start="2026-04-01T00:00:00Z",
+ end="2026-04-07T23:59:59Z",
+ max=25,
+ calendar="primary",
+ func=api_module.calendar_list,
+ )
- creds = google_api_module.get_credentials()
+ with patch.object(subprocess, "run", side_effect=capture_run):
+ with pytest.raises(SystemExit):
+ api_module.calendar_list(args)
- assert creds is FakeCredentialsFactory.creds
+ gws_args = captured["cmd"][2:]
+ assert "events" in gws_args
+ assert "list" in gws_args
+ params_idx = gws_args.index("--params")
+ params = json.loads(gws_args[params_idx + 1])
+ assert params["timeMin"] == "2026-04-01T00:00:00Z"
+ assert params["timeMax"] == "2026-04-07T23:59:59Z"
```
- **#6713: feat(skills): migrate google-workspace to gws CLI backend**. closed without merge
- maintainer direction: category `merged_via`; implication `superseded`; refs `#6788`
- **#23465: feat(google-workspace): add gmail attachment list/get CLI verbs (#22872)**. closed without merge
- maintainer direction: category `volume_quota`; implication `active`
## Fix status
- Merged to main in PR #6788 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-04-09`. This confirmation is old and may be out of date (a revert/reopen would not have invalidated it here); re-verify.
- Subsystem: `skills`