Overview
Wraps agent sessions with a credit-tracking layer that logs search API spend per run, aggregates daily totals, and sends alerts when a session or daily budget threshold is breached.
Trigger
After each agent run completes (post-run hook or wrapper function)
Schedule
After each agent run (event-driven, not scheduled)
Workflow Steps
Initialize session credit tracker
Create a session object with start timestamp, session_id, and credit_count = 0 before the agent starts.
Intercept search tool calls
Wrap the search API call function to increment credit_count by 1 (or by the actual credits_used field from the API response) on each call.
Log session summary to database
After the agent run completes (success or failure), write session_id, credits_used, timestamp, agent_type, and task_description to a SQLite or PostgreSQL sessions table.
Check against thresholds
Compare session credits_used against per-session limit (e.g., 50 credits) and daily total against daily limit (e.g., 500 credits). Query daily aggregate from the sessions table.
Fire alert if threshold breached
Send a webhook or email alert with session details, credits used, and remaining daily budget when either threshold is exceeded.
Python Implementation
import sqlite3
import time
import requests
from datetime import date
from functools import wraps
DB_PATH = "agent_costs.db"
SCRAPING_API_BASE = "https://api.scavio.dev/api/v1/search"
SCRAVIO_API_KEY = "YOUR_API_KEY"
SESSION_LIMIT = 50 # credits per session
DAILY_LIMIT = 500 # credits per day
ALERT_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
credits_used INTEGER,
agent_type TEXT,
task TEXT,
ts TEXT
)
""")
conn.commit()
return conn
class SearchCostTracker:
def __init__(self, session_id: str, agent_type: str, task: str):
self.session_id = session_id
self.agent_type = agent_type
self.task = task
self.credits_used = 0
self.conn = init_db()
def search(self, query: str, platform: str = "google") -> dict:
if self.credits_used >= SESSION_LIMIT:
raise RuntimeError(f"Session credit limit {SESSION_LIMIT} reached")
resp = requests.post(
SCRAPING_API_BASE,
headers={"x-api-key": SCRAVIO_API_KEY},
json={"query": query, "platform": platform}
)
resp.raise_for_status()
self.credits_used += 1
return resp.json()
def finalize(self):
today = date.today().isoformat()
self.conn.execute(
"INSERT INTO sessions VALUES (?, ?, ?, ?, ?)",
(self.session_id, self.credits_used, self.agent_type, self.task, today)
)
self.conn.commit()
# Check daily total
row = self.conn.execute(
"SELECT SUM(credits_used) FROM sessions WHERE ts = ?", (today,)
).fetchone()
daily_total = row[0] or 0
if self.credits_used > SESSION_LIMIT * 0.8 or daily_total > DAILY_LIMIT:
requests.post(ALERT_WEBHOOK, json={
"text": f"Search cost alert: session={self.credits_used} credits, daily={daily_total}/{DAILY_LIMIT}"
})
# Usage
tracker = SearchCostTracker("sess_001", "research_agent", "competitor pricing")
try:
result = tracker.search("competitor product pricing 2026")
# ... agent uses result ...
finally:
tracker.finalize()
JavaScript Implementation
const Database = require('better-sqlite3');
const fetch = require('node-fetch');
const DB_PATH = 'agent_costs.db';
const API_BASE = 'https://api.scavio.dev/api/v1/search';
const API_KEY = 'YOUR_API_KEY';
const SESSION_LIMIT = 50;
const DAILY_LIMIT = 500;
const ALERT_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK';
const db = new Database(DB_PATH);
db.prepare(`CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY, credits_used INTEGER,
agent_type TEXT, task TEXT, ts TEXT
)`).run();
class SearchCostTracker {
constructor(sessionId, agentType, task) {
this.sessionId = sessionId;
this.agentType = agentType;
this.task = task;
this.creditsUsed = 0;
}
async search(query, platform = 'google') {
if (this.creditsUsed >= SESSION_LIMIT) throw new Error('Session credit limit reached');
const res = await fetch(API_BASE, {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, platform })
});
if (!res.ok) throw new Error(`Search failed: ${res.status}`);
this.creditsUsed++;
return res.json();
}
async finalize() {
const today = new Date().toISOString().slice(0, 10);
db.prepare('INSERT INTO sessions VALUES (?,?,?,?,?)').run(
this.sessionId, this.creditsUsed, this.agentType, this.task, today
);
const { dailyTotal } = db.prepare(
'SELECT SUM(credits_used) as dailyTotal FROM sessions WHERE ts = ?'
).get(today);
if (this.creditsUsed > SESSION_LIMIT * 0.8 || dailyTotal > DAILY_LIMIT) {
await fetch(ALERT_WEBHOOK, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text: `Search cost alert: session=${this.creditsUsed}, daily=${dailyTotal}/${DAILY_LIMIT}` })
});
}
}
}
// Usage
const tracker = new SearchCostTracker('sess_001', 'research_agent', 'competitor pricing');
(async () => {
try {
const result = await tracker.search('competitor product pricing 2026');
console.log(result);
} finally {
await tracker.finalize();
}
})();
Platforms Used
Web search with knowledge graph, PAA, and AI overviews