AI Agent for Google Reviews: Python Reply Script

Get Access First

1. A Google Cloud project.
2. A verified Business Profile, yours or a client's, that has been active for 60 or more days, with a website listed on it.
3. A submitted "Application for Basic API Access" through Google's contact form, with your project number, from an email address that owns or manages the profile.
4. After approval, the Business Profile APIs enabled for the project.

How the Script Works

Refresh OAuth access token
        |
GET reviews (50 per page, newest first)
        |
Skip any review that already has a reviewReply
        |
Stop words (refund, lawyer, injured ...) -> flag for the owner, no draft
        |
Draft a reply with the LLM (model may answer HOLD -> flag for the owner)
        |
1-3 stars, or DRY_RUN on -> print the draft, post nothing
        |
4-5 stars and DRY_RUN off -> PUT the reply, log the review

• Dry run is the default. Nothing posts until you set DRY_RUN=0.
• Only 4 to 5 star reviews can ever be auto-posted. Everything lower is draft-only.
• The model can say HOLD. The prompt tells it to output HOLD for health, money, legal or safety content, and the script treats that as "owner decides".
• No generic fallback reply. If the model fails, the script skips the review rather than posting boilerplate.
• A log file prevents double replies.

Find Your Account and Location IDs

# 1. List accounts (the "name" looks like accounts/123...)
curl -H "Authorization: Bearer ACCESS_TOKEN" \
  "https://mybusinessaccountmanagement.googleapis.com/v1/accounts"

# 2. List that account's locations (the "name" looks like locations/456...)
curl -H "Authorization: Bearer ACCESS_TOKEN" \
  "https://mybusinessbusinessinformation.googleapis.com/v1/accounts/ACCOUNT_ID/locations?readMask=name,title"

The Script

#!/usr/bin/env python3
"""Draft (and optionally post) replies to Google reviews.
UNTESTED STARTING POINT. Run with DRY_RUN=1 first and read every draft."""

import json
import os
import re
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

GROQ_API_KEY = os.environ["GROQ_API_KEY"]
GROQ_MODEL = os.environ.get("GROQ_MODEL", "llama-3.3-70b-versatile")
CLIENT_ID = os.environ["GOOGLE_CLIENT_ID"]
CLIENT_SECRET = os.environ["GOOGLE_CLIENT_SECRET"]
REFRESH_TOKEN = os.environ["GOOGLE_REFRESH_TOKEN"]
LOCATION_NAME = os.environ["LOCATION_NAME"]  # accounts/ACCOUNT_ID/locations/LOCATION_ID
BUSINESS_NAME = os.environ.get("BUSINESS_NAME", "our business")
OWNER_NAME = os.environ.get("OWNER_NAME", "The team")
DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"  # anything but "0" means dry run

LOG_FILE = Path("replied_reviews.json")
MAX_PAGES = 3  # 3 pages x 50 reviews; raise if you have a backlog
STARS = {"ONE": 1, "TWO": 2, "THREE": 3, "FOUR": 4, "FIVE": 5}
STOP_WORDS = re.compile(
    r"refund|lawyer|attorney|sue|injur|sick|poison|police|discriminat|allerg",
    re.IGNORECASE,
)


def call(url, method="GET", headers=None, body=None, timeout=20):
    data = body if isinstance(body, bytes) else (json.dumps(body).encode() if body else None)
    req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read()
    return json.loads(raw) if raw else {}


def get_access_token():
    body = urllib.parse.urlencode({
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "refresh_token": REFRESH_TOKEN,
        "grant_type": "refresh_token",
    }).encode()
    resp = call("https://oauth2.googleapis.com/token", "POST",
                {"Content-Type": "application/x-www-form-urlencoded"}, body)
    return resp["access_token"]


def fetch_unanswered(token):
    reviews, page_token = [], None
    for _ in range(MAX_PAGES):
        params = {"pageSize": 50}
        if page_token:
            params["pageToken"] = page_token
        url = f"https://mybusiness.googleapis.com/v4/{LOCATION_NAME}/reviews?" + urllib.parse.urlencode(params)
        data = call(url, headers={"Authorization": f"Bearer {token}"})
        reviews += data.get("reviews", [])
        page_token = data.get("nextPageToken")
        if not page_token:
            break
    return [r for r in reviews if "reviewReply" not in r]


