Blog / AI agents

Build a Price Tracking Agent in Python with Live Data

17 Sept 2026Perceptlayer8 min read

A price tracking agent in Python is a loop that reads a watchlist, asks two live sources for the current lowest price, saves each reading and alerts you when the price falls. This tutorial builds one with the Google Shopping and eBay endpoints, a single Python file and SQLite. Each product costs 2 credits per run, so the monthly bill is a multiplication you can do before you start.

What does a price tracking agent have to do?

A tracker has four jobs: fetch the current price for each product, store it with a timestamp, compare it with history, and tell you when something changed. Fetching is the only part that needs live web data. Everything else is a small database and a few lines of arithmetic, which is why this build uses no framework and no model call on the hot path.

We use two sources because they answer different questions. Google Shopping returns offers from many retailers for a query, so it tells you the going retail price. eBay search returns individual listings, so it shows the marketplace price.

The result is a script you can run from cron. Because the loop is deterministic, it is cheap, easy to test, and it never invents a price.

How do you build the tracker?

You build it in seven steps: get a key, write a watchlist, add a small API client, write one fetch function per source, store readings, detect drops, then run it on a schedule. The code below is one file, tracker.py. Each step adds to it, and the pieces concatenate in order into a complete script.

Step 1: Get a key and install two packages

Create a free account at /signup to get a key that starts with sk_live_. Free accounts get 1,000 credits every month after email verification, with no credit card. Put the key in an environment variable, as the authentication docs describe.

export API_KEY="sk_live_..."
pip install requests

SQLite ships with Python, so requests is the only dependency. See the Requests documentation if you want to tune sessions or timeouts.

Step 2: Write the watchlist

Search is fuzzy, so each product carries its own matching rules. must_include words must all appear in a result title, exclude words drop accessories, and min_price removes cases, chargers and parts that match the query by accident. target is the price at which you want an alert regardless of history.

Save this as watchlist.json:

[
  {
    "name": "iPhone 15 128GB",
    "google_q": "iphone 15 128gb",
    "ebay_q": "iphone 15 128gb unlocked",
    "must_include": ["iphone 15", "128"],
    "exclude": ["case", "cover", "protector", "charger", "cable", "pro", "plus", "max"],
    "min_price": 300,
    "target": 400
  },
  {
    "name": "Keurig K-Cafe",
    "google_q": "keurig k-cafe coffee maker",
    "ebay_q": "keurig k-cafe coffee maker",
    "must_include": ["k-cafe"],
    "exclude": ["pods", "filter", "descaler", "parts"],
    "min_price": 60,
    "target": 120
  }
]

These two products are examples. Replace them with yours, and expect to adjust the filters after the first run.

Step 3: Add an API client that respects the limits

The client sends the key in the X-API-Key header, waits when the API says to, and records the credit balance from every response. Rate-limited (429) and temporarily unavailable (503) responses carry a Retry-After header with the seconds to wait, and they are failed calls, so retrying them costs nothing extra. See rate limits for how the per-minute counting works.

import json
import os
import sqlite3
import time
from datetime import datetime, timezone

import requests

BASE = "https://api.perceptlayer.com"
DROP = 0.05            # alert when 5% below the recent average
HISTORY = 5            # readings that make up the average
PAUSE = float(os.environ.get("PAUSE", "1.0"))        # seconds between calls
CREDIT_FLOOR = int(os.environ.get("CREDIT_FLOOR", "50"))

session = requests.Session()
session.headers["X-API-Key"] = os.environ["API_KEY"]
state = {"remaining": None}


def call(path, params, tries=3):
    for _ in range(tries):
        r = session.get(BASE + path, params=params, timeout=60)
        if r.status_code in (429, 503):
            time.sleep(int(r.headers.get("Retry-After", "5")))
            continue
        r.raise_for_status()
        state["remaining"] = int(r.headers.get("X-Credits-Remaining", 0))
        return r.json()
    raise RuntimeError(f"{path}: still limited after {tries} tries")

The default PAUSE of one second keeps a free account, which gets 1 request a second, under its limit. Paying accounts run at least 5 requests a second, so you can lower it.

Step 4: Fetch the lowest matching price from each source

Each fetch function calls one endpoint, filters the results with the watchlist rules and returns the cheapest survivor as (price, title, link). Google Shopping results sit in shopping_results with a numeric extracted_price. The eBay search response nests listings under data.results, each with a price object.

