credentials.json

{
  "client_id": "CLIENTIDHERE",
  "client_secret": "SECRET HERE",
  "refresh_token": "1//REFRESHTOKENHERE"
}

drive.py

#!/usr/bin/env python3
"""Google Drive REST skill: search, download, and inspect Drive files.

Uses the Drive v3 REST API with an OAuth refresh token (auto-refreshed).
Credentials are read from env vars (GDRIVE_*) or a credentials.json file.

Subcommands:
  search   Search files by name / content / both.
  recent   List recently modified files.
  get      Show metadata for one file id.
  download Download a file's bytes (or export a Google Doc/Sheet/Slide).

No third-party deps: stdlib only.
"""
import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

API = "https://www.googleapis.com/drive/v3"
TOKEN_URL = "https://oauth2.googleapis.com/token"
SKILL_DIR = Path(__file__).resolve().parent.parent
CRED_FILE = SKILL_DIR / "credentials.json"
CACHE_FILE = SKILL_DIR / ".token_cache.json"

# Export mapping for Google Workspace docs -> a downloadable MIME type.
EXPORT_MIME = {
    "application/vnd.google-apps.document": "text/markdown",
    "application/vnd.google-apps.spreadsheet": "text/csv",
    "application/vnd.google-apps.presentation": "application/pdf",
    "application/vnd.google-apps.drawing": "image/png",
    "application/vnd.google-apps.script": "application/vnd.google-apps.script+json",
}
# Fallback export for Docs if the preferred type is rejected.
EXPORT_FALLBACK = {
    "application/vnd.google-apps.document": "text/plain",
}
TEXT_MIMES = {"text/markdown", "text/plain", "text/csv", "text/html",
              "application/json", "application/vnd.google-apps.script+json"}


def die(msg, code=1):
    print(f"error: {msg}", file=sys.stderr)
    sys.exit(code)


def creds():
    cid = os.environ.get("GDRIVE_CLIENT_ID")
    sec = os.environ.get("GDRIVE_CLIENT_SECRET")
    rt = os.environ.get("GDRIVE_REFRESH_TOKEN")
    if not (cid and sec and rt):
        if not CRED_FILE.exists():
            die("no credentials. Set GDRIVE_CLIENT_ID/GDRIVE_CLIENT_SECRET/"
                "GDRIVE_REFRESH_TOKEN or create credentials.json next to the skill.")
        d = json.loads(CRED_FILE.read_text())
        cid = cid or d.get("client_id")
        sec = sec or d.get("client_secret")
        rt = rt or d.get("refresh_token")
    if not (cid and sec and rt):
        die("credentials incomplete (need client_id, client_secret, refresh_token).")
    return cid, sec, rt


def access_token():
    """Return a live access token, refreshing (and caching) as needed."""
    cid, sec, rt = creds()
    now = time.time()
    if CACHE_FILE.exists():
        try:
            c = json.loads(CACHE_FILE.read_text())
            if c.get("access_token") and c.get("exp", 0) - now > 60:
                return c["access_token"]
        except Exception:
            pass
    data = urllib.parse.urlencode({
        "client_id": cid, "client_secret": sec,
        "refresh_token": rt, "grant_type": "refresh_token",
    }).encode()
    req = urllib.request.Request(TOKEN_URL, data=data,
                                 headers={"Content-Type": "application/x-www-form-urlencoded"})
    try:
        resp = json.loads(urllib.request.urlopen(req, timeout=30).read().decode())
    except urllib.error.HTTPError as e:
        die(f"token refresh failed ({e.code}): {e.read().decode()[:300]}")
    token = resp["access_token"]
    CACHE_FILE.write_text(json.dumps({
        "access_token": token, "exp": now + int(resp.get("expires_in", 3599)),
    }))
    try:
        os.chmod(CACHE_FILE, 0o600)
    except OSError:
        pass
    return token


def drive(path, params=None, headers=None, binary=False):
    url = API + path
    if params:
        url += "?" + urllib.parse.urlencode(params)
    h = {"Authorization": "Bearer " + access_token()}
    if headers:
        h.update(headers)
    req = urllib.request.Request(url, headers=h)
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            return r.read() if binary else r.read().decode()
    except urllib.error.HTTPError as e:
        body = e.read().decode(errors="replace")
        try:
            ej = json.loads(body).get("error", {})
            msg = ej.get("message") or body
        except Exception:
            msg = body
        die(f"Drive API {e.code}: {msg}")