def draft_reply(review):
    reviewer = review.get("reviewer", {})
    first_name = ""
    if not reviewer.get("isAnonymous") and reviewer.get("displayName"):
        first_name = reviewer["displayName"].split()[0]
    stars = STARS.get(review.get("starRating"), 0)
    prompt = (
        f"You write Google review replies for {BUSINESS_NAME}. Tone: warm, brief, human.\n"
        f"Rules: use the reviewer's first name if given, mention one specific detail from the review, "
        f"under 90 words, no discounts, no admitting legal fault, no private details, "
        f"sign off as {OWNER_NAME}. For 1-3 stars, thank them, apologise that the visit fell short "
        f"and invite them to contact us directly. If the review mentions health, safety, money or "
        f"legal action, output exactly HOLD. Output only the reply text.\n\n"
        f"First name: {first_name or '(not given)'}\nStars: {stars}\n"
        f"Review: {review.get('comment') or '(no text)'}"
    )
    payload = {
        "model": GROQ_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "max_completion_tokens": 250,
        "temperature": 0.6,
    }
    try:
        resp = call("https://api.groq.com/openai/v1/chat/completions", "POST",
                    {"Authorization": f"Bearer {GROQ_API_KEY}", "Content-Type": "application/json"},
                    payload)
        text = resp["choices"][0]["message"]["content"].strip()
    except Exception as exc:  # no generic fallback: skip instead
        print(f"  model call failed, skipping: {exc}")
        return None
    if text.upper().startswith("HOLD") or len(text.encode()) > 4096:
        return None
    return text


def post_reply(token, review_name, text):
    url = f"https://mybusiness.googleapis.com/v4/{review_name}/reply"
    call(url, "PUT", {"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
         {"comment": text})


def main():
    log = json.loads(LOG_FILE.read_text()) if LOG_FILE.exists() else {}
    token = get_access_token()
    reviews = fetch_unanswered(token)
    print(f"{len(reviews)} unanswered review(s). DRY_RUN={DRY_RUN}")

    for review in reviews:
        name = review["name"]
        if name in log:
            continue
        stars = STARS.get(review.get("starRating"), 0)
        comment = review.get("comment", "")
        print(f"\n{stars} stars: {comment[:120]!r}")

        if STOP_WORDS.search(comment):
            print("  FLAGGED: stop word found. Owner replies personally.")
            continue
        text = draft_reply(review)
        if text is None:
            print("  No draft (HOLD or error). Owner decides.")
            continue
        print(f"  DRAFT: {text}")

        if DRY_RUN or stars < 4:
            print("  Draft only. Review and post manually.")
            continue
        try:
            post_reply(token, name, text)
        except Exception as exc:
            print(f"  post failed: {exc}")
            continue
        log[name] = {"at": datetime.now(timezone.utc).isoformat(), "stars": stars}
        print("  posted")

    LOG_FILE.write_text(json.dumps(log, indent=2))


if __name__ == "__main__":
    main()

• Moderation. Google's reply help page says each reply is checked against content policies before it posts, usually in about 10 minutes but sometimes up to 30 days, and that a rejected reply must be edited. The review resource includes a reviewReplyState of PENDING, APPROVED or REJECTED and a policyViolation field when a reply is rejected. The script does not read these yet; adding a check is a sensible next step.
• Error handling. It stops on the first Google error when fetching and does not retry. Add exponential backoff for 429 responses, as Google recommends.
• Scheduling. Run it once a day from cron or a scheduled GitHub Action, not in a tight loop.
• Stop words. The list above is a starting point. Add words that matter in your trade.

Example Replies (Illustrative)

The Math

Do This Today

1. Check whether your Business Profile is verified, has a website listed and is at least 60 days old.
2. Create a Google Cloud project, submit the "Application for Basic API Access", and note the project number you used.
3. Count your unanswered reviews from the last 90 days. That is your volume and your backlog.
4. Write a voice guide: three replies you are proud of and three phrases you never use. Paste them into the prompt.
5. When access arrives, run the script with DRY_RUN=1 for a week and compare each draft with what you would have written. Only then consider DRY_RUN=0, and keep 1 to 3 star reviews manual.

FAQ

Sources

• Google Business Profile APIs: Prerequisites
• Google Business Profile APIs: Basic setup
• Google Business Profile APIs: Review data
• Google Business Profile APIs: Usage limits
• REST reference: reviews.list (v4)
• REST reference: reviews resource (v4)
• REST reference: reviews.updateReply (v4)
• REST reference: Business Information locations.list
• Google Identity: OAuth 2.0 for installed apps, refreshing tokens
• Google Business Profile Help: Reply to reviews
• Groq: API reference
• Groq: Models