Here is an excerpt of a recorded response from GET /v1/ebay/search?q=coffee+maker&page=1&per_page=60, trimmed with …:

{
  "success": true,
  "data": {
    "query": "coffee maker",
    "page": 1,
    "per_page": 60,
    "has_more": true,
    "results": [
      {
        "product_id": "366702966104",
        "title": "Mr. Coffee 4-in-1 Single-Serve Latte, Iced, and Hot Coffee",
        "link": "https://www.ebay.com/itm/366702966104",
        "price": {
          "raw": "$60.00",
          "extracted": 60,
          "currency": "USD"
        },
        "sponsored": true,
        …
      },
      …
    ],
    …
  },
  …
}

The tracker reads title, link, price.extracted and price.currency. It also passes the optional boolean buy_it_now parameter, because a Buy It Now price is easier to compare over time than a bid. Confirm exactly what the filter does on the eBay search endpoint page, which lists every parameter, including price_min and free_shipping, and spot-check the titles in your first runs.

def matches(title, item):
    t = title.lower()
    return all(w.lower() in t for w in item["must_include"]) and not any(
        w.lower() in t for w in item.get("exclude", [])
    )


def cheapest(rows, item):
    best = None
    for title, price, link in rows:
        if price is None or price < item["min_price"] or not matches(title, item):
            continue
        if best is None or price < best[0]:
            best = (price, title, link)
    return best


def google_lowest(item):
    data = call(
        "/v1/google-shopping/search",
        {"q": item["google_q"], "gl": "us", "hl": "en"},
    )
    rows = [
        (r.get("title", ""), r.get("extracted_price"), r.get("link"))
        for r in data.get("shopping_results", [])
    ]
    return cheapest(rows, item)


def ebay_lowest(item):
    data = call(
        "/v1/ebay/search",
        {
            "q": item["ebay_q"],
            "buy_it_now": "true",
            "price_min": item["min_price"],
            "per_page": 60,
        },
    )
    rows = []
    for r in data["data"]["results"]:
        price = r.get("price") or {}
        if price.get("currency") == "USD":
            rows.append((r.get("title", ""), price.get("extracted"), r.get("link")))
    return cheapest(rows, item)

Both functions cost one credit when they succeed. The Google Shopping call passes gl=us and hl=en, the same values as the recorded example for that endpoint. Both are optional two-character parameters. This tutorial only tests with those values, so if you change them, check the prices you get back before trusting the history. The eBay function keeps only USD listings, so change that check if you track in another currency.

Step 5: Store readings and detect drops

Every reading goes into a prices table with a UTC timestamp, so history survives restarts. A drop is a price at least 5% below the mean of the previous five readings for the same product and source, or any price at or below your target. Comparing against an average rather than the last reading catches slow slides that a one-step comparison misses, and it ignores a single noisy listing.

def open_db(path="prices.db"):
    db = sqlite3.connect(path)
    db.execute(
        "CREATE TABLE IF NOT EXISTS prices ("
        "item TEXT, source TEXT, ts TEXT, price REAL, title TEXT, link TEXT)"
    )
    return db


def record_and_check(db, item, source, hit):
    price, title, link = hit
    prev = [
        row[0]
        for row in db.execute(
            "SELECT price FROM prices WHERE item = ? AND source = ? "
            "ORDER BY ts DESC LIMIT ?",
            (item["name"], source, HISTORY),
        )
    ]
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    db.execute(
        "INSERT INTO prices VALUES (?, ?, ?, ?, ?, ?)",
        (item["name"], source, now, price, title, link),
    )
    db.commit()

    reasons = []
    if prev:
        avg = sum(prev) / len(prev)
        if price <= avg * (1 - DROP):
            reasons.append(f"{(1 - price / avg):.0%} below the recent average of {avg:.2f}")
    if price <= item["target"]:
        reasons.append(f"at or below your target of {item['target']}")
    return reasons

The sqlite3 module documentation covers the ? placeholders used here. Using them instead of string formatting keeps product names from breaking the query.

Step 6: Run the loop with a spending guard

The main function walks the watchlist, calls both sources per product, and prints an alert for every drop. It stops early in two cases: the balance in X-Credits-Remaining falls below CREDIT_FLOOR, or the API returns insufficient_credits (402). The service is prepaid and there is no automatic top-up, so running out stops the tracker rather than billing you further.

