Controlling Google Sheets via Claude Code

google-auth
google-auth-oauthlib
google-api-python-client
python-dotenv

pip install -r requirements.txt

GOOGLE_OAUTH_CREDENTIALS=./cli/client_secret.json
GOOGLE_SHEET_URL=https://docs.google.com/spreadsheets/d/YOUR_SHEET_ID/edit
GOOGLE_SHEET_TAB=Sheet1

import os
import pickle
from pathlib import Path

from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
CLI_ROOT = Path(__file__).resolve().parent
PROJECT_ROOT = CLI_ROOT.parent
TOKEN_PATH = CLI_ROOT / "token.pickle"


def _resolve_runtime_path(raw_path: str) -> Path:
    path = Path(raw_path).expanduser()
    if path.is_absolute():
        return path
    candidates = [
        (PROJECT_ROOT / path).resolve(),
        (CLI_ROOT / path).resolve(),
    ]
    for candidate in candidates:
        if candidate.exists():
            return candidate
    return candidates[0]


# ── Auth ──────────────────────────────────────────────────────────

def get_service():
    """Authenticate and return a Google Sheets API service object."""
    creds = None

    if TOKEN_PATH.exists():
        with open(TOKEN_PATH, "rb") as f:
            creds = pickle.load(f)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                str(_resolve_runtime_path(os.environ["GOOGLE_OAUTH_CREDENTIALS"])),
                SCOPES,
            )
            creds = flow.run_local_server(port=0)

        with open(TOKEN_PATH, "wb") as f:
            pickle.dump(creds, f)

    return build("sheets", "v4", credentials=creds)


# ── Read ──────────────────────────────────────────────────────────

def read_rows(service, sheet_id: str, tab_name: str):
    """Return (rows, headers). rows is a list of dicts keyed by header name."""
    tab_range = f"'{tab_name}'!A1:ZZ"
    result = (
        service.spreadsheets()
        .values()
        .get(spreadsheetId=sheet_id, range=tab_range)
        .execute()
    )
    values = result.get("values", [])
    if not values:
        return [], []

    headers = values[0]
    rows = []
    for row_values in values[1:]:
        row_dict = {
            header: (row_values[i] if i < len(row_values) else "")
            for i, header in enumerate(headers)
        }
        rows.append(row_dict)

    return rows, headers


# ── Sheet metadata ─────────────────────────────────────────────────

def get_sheet_int_id(service, spreadsheet_id: str, tab_name: str) -> int:
    """Return the integer sheetId for a named tab."""
    meta = service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute()
    sheet = next(
        (s for s in meta["sheets"] if s["properties"]["title"] == tab_name), None
    )
    if sheet is None:
        raise ValueError(f"Tab '{tab_name}' not found in spreadsheet")
    return sheet["properties"]["sheetId"]


def get_sheet_column_count(service, spreadsheet_id: str, tab_name: str) -> int:
    """Return the current grid column count for a named tab."""
    meta = service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute()
    sheet = next(
        (s for s in meta["sheets"] if s["properties"]["title"] == tab_name), None
    )
    if sheet is None:
        raise ValueError(f"Tab '{tab_name}' not found in spreadsheet")
    return sheet["properties"]["gridProperties"]["columnCount"]


# ── Column management ──────────────────────────────────────────────

def ensure_output_columns(
    service, sheet_id: str, tab_name: str, headers: list, needed: list,
    sheet_int_id: int,
) -> dict:
    """
    Ensure each column in needed exists. Adds missing headers to the sheet.
    Returns dict: column_name -> 0-based column index.
    """
    missing = [c for c in needed if c not in headers]

    if missing:
        required_cols = len(headers) + len(missing)
        current_cols = get_sheet_column_count(service, sheet_id, tab_name)
        if required_cols > current_cols:
            _expand_sheet(service, sheet_id, sheet_int_id, required_cols)

    col_indices = {}
    for col_name in needed:
        if col_name in headers:
            col_indices[col_name] = headers.index(col_name)
        else:
            new_idx = len(headers)
            headers.append(col_name)
            write_cell(service, sheet_id, sheet_int_id, 1, new_idx, col_name)
            col_indices[col_name] = new_idx

    return col_indices


def _expand_sheet(
    service, spreadsheet_id: str, sheet_int_id: int, min_columns: int
) -> None:
    """Expand the sheet grid to at least min_columns columns."""
    (
        service.spreadsheets()
        .batchUpdate(
            spreadsheetId=spreadsheet_id,
            body={
                "requests": [{
                    "updateSheetProperties": {
                        "properties": {
                            "sheetId": sheet_int_id,
                            "gridProperties": {"columnCount": min_columns},
                        },
                        "fields": "gridProperties.columnCount",
                    }
                }]
            },
        )
        .execute()
    )


