import os, json, time, base64, threading, logging
from email.mime.text import MIMEText
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import anthropic, psycopg2, psycopg2.extras
from flask import Flask, request, render_template_string, abort
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.send"]
log = logging.getLogger("inbox-bot")
app = Flask(__name__)
def gmail_service():
creds = Credentials.from_authorized_user_file("token.json", SCOPES) if os.path.exists("token.json") else None
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token: creds.refresh(Request())
else: raise RuntimeError("run the oauth flow first")
open("token.json", "w").write(creds.to_json())
return build("gmail", "v1", credentials=creds, cache_discovery=False)
gmail, llm = gmail_service(), anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
db = psycopg2.connect(os.environ["DATABASE_URL"], cursor_factory=psycopg2.extras.RealDictCursor)
def decode_body(payload):
if payload.get("body", {}).get("data"): return base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", "replace")
for part in payload.get("parts", []) or []:
if part.get("mimeType") == "text/plain" and part["body"].get("data"): return decode_body(part)
for part in payload.get("parts", []) or []:
found = decode_body(part)
if found: return found
return ""
def draft_reply(text, attempt=0):
try:
r = llm.messages.create(model="claude-sonnet-5-5", max_tokens=1024, messages=[{"role": "user", "content": text}])
return "".join(b.text for b in r.content if b.type == "text")
except (anthropic.RateLimitError, anthropic.APIConnectionError) as e:
if attempt >= 5: raise
time.sleep(2 ** attempt); return draft_reply(text, attempt + 1)
def poll_forever():
seen = {r["msg_id"] for r in query("SELECT msg_id FROM reviews")}
while True:
try:
res = gmail.users().messages().list(userId="me", q="label:inbox -from:me", maxResults=50).execute()
except HttpError as e:
log.warning("gmail list failed: %s", e); time.sleep(30); continue
for m in res.get("messages", []):
if m["id"] in seen: continue
msg = gmail.users().messages().get(userId="me", id=m["id"], format="full").execute()
headers = {h["name"].lower(): h["value"] for h in msg["payload"]["headers"]}
draft = draft_reply(decode_body(msg["payload"]))
execute("INSERT INTO reviews (msg_id, thread_id, sender, message_id, draft, status) VALUES (%s,%s,%s,%s,%s,'pending')",
(m["id"], msg["threadId"], headers.get("from"), headers.get("message-id"), draft))
seen.add(m["id"]); notify_reviewer(m["id"])
time.sleep(60)
@app.route("/review/<msg_id>", methods=["GET", "POST"])
def review(msg_id):
row = one("SELECT * FROM reviews WHERE msg_id = %s", (msg_id,)) or abort(404)
if request.method == "GET": return render_template_string(REVIEW_PAGE, row=row)
if request.form.get("decision") != "send": return mark(msg_id, "rejected")
reply = MIMEText(request.form["reply"])
reply["To"], reply["In-Reply-To"], reply["References"] = row["sender"], row["message_id"], row["message_id"]
raw = base64.urlsafe_b64encode(reply.as_bytes()).decode()
gmail.users().messages().send(userId="me", body={"raw": raw, "threadId": row["thread_id"]}).execute()
...
...
import os, json, time, base64, threading, logging
from email.mime.text import MIMEText
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import anthropic, psycopg2, psycopg2.extras
from flask import Flask, request, render_template_string, abort
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.send"]
log = logging.getLogger("inbox-bot")
app = Flask(__name__)
def gmail_service():
creds = Credentials.from_authorized_user_file("token.json", SCOPES) if os.path.exists("token.json") else None
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token: creds.refresh(Request())
else: raise RuntimeError("run the oauth flow first")
open("token.json", "w").write(creds.to_json())
return build("gmail", "v1", credentials=creds, cache_discovery=False)
gmail, llm = gmail_service(), anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
db = psycopg2.connect(os.environ["DATABASE_URL"], cursor_factory=psycopg2.extras.RealDictCursor)
def decode_body(payload):
if payload.get("body", {}).get("data"): return base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", "replace")
for part in payload.get("parts", []) or []:
if part.get("mimeType") == "text/plain" and part["body"].get("data"): return decode_body(part)
for part in payload.get("parts", []) or []:
found = decode_body(part)
if found: return found
return ""
def draft_reply(text, attempt=0):
try:
r = llm.messages.create(model="claude-sonnet-5-5", max_tokens=1024, messages=[{"role": "user", "content": text}])
return "".join(b.text for b in r.content if b.type == "text")
except (anthropic.RateLimitError, anthropic.APIConnectionError) as e:
if attempt >= 5: raise
time.sleep(2 ** attempt); return draft_reply(text, attempt + 1)
def poll_forever():
seen = {r["msg_id"] for r in query("SELECT msg_id FROM reviews")}
while True:
try:
res = gmail.users().messages().list(userId="me", q="label:inbox -from:me", maxResults=50).execute()
except HttpError as e:
log.warning("gmail list failed: %s", e); time.sleep(30); continue
for m in res.get("messages", []):
if m["id"] in seen: continue
msg = gmail.users().messages().get(userId="me", id=m["id"], format="full").execute()
headers = {h["name"].lower(): h["value"] for h in msg["payload"]["headers"]}
draft = draft_reply(decode_body(msg["payload"]))
execute("INSERT INTO reviews (msg_id, thread_id, sender, message_id, draft, status) VALUES (%s,%s,%s,%s,%s,'pending')",
(m["id"], msg["threadId"], headers.get("from"), headers.get("message-id"), draft))
seen.add(m["id"]); notify_reviewer(m["id"])
time.sleep(60)
@app.route("/review/<msg_id>", methods=["GET", "POST"])
def review(msg_id):
row = one("SELECT * FROM reviews WHERE msg_id = %s", (msg_id,)) or abort(404)
if request.method == "GET": return render_template_string(REVIEW_PAGE, row=row)
if request.form.get("decision") != "send": return mark(msg_id, "rejected")
reply = MIMEText(request.form["reply"])
reply["To"], reply["In-Reply-To"], reply["References"] = row["sender"], row["message_id"], row["message_id"]
raw = base64.urlsafe_b64encode(reply.as_bytes()).decode()
gmail.users().messages().send(userId="me", body={"raw": raw, "threadId": row["thread_id"]}).execute()
...
...