def fmt_size(n):
    try:
        n = int(n)
    except (TypeError, ValueError):
        return ""
    for unit in ("B", "KB", "MB", "GB"):
        if n < 1024:
            return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}"
        n /= 1024
    return f"{n:.1f}TB"


def print_rows(files):
    if not files:
        print("(no files)")
        return
    for f in files:
        name = f.get("name", "")
        fid = f.get("id", "")
        mime = f.get("mimeType", "")
        mod = f.get("modifiedTime", "").replace("T", " ").split(".")[0]
        size = fmt_size(f.get("size", ""))
        kind = "DOC" if "document" in mime else (
               "SHEET" if "spreadsheet" in mime else (
               "SLIDE" if "presentation" in mime else "FILE"))
        print(f"{kind:5} {mod:20} {size:>8}  {name}")
        print(f"      id={fid}")


def cmd_search(a):
    fields = "nextPageToken,files(id,name,mimeType,modifiedTime,size)"
    if a.raw:
        q = a.raw
    else:
        term = a.term.replace("'", "\\'")
        col = {"name": "name", "content": "fullText", "both": None}[a.in_]
        if col:
            q = f"{col} contains '{term}'"
        else:
            q = f"name contains '{term}' or fullText contains '{term}'"
    # Drive forbids orderBy on fullText queries (results are relevance-ranked).
    order_by = "recency desc" if a.in_ == "name" else ""
    params = {"q": q, "pageSize": a.num, "fields": fields, "orderBy": order_by}
    params = {k: v for k, v in params.items() if v != ""}
    data = json.loads(drive("/files", params))
    print_rows(data.get("files", []))


def cmd_recent(a):
    fields = "files(id,name,mimeType,modifiedTime,size)"
    data = json.loads(drive("/files", {
        "pageSize": a.num, "orderBy": "modifiedTime desc", "fields": fields}))
    print_rows(data.get("files", []))


def cmd_get(a):
    f = json.loads(drive(f"/files/{a.file_id}", {
        "fields": "id,name,mimeType,size,createdTime,modifiedTime,"
                  "description,parents,owners(emailAddress),webViewLink,exportLinks"}))
    for k, v in f.items():
        print(f"{k}: {v}")


def cmd_download(a):
    meta = json.loads(drive(f"/files/{a.file_id}", {
        "fields": "id,name,mimeType,size"}))
    mime = meta.get("mimeType", "")
    name = meta.get("name", "download")
    is_native = mime.startswith("application/vnd.google-apps.")
    if is_native:
        export = a.mime or EXPORT_MIME.get(mime)
        if not export:
            die(f"no default export for {mime}; pass --mime (e.g. application/pdf).")
        url = f"{API}/files/{a.file_id}/export"
        content = drive(f"/files/{a.file_id}/export", {"mimeType": export}, binary=True)
        # try fallback if markdown export was rejected
        if not content and mime in EXPORT_FALLBACK and export != EXPORT_FALLBACK[mime]:
            content = drive(f"/files/{a.file_id}/export",
                            {"mimeType": EXPORT_FALLBACK[mime]}, binary=True)
            export = EXPORT_FALLBACK[mime]
        eff_mime = export
    else:
        url = f"{API}/files/{a.file_id}"
        content = drive(f"/files/{a.file_id}", {"alt": "media"}, binary=True)
        eff_mime = mime

    is_text = eff_mime in TEXT_MIMES or eff_mime.startswith("text/")
    out = a.out
    if not out and is_text and not a.file:
        # stream to stdout for agent consumption
        sys.stdout.buffer.write(content)
        sys.stdout.buffer.flush()
        return
    if not out:
        out = name
    Path(out).write_bytes(content)
    print(f"wrote {len(content)} bytes -> {out}  ({eff_mime})")