# ── Write ──────────────────────────────────────────────────────────

def write_cell(
    service,
    spreadsheet_id: str,
    sheet_int_id: int,
    row_num: int,
    col_idx: int,
    value: str,
) -> None:
    """
    Write a single cell as plain, unformatted text.
    Uses batchUpdate so Sheets doesn't auto-link URLs or format text blue.
    row_num is 1-based (1 = header row, 2 = first data row).
    col_idx is 0-based.
    """
    (
        service.spreadsheets()
        .batchUpdate(
            spreadsheetId=spreadsheet_id,
            body={
                "requests": [{
                    "updateCells": {
                        "rows": [{
                            "values": [{
                                "userEnteredValue": {"stringValue": value},
                                "userEnteredFormat": {
                                    "textFormat": {
                                        "foregroundColorStyle": {
                                            "rgbColor": {"red": 0, "green": 0, "blue": 0}
                                        },
                                        "underline": False,
                                        "link": {"uri": ""},
                                    }
                                },
                            }]
                        }],
                        "fields": (
                            "userEnteredValue,"
                            "userEnteredFormat.textFormat.foregroundColorStyle,"
                            "userEnteredFormat.textFormat.underline,"
                            "userEnteredFormat.textFormat.link"
                        ),
                        "range": {
                            "sheetId": sheet_int_id,
                            "startRowIndex": row_num - 1,
                            "endRowIndex": row_num,
                            "startColumnIndex": col_idx,
                            "endColumnIndex": col_idx + 1,
                        },
                    }
                }]
            },
        )
        .execute()
    )

import re

def extract_sheet_id(url: str) -> str:
    match = re.search(r"/spreadsheets/d/([a-zA-Z0-9-_]+)", url)
    if not match:
        raise ValueError(f"Could not parse sheet ID from URL: {url}")
    return match.group(1)

import os
from dotenv import load_dotenv
from sheets_client import get_service, read_rows, write_cell, get_sheet_int_id, ensure_output_columns

load_dotenv()

SHEET_URL = os.environ["GOOGLE_SHEET_URL"]
TAB_NAME  = os.environ["GOOGLE_SHEET_TAB"]
SHEET_ID  = extract_sheet_id(SHEET_URL)

# 1. Authenticate
service = get_service()  # Opens browser on first run; uses token.pickle after

# 2. Read all rows
rows, headers = read_rows(service, SHEET_ID, TAB_NAME)

# 3. Get the integer sheet ID (needed for write/format operations)
sheet_int_id = get_sheet_int_id(service, SHEET_ID, TAB_NAME)

# 4. Ensure output columns exist (creates them if missing)
col_map = ensure_output_columns(
    service, SHEET_ID, TAB_NAME, headers,
    needed=["output_col_1", "output_col_2"],
    sheet_int_id=sheet_int_id,
)

# 5. Write to a cell
# row_num is 1-based: row 1 = headers, row 2 = first data row
for i, row in enumerate(rows):
    row_num = i + 2  # offset for header row
    write_cell(service, SHEET_ID, sheet_int_id, row_num, col_map["output_col_1"], "hello")

your-project/
├── cli/
│   ├── sheets_client.py          # The module above
│   ├── client_secret.json        # Downloaded from Google Cloud Console
│   ├── token.pickle              # Auto-generated on first run (gitignore this)
│   └── your_script.py
├── .env                          # GOOGLE_OAUTH_CREDENTIALS, GOOGLE_SHEET_URL, etc.
├── .gitignore                    # Must include: .env, token.pickle, client_secret.json
└── requirements.txt

"Here's my sheet: https://docs.google.com/spreadsheets/d/1BxiMVs0XRA5.../edit
— tab is called Leads. Read all the rows and tell me what columns exist."

"The data is in ~/Downloads/leads.csv. Read it and write each company
name into column B of my sheet."

"Add two new columns to the sheet — status and notes — then fill status
with 'pending' for every row."

"Run the pipeline on my sheet. Sheet URL is [url], tab is Master,
process rows 2–50."

You:     "Here's my sheet [url], tab = Sheet1.
          Add a column called 'enriched' and fill it with 'yes' for every row
          where column D is not empty."

Claude:  reads the sheet → finds column D → identifies non-empty rows
         → calls ensure_output_columns() → loops write_cell() for each match
         → "Done. Wrote 'yes' to 34 rows in column 'enriched'."