Agent SQL Read-Only Guard (Python Generator)

import re

ALLOWED_OBJECTS = {name.strip().lower() for name in "vw_open_orders, vw_inventory_onhand, vw_supplier_scorecard".split(",") if name.strip()}
MAX_ROWS = 200
ROW_LIMIT_STYLE = "LIMIT"

BLOCKED_WORDS = re.compile(
    r"\b(insert|update|delete|drop|alter|truncate|create|grant|revoke|merge|exec|execute|call|into|xp_cmdshell|pg_sleep|waitfor)\b",
    re.IGNORECASE,
)
OBJECT_REFS = re.compile(r"\b(?:from|join)\s+([A-Za-z0-9_.\[\]\"]+)", re.IGNORECASE)


class SqlRejected(Exception):
    pass


def guard_sql(sql):
    text = sql.strip().rstrip(";").strip()

    if ";" in text:
        raise SqlRejected("Only one statement is allowed.")
    if "--" in text or "/*" in text:
        raise SqlRejected("Comments are not allowed.")
    if not re.match(r"^(select|with)\b", text, re.IGNORECASE):
        raise SqlRejected("Only SELECT queries are allowed.")
    if BLOCKED_WORDS.search(text):
        raise SqlRejected("The query contains a blocked keyword.")

    for ref in OBJECT_REFS.findall(text):
        name = ref.replace("[", "").replace("]", "").replace("\"", "").lower()
        name = name.split(".")[-1]
        if name not in ALLOWED_OBJECTS:
            raise SqlRejected("Object not on the allow list: " + name)

    # Wrap so the row cap applies even if the model forgets a limit
    if ROW_LIMIT_STYLE == "TOP":
        return "SELECT TOP " + str(MAX_ROWS) + " * FROM (" + text + ") AS guarded"
    if ROW_LIMIT_STYLE == "FETCH FIRST":
        return "SELECT * FROM (" + text + ") guarded FETCH FIRST " + str(MAX_ROWS) + " ROWS ONLY"
    return "SELECT * FROM (" + text + ") AS guarded LIMIT " + str(MAX_ROWS)

try:
    safe_sql = guard_sql(model_generated_sql)
except SqlRejected as problem:
    # Return the message to the model so it can correct the query
    result = "Query rejected: " + str(problem)
else:
    cursor.execute(safe_sql)  # connection must use a read-only database role