def alert(item, source, hit, reasons):
    price, title, link = hit
    print(f"DROP {item['name']} [{source}] {price:.2f}: {'; '.join(reasons)}")
    print(f"     {title}\n     {link}")
    # Send to Slack, email or a webhook here.


def run():
    with open("watchlist.json") as f:
        items = json.load(f)
    db = open_db()
    for item in items:
        for source, fetch in (("google-shopping", google_lowest), ("ebay", ebay_lowest)):
            if state["remaining"] is not None and state["remaining"] < CREDIT_FLOOR:
                print(f"Stopping: {state['remaining']} credits left")
                return
            try:
                hit = fetch(item)
            except requests.Timeout:
                # The call may have succeeded and been billed; skip it this run.
                print(f"timeout: {item['name']} [{source}]")
                continue
            except requests.HTTPError as e:
                body = e.response.json().get("error", {})
                print(f"{e.response.status_code} {body.get('code')}: {item['name']} [{source}]")
                if e.response.status_code == 402:
                    return
                continue
            finally:
                time.sleep(PAUSE)
            if hit is None:
                print(f"no match: {item['name']} [{source}]")
                continue
            reasons = record_and_check(db, item, source, hit)
            if reasons:
                alert(item, source, hit, reasons)
    print(f"Done. Credits remaining: {state['remaining']}")


if __name__ == "__main__":
    run()

Run it once with python tracker.py. The first run only records a baseline, so alerts on that run come only from your target prices. Drops against history start on the second run. For the error codes the handler prints, see the errors reference.

Step 7: Schedule it

Run the script from cron, a systemd timer or any scheduler. This crontab entry runs it every six hours, which is four checks a day:

0 */6 * * * cd /opt/tracker && API_KEY=sk_live_... python tracker.py >> tracker.log 2>&1

Keep the working directory fixed, because watchlist.json and prices.db are read relative to it. A scheduler that overlaps two runs would double the spend, so keep the interval longer than a run takes.

How many credits will it use a month?

Each product costs one Google Shopping call and one eBay call per run, so credits a month equal products × 2 × runs a day × 30. Failed calls cost nothing, so retries after a 429 or 503 do not add to the total. The figures below are credits only; pricing shows what credits cost and the credits docs explain how debits work.

Purchased credits last 12 months, so a pack that outlives one month of tracking is not wasted. Two levers cut the bill. Check slow-moving products less often, and drop a source for products where one source gives no matches. If you want a hard ceiling, the pattern in AI agent API cost control shows how to cap spend per run, and the CREDIT_FLOOR guard above is the simplest version of it.

How do you keep a price tracker reliable?

Three problems cause most bad alerts: a wrong match, a duplicate charge and a silent stop. A wrong match is a filter problem, fixed by tightening must_include, exclude and min_price after reading what the first runs matched. A duplicate charge comes from retrying a call whose reply was lost. A silent stop comes from an empty balance or a scheduler that stopped running.

Requests have no idempotency key. A request whose reply was lost, such as a client timeout, may have succeeded and been billed. That is why the script skips a timed-out check instead of repeating it; one missed reading costs less than a doubled one. AI agent API cost control covers spend limits that work alongside this.

For a silent stop, check the last line of tracker.log or alert when no new rows land in prices for a day. Data is returned as found at the time of the call, so a price you read is a price that was listed then, not a guarantee that the item is in stock now. Treat an alert as a prompt to open the link.

Where does an LLM fit in a price tracking agent?

The scheduled loop does not need a model, and leaving one out keeps each run at exactly 2 credits per product with no token cost. A model earns its place at the edges: turning "watch the cheapest 128GB iPhone 15 under 400" into a watchlist entry, or explaining a drop in plain language when you ask for it.

For that, give the model the same data over MCP. The Perceptlayer MCP server at https://api.perceptlayer.com/mcp exposes search_endpoints, get_endpoint and call_endpoint; the first two only look up the catalogue and cost nothing, while call_endpoint spends credits on success only. The AI agents docs show the setup, and MCP vs Function Calling for Live Web Data in Agents helps you pick between that and a tool you write yourself. For a Python tool wrapper, see the LangChain custom tool tutorial.

Next step: get a free API key, run one call from each endpoint in the playground on the Google Shopping and eBay pages, then export your key as API_KEY and run tracker.py on your own watchlist.

price trackingpythonai agentsgoogle shoppingebay