Building Production-Ready SaaS Data Connectors & AI Safety Hooks in Python (With Checkpoint Pagination)
Building reliable SaaS data connectors and secure AI agent tooling requires two critical components: robust checkpoint-based pagination for incremental syncs and strict pre-tool execution safety hooks. In this post, we e
Building reliable SaaS data connectors and secure AI agent tooling requires two critical components: robust checkpoint-based pagination for incremental syncs and strict pre-tool execution safety hooks. In this post, we explore how to implement both in Python.
1. SaaS REST Connector with Checkpoint Pagination & Exponential Backoff
When syncing data from APIs (like Jira, GitHub, or Salesforce), handling pagination and transient rate limits gracefully is essential. Here is a production-grade Python pattern using Pydantic v2 for state serialization:
import time
from typing import Dict, Any, Generator, Optional
from pydantic import BaseModel, Field
class ConnectorState(BaseModel):
cursor: Optional[str] = None
offset: int = 0
last_sync_timestamp: float = 0.0
class SaaSRESTConnector:
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.state = ConnectorState()
def fetch_page_with_backoff(self, endpoint: str, params: Dict[str, Any], max_retries: int = 5) -> Dict[str, Any]:
"""Fetch API pages with exponential backoff on failure."""
for attempt in range(max_retries):
try:
# Simulated REST request
return {"items": [{"id": 1}], "next_cursor": None}
except Exception as e:
if attempt == max_retries - 1:
raise e
time.sleep(2 ** attempt)
def load_from_state(self, state_dict: Dict[str, Any]):
"""Resume connector from saved checkpoint state."""
self.state = ConnectorState(**state_dict)
def stream_documents(self) -> Generator[Dict[str, Any], None, None]:
"""Stream documents iteratively while updating checkpoint state."""
while True:
params = {"cursor": self.state.cursor, "offset": self.state.offset}
data = self.fetch_page_with_backoff("/documents", params)
for item in data.get("items", []):
yield item
self.state.cursor = data.get("next_cursor")
if not self.state.cursor:
break
2. AI Agent Safety Hook (PreToolUse Destructive Bash Blocker)
When deploying AI coding agents (like Claude Code), blocking dangerous bash commands before execution prevents accidental system or data loss:
import sys
import json
import logging
DESTRUCTIVE_PATTERNS = [
"rm -rf", "drop database", "git reset --hard",
"mkfs", "dd if=", "chmod -r 777"
]
def audit_and_block(command: str) -> bool:
for pattern in DESTRUCTIVE_PATTERNS:
if pattern in command.lower():
logging.error(f"BLOCKED DESTRUCTIVE BASH COMMAND: {command}")
return True
return False
if __name__ == "__main__":
tool_input = json.loads(sys.stdin.read()) if not sys.stdin.isatty() else {}
cmd = tool_input.get("command", "")
if audit_and_block(cmd):
print("Error: Destructive command blocked by safety hook.", file=sys.stderr)
sys.exit(1)
sys.exit(0)
Production Developer Kit & Resources
Get the full boilerplate, hooks, test suites, and template documentation:
- Monetized Developer Kit: Production Python SaaS REST Connector & AI Agent Safety Kit on Gumroad
- GitHub Profile: Rish Anu on GitHub
Originally published by Dev.to WebDev. Aggregated on AIWithGhost for educational purposes β full credit and traffic to the original publisher.