Developers › Python examples

Python examples

Complete, runnable programs for the tasks most integrations start with. They are not one snippet per endpoint: each recipe shows a whole process, including pagination, waiting for a run, and error handling. For one endpoint's fields, use its group page; every operation there has a curl example.

API client helper

File sheetgo_api.py · tested against the live API by the smoke run before each release

import os
import time

import requests

API_URL = os.environ.get("SHEETGO_API_URL", "https://dev-api.sheetgo.com/rest").rstrip("/")
API_KEY = os.environ["SHEETGO_API_KEY"]  # sg_..., created on your Sheetgo Account page


class SheetgoError(Exception):
    """Any non-2xx answer. `code` is a stable string you can branch on, e.g. `invalid_column`."""

    def __init__(self, status, code, detail, body):
        super().__init__(f"HTTP {status} {code}: {detail}")
        self.status, self.code, self.detail, self.body = status, code, detail, body


def _error(resp):
    try:
        body = resp.json()
    except ValueError:
        body = {"detail": resp.text}
    if isinstance(body.get("detail"), dict):  # some routes nest the error in "detail"
        body = body["detail"]
    detail = body.get("detail") or body.get("message")
    if isinstance(detail, list):  # 422 request validation: [{"loc": [...], "msg": ..., "type": ...}]
        code = "validation_error"
        detail = "; ".join(f'{".".join(map(str, e["loc"]))}: {e["msg"]}' for e in detail)
    else:  # {code, detail}; Outlook/Gmail {status: 0, error, detail}; 429 {error, message, retry_after}
        code = body.get("code") or body.get("error") or str(resp.status_code)
    return SheetgoError(resp.status_code, code, detail, body)


class SheetgoAPI:
    def __init__(self, api_url=API_URL, api_key=API_KEY, max_retries=4, timeout=60):
        self.api_url, self.max_retries, self.timeout = api_url, max_retries, timeout
        self.session = requests.Session()
        self.session.headers["Authorization"] = f"Bearer {api_key}"

    def request(self, method, path, **kwargs):
        url = path if path.startswith("http") else self.api_url + path
        for attempt in range(self.max_retries + 1):
            resp = self.session.request(method, url, timeout=self.timeout, **kwargs)
            if resp.ok:
                return resp.json() if resp.content else None
            last = attempt == self.max_retries
            if resp.status_code == 429 and not last:
                time.sleep(int(resp.headers.get("Retry-After", 2 ** attempt)))
            elif resp.status_code in (502, 503, 504) and method == "GET" and not last:
                time.sleep(2 ** attempt)
            else:
                raise _error(resp)

    def get(self, path, **params):
        return self.request("GET", path, params=params)

    def post(self, path, json=None, **kwargs):
        return self.request("POST", path, json=json, **kwargs)

    def put(self, path, json=None):
        return self.request("PUT", path, json=json)

    def delete(self, path, json=None):
        return self.request("DELETE", path, json=json)


if __name__ == "__main__":
    api = SheetgoAPI()
    try:
        api.get("/beta/workflows/no-such-id")
    except SheetgoError as e:
        print("error handling works:", e.status, e.code)

Read every page of a list

File paginate.py · tested against the live API by the smoke run before each release

from sheetgo_api import SheetgoAPI


def paginate(api, path, per_page=100, **params):
    """Yield every item of a list endpoint, one page request at a time."""
    page = 1
    while True:
        body = api.get(path, page=page, per_page=per_page, **params)
        yield from body["results"]
        if body.get("truncated"):
            print(f"warning: {path} was capped; narrow the query to see everything")
        if not body["results"] or page >= (body.get("total") or 0):
            return
        page += 1


if __name__ == "__main__":
    api = SheetgoAPI()
    workflows = list(paginate(api, "/beta/workflows/"))
    print(f"{len(workflows)} workflows")
    for wf in workflows[:5]:
        print(f'- {wf["name"]} ({wf["id"]}): {len(wf["automations"])} automations')

    sheets = list(paginate(api, "/beta/files/drive/search", q="Game Industry", type="gsheets"))
    print(f"{len(sheets)} Google Sheets match 'Game Industry'")

Create, run and wait for an automation

File create_run_wait.py