def main():
    p = argparse.ArgumentParser(prog="drive.py", description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("search", help="search files by name/content")
    s.add_argument("term", nargs="?", help="search term")
    s.add_argument("--in", dest="in_", choices=["name", "content", "both"],
                   default="name", help="where to search (default: name)")
    s.add_argument("--num", type=int, default=15)
    s.add_argument("--raw", help="raw Drive query, e.g. \"name='X' and modifiedTime > '2024-01-01'\"")
    s.set_defaults(func=cmd_search)

    r = sub.add_parser("recent", help="list recently modified files")
    r.add_argument("--num", type=int, default=15)
    r.set_defaults(func=cmd_recent)

    g = sub.add_parser("get", help="show metadata for a file id")
    g.add_argument("file_id")
    g.set_defaults(func=cmd_get)

    d = sub.add_parser("download", help="download / export a file")
    d.add_argument("file_id")
    d.add_argument("--out", help="output path (default: original name, or stdout for text)")
    d.add_argument("--mime", help="export MIME for Google docs (default: markdown/csv/pdf/png)")
    d.add_argument("--file", action="store_true", help="always write to a file, never stdout")
    d.set_defaults(func=cmd_download)

    a = p.parse_args()
    if a.cmd == "search" and not a.term and not a.raw:
        p.error("search requires a term or --raw")
    a.func(a)


if __name__ == "__main__":
    main()

SKILL.md

name	google-drive-api
description	Search, list, inspect, and download files from a user's Google Drive via the Drive REST API (auto-refreshing OAuth). Use when the user asks to find, look up, read, or download a Google Drive file or Google Doc/Sheet/Slide by name or content; returns clean id+name listings and can export Docs to markdown. Bypasses the remote Drive MCP (which has a broken OAuth flow in opencode).

Google Drive API

Search and download files from Google Drive using the Drive v3 REST API with a stored OAuth refresh token. No remote MCP required — this sidesteps opencode's broken remote-MCP OAuth flow entirely.

Prerequisites

Credentials live in credentials.json next to this skill (chmod 600), holding client_id, client_secret, and refresh_token. The token is auto-refreshed and cached in .token_cache.json. You can also override via env vars:

export GDRIVE_CLIENT_ID="..."
export GDRIVE_CLIENT_SECRET="..."
export GDRIVE_REFRESH_TOKEN="..."

Point at the script once per shell:

export GDRIVE="${GDRIVE:-$HOME/.config/opencode/skills/google-drive-api/scripts/drive.py}"

Run with python3 "$GDRIVE" <command> (stdlib only — no pip installs).

Commands

search — find files by name, content, or both

python3 "$GDRIVE" search "eyegaze"                       # by name (default)
python3 "$GDRIVE" search "strabismus" --in content        # full-text body search
python3 "$GDRIVE" search "eyegaze" --in both              # name OR content
python3 "$GDRIVE" search --raw "name='X' and modifiedTime > '2025-01-01'"

Options: --num N (page size, default 15), --in name|content|both, --raw "<Drive query>" (raw query, see Drive query syntax).

Output rows: <kind> <modifiedTime> <size> <name> then id=.... Keep the id for download / get.

recent — recently modified files

python3 "$GDRIVE" recent --num 10

get — metadata for one file id

python3 "$GDRIVE" get <fileId>

download — download bytes or export a Google Doc/Sheet/Slide

python3 "$GDRIVE" download <fileId>                      # Doc -> markdown to stdout
python3 "$GDRIVE" download <fileId> --out report.md      # write to a file
python3 "$GDRIVE" download <fileId> --mime application/pdf --out doc.pdf

Defaults for Google Workspace files: Doc → text/markdown (falls back to text/plain), Sheet → text/csv, Slide → application/pdf, Drawing → image/png. Regular files download as-is. Text exports stream to stdout when no --out is given (ideal for reading content); binary always writes to a file named after the original. Use --file to force writing to a file.

Workflow

1. search "<term>" to find candidates and grab the id.
2. get <id> if you need metadata or the view link.
3. download <id> to read/export it (Docs stream as markdown to stdout).

Notes & gotchas

• The token was issued with the broad https://www.googleapis.com/auth/drive scope (full access). If you want read-only, re-run a browser OAuth flow requesting drive.readonly and replace refresh_token in credentials.json.
• credentials.json and .token_cache.json contain secrets — keep them chmod 600 and never commit them.
• Content (fullText) searches cannot be sorted; results are relevance-ranked by Drive (handled automatically).
• For large binary files, prefer --out.