import sys
import time

from sheetgo_api import SheetgoAPI, SheetgoError


def run_and_wait(api, automation_id, every=5, timeout=900):
    path = f"/beta/automations/{automation_id}/run"
    before = api.get(path)
    api.post(path)
    seen_running, deadline = False, time.monotonic() + timeout
    while time.monotonic() < deadline:
        time.sleep(every)
        status = api.get(path)
        if status["execution"] == "running":
            seen_running = True
        elif seen_running or status != before:
            return status
    raise TimeoutError(f"automation {automation_id} did not finish within {timeout}s")


if __name__ == "__main__":
    source_id, tab, destination_id = sys.argv[1:4]
    api = SheetgoAPI()
    try:
        workflow = api.post("/beta/workflows/", {"name": "API recipe: copy a tab"})
        automation = api.post("/beta/automations/", {
            "workflow_id": workflow["id"],
            "name": f"Copy {tab}",
            "sources": [{"type": "file", "provider": "drive", "id": source_id,
                         "file_type": "gsheets", "sheet_tabs": tab}],
            "destinations": [{"type": "file", "provider": "drive", "id": destination_id,
                              "file_type": "gsheets", "sheet_name": f"{tab} copy"}],
        })
        print("created automation", automation["id"], "in workflow", workflow["id"])

        status = run_and_wait(api, automation["id"])
    except SheetgoError as e:
        # e.g. 422 invalid_column, 403 when the key's scope does not allow it, 412 when an account is not connected
        raise SystemExit(f"{e.code}: {e.detail}")

    if status["result"] == "success":
        print(f'done in {status.get("duration_seconds")}s')
    else:
        print(f'run failed ({status["status_code"]}): {status.get("message")}  help: {status.get("docs_link")}')

Find a file and build an automation payload

File file_to_automation.py · tested against the live API by the smoke run before each release

import json
import sys

from sheetgo_api import SheetgoAPI, SheetgoError
from paginate import paginate


def find_spreadsheet(api, name, provider="drive"):
    for item in paginate(api, f"/beta/files/{provider}/search", q=name):
        if item["name"] == name and item.get("supports_structure"):
            return item
    raise SystemExit(f"no spreadsheet named {name!r} on {provider}")


def column_letters(api, item, tab, header_row=1):
    """{header: letter} for one tab, e.g. {"Platform": "C"}."""
    body = api.get(f'/beta/files/{item["provider"]}/{item["id"]}/structure',
                   include="columns", tab=tab, header_row=header_row)
    for t in body["spreadsheet"]["tabs"]:
        if t["name"] == tab and t["columns"] is not None:
            return {c["header"]: c["letter"] for c in t["columns"] if c["header"]}
    raise SystemExit(f"tab {tab!r} not found or its columns were not read")


if __name__ == "__main__":
    # python file_to_automation.py "<file name>" <tab> <header> <value>
    file_name, tab, header, value = sys.argv[1:5] or ["Game Industry", "Hundred", "Platform", "Wii"]
    api = SheetgoAPI()
    try:
        source = find_spreadsheet(api, file_name)
        letters = column_letters(api, source, tab)
    except SheetgoError as e:
        raise SystemExit(f"{e.code}: {e.detail}")
    print(f"{file_name} / {tab}: {letters}")

    payload = {
        "workflow_id": "<workflow id>",
        "name": f"{tab} where {header} is {value}",
        "sources": [{"type": "file", "provider": source["provider"], "id": source["id"],
                     "file_type": source["file_type"], "sheet_tabs": tab}],
        "filter": {"type": "condition", "condition_type": "and",
                   "expressions": [{"column": letters[header], "criteria": "text_is_exactly", "value": value}]},
        "destinations": [{"type": "file", "provider": "drive", "id": "<destination spreadsheet id>",
                          "file_type": "gsheets", "sheet_name": f"{value} only"}],
    }
    print(json.dumps(payload, indent=2))

Call MCP tools from Python

File mcp_client.py · tested against the live API by the smoke run before each release

import asyncio
import json
import os

import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

MCP_URL = os.environ.get("SHEETGO_MCP_URL", "https://dev-api.sheetgo.com/mcp")
API_KEY = os.environ["SHEETGO_API_KEY"]


class ToolError(Exception):
    def __init__(self, data):
        super().__init__(data.get("message"))
        self.data = data


def _payload(result):
    """The tool's JSON answer: structured content when the server sends it, else the text block."""
    if result.structuredContent is not None:
        return result.structuredContent.get("result", result.structuredContent)
    text = "".join(c.text for c in result.content if c.type == "text")
    try:
        return json.loads(text)
    except ValueError:
        return {"isError": result.isError, "message": text}


async def call(session, tool, attempts=4, **arguments):
    for attempt in range(attempts):
        try:
            result = await session.call_tool(tool, arguments)
        except Exception as exc:  # HTTP 429 from the per-key rate limit surfaces as an exception
            if attempt == attempts - 1 or "429" not in repr(exc):
                raise
            await asyncio.sleep(60)
            continue
        data = _payload(result)
        if not (result.isError or (isinstance(data, dict) and data.get("isError"))):
            return data
        if data.get("code") == 103:
            raise ToolError(data)  # Google access was revoked: the user must reconnect Google in Sheetgo
        if not data.get("isRetryable") or attempt == attempts - 1:
            raise ToolError(data)
        await asyncio.sleep(5 * (attempt + 1))


async def read_tab(session, file_id, tab, page_size=200):
    """(headers, rows) of a whole tab, following `cursor` until it comes back empty."""
    rows, cursor = [], None
    while True:
        page = await call(session, "sheets_get_values", file_id=file_id, tab=tab, max_rows=page_size, cursor=cursor)
        rows.extend(page["rows"])
        cursor = page["cursor"]
        if not cursor:
            return page["headers"], rows


async def main():
    http = httpx.AsyncClient(headers={"Authorization": f"Bearer {API_KEY}"}, timeout=httpx.Timeout(30, read=300))
    async with http, streamable_http_client(MCP_URL, http_client=http) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = (await session.list_tools()).tools
            print(f"{len(tools)} tools, signed in as", (await call(session, "whoami"))["email"])

            found = await call(session, "drive_search_files", query="Game Industry")
            sheet = next(f for f in found["files"] if f["name"] == "Game Industry")
            headers, rows = await read_tab(session, sheet["id"], "Hundred", page_size=40)
            print(f'{sheet["name"]}: {len(rows)} rows, columns {headers}')

            try:
                await call(session, "sheets_list_tabs", file_id="not-a-real-file-id")
            except ToolError as e:
                print("handled tool error:", e.data.get("kind"), "-", e)


if __name__ == "__main__":
    asyncio.run(main())

Give an AI agent the Sheetgo tools

File mcp_agent.py

import os
import sys

import anthropic

MCP_URL = os.environ.get("SHEETGO_MCP_URL", "https://dev-api.sheetgo.com/mcp")
SHEETGO_KEY = os.environ["SHEETGO_API_KEY"]

client = anthropic.Anthropic()
question = " ".join(sys.argv[1:]) or "Find my spreadsheet 'Game Industry' and tell me which platform sold most in tab 'Hundred'."
messages = [{"role": "user", "content": question}]

while True:
    response = client.beta.messages.create(
        model="claude-opus-5-5",
        max_tokens=16000,
        betas=["mcp-client-2025-11-20", "server-side-fallback-2026-07-01"],
        fallbacks="default",  # if the request is declined, the API retries it on a fallback model
        mcp_servers=[{"type": "url", "url": MCP_URL, "name": "sheetgo", "authorization_token": SHEETGO_KEY}],
        tools=[{"type": "mcp_toolset", "mcp_server_name": "sheetgo"}],
        messages=messages,
    )
    if response.stop_reason != "pause_turn":  # a long tool loop pauses; send it back to continue
        break
    messages.append({"role": "assistant", "content": response.content})

if response.stop_reason == "refusal":
    raise SystemExit("the request was declined")
for block in response.content:
    if block.type == "mcp_tool_use":
        print(f"[tool] {block.name} {block.input}")
    elif block.type == "mcp_tool_result" and block.is_error:
        print("[tool error]", block.content)  # same JSON as in the previous recipe: isError, message, ...
    elif block.type == "text":
        print(block.text)