Competitor Monitoring AI Agent with Python Jobs

A useful competitor monitoring AI agent does not browse the entire web and produce a vague weekly summary. It watches approved sources, records what it observed, compares the result with the last successful run, and reports only meaningful changes.

This guide builds that workflow as a scheduled Python job. The agent uses an explicit URL allowlist, preserves snapshots, handles partial failures, and attaches a proof URL to every reported observation.

Define the agent’s scope, sources, and measurable alerts

Competitive data is fragmented across pricing pages, press releases, review sites, social feeds, and public filings. Different teams also care about different signals. Sales may want positioning changes. Product may track feature launches. Marketing may follow campaigns or customer sentiment.

Start with an input contract rather than a crawler.

For each monitored source, define:

• The competitor.
• The exact URL or approved URL pattern.
• The source type.
• The fields you expect to extract.
• The collection cadence.
• The alert condition.
• The source owner.
• Whether the page contains competitor claims or independent evidence.

A minimal configuration can live in YAML:

sources:
  - source_id: acme_pricing
    competitor: Acme
    url: https://example.com/pricing
    type: html
    fields:
      - plans
      - prices
    cadence: frequent
    alert_on:
      - price_changed
      - plan_removed

  - source_id: acme_news
    competitor: Acme
    url: https://example.com/news.xml
    type: rss
    fields:
      - title
      - published_at
      - link
    cadence: daily
    alert_on:
      - new_item

Make the objective testable. “Monitor pricing” is ambiguous. “Detect a specified competitor price change within 30 minutes” gives you a target that can be measured, an example also recommended in Datagrid’s competitor tracking guide.

The same principle applies to other signals:

• Detect a new product announcement.
• Flag a removed feature.
• Identify a changed positioning headline.
• Record new customer-review themes without treating them as verified product facts.

Do not collect a source merely because it might be useful later. Every source adds failure modes, storage, noise, and operational cost. If you need a broader introduction to scheduled workflows, read the developer’s guide to Python AI agent automation.

Build a Python architecture around an explicit URL allowlist

Use a narrow pipeline:

scheduler
  -> allowlist validator
  -> fetcher
  -> normalizer
  -> snapshot store
  -> diff engine
  -> summarizer
  -> evidence log

Some commercial competitive analysts search competitor sites, pricing pages, comparison pages, search results, LinkedIn, and third-party sources. This implementation intentionally does less. It only fetches sources you approved.

Create an allowlist with exact URLs or explicit patterns:

from urllib.parse import urlparse

ALLOWED_URLS = {
    "https://example.com/pricing",
    "https://example.com/news.xml",
}

ALLOWED_PREFIXES = {
    "https://example.com/releases/",
}

def is_allowed(url: str) -> bool:
    parsed = urlparse(url)
    if parsed.scheme != "https":
        return False

    normalized = parsed._replace(fragment="").geturl()
    return (
        normalized in ALLOWED_URLS
        or any(normalized.startswith(prefix) for prefix in ALLOWED_PREFIXES)
    )

Do not automatically follow links discovered in page content. A pricing page may contain social links, tracking URLs, or user-generated content. Discovery should produce a review candidate, not expand the collection scope.

Redirects need the same check. Validate both the requested URL and the final URL after the HTTP request. Reject the result if either falls outside the allowlist.

Choose the least fragile source interface available:

• API: Prefer it when the approved provider exposes stable structured data.
• RSS: Useful for announcements and news items.
• Static page: Suitable for pricing, feature matrices, and positioning copy.
• HTML extraction: Use when no structured source exists. Keep selectors isolated and test them.
• Search or social source: Include only through an authorized API or explicitly approved endpoint.

The allowlist is a policy boundary. It also makes failures easier to explain: every observation maps back to a known source definition.

Implement resilient collection with auditable observations

A fetcher should fail clearly. Add timeouts, bounded retries, HTTP status checks, content-size limits, empty-content detection, and per-host rate limiting.

Here is a compact starting point:

import hashlib
import time
import requests
from dataclasses import dataclass
from datetime import datetime, timezone

@dataclass(frozen=True)
class FetchResult:
    requested_url: str
    final_url: str | None
    status: int | None
    body: str | None
    error: str | None

def fetch(url: str, attempts: int = 3) -> FetchResult:
    if not is_allowed(url):
        return FetchResult(url, None, None, None, "requested_url_not_allowed")

    for attempt in range(attempts):
        try:
            response = requests.get(
                url,
                timeout=(5, 20),
                allow_redirects=True,
                headers={"User-Agent": "ApprovedCompetitorMonitor/1.0"},
            )

            final_url = response.url
            if not is_allowed(final_url):
                return FetchResult(
                    url, final_url, response.status_code, None,
                    "redirect_outside_allowlist",
                )

            response.raise_for_status()
            body = response.text.strip()

            if not body:
                return FetchResult(
                    url, final_url, response.status_code, None,
                    "empty_response",
                )

            return FetchResult(
                url, final_url, response.status_code, body, None
            )

        except requests.RequestException as exc:
            if attempt == attempts - 1:
                return FetchResult(url, None, None, None, str(exc))
            time.sleep(2 ** attempt)

    raise RuntimeError("unreachable")

Treat the timeout and retry values as configuration, not universal defaults. Respect source terms and published access limits.

Store an observation even when collection fails:

{
  "observation_id": "generated-stable-id",
  "competitor": "Acme",
  "source_id": "acme_pricing",
  "fetched_at": "timestamp",
  "requested_url": "https://example.com/pricing",
  "final_url": "https://example.com/pricing",
  "content_hash": "sha256-value",
  "extracted_text": "normalized content",
  "http_status": 200,
  "error": null
}

The evidence URL must come from the fetch result. Never ask the language model to generate or recover a citation from memory.

Keep collection errors separate from business changes. A missing page is not proof that a product or plan was removed. A changed HTML selector is not a pricing update. If some sources fail, mark the run as partial and preserve the previous successful state for those sources.

Emit structured logs for the run ID, source ID, duration, HTTP result, extraction result, and snapshot decision. The Python agent observability guide covers the logging and monitoring layer in more depth.

Compare successive snapshots instead of re-summarizing everything

Do not send the entire page to a model on every run. Normalize it, create a deterministic snapshot, and compare it with the previous successful snapshot.

Normalization should be source-specific. It can:

• Collapse repeated whitespace.
• Remove known navigation and cookie boilerplate.
• Preserve headings.
• Convert prices into stable fields.
• Sort feature lists only when order has no meaning.
• Keep raw values alongside normalized values.

import hashlib
import json
import re

def normalize_text(text: str) -> str:
    return re.sub(r"\s+", " ", text).strip()

def snapshot_hash(fields: dict) -> str:
    payload = json.dumps(
        fields,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

def diff_fields(previous: dict, current: dict) -> list[dict]:
    changes = []
    for field in sorted(previous.keys() | current.keys()):
        before = previous.get(field)
        after = current.get(field)

        if before == after:
            continue

        if field not in previous:
            change_type = "added"
        elif field not in current:
            change_type = "removed"
        else:
            change_type = "modified"

        changes.append({
            "change_type": change_type,
            "field": field,
            "before": before,
            "after": after,
        })

    return changes

The first successful run creates a baseline. It should not generate a report full of “new” changes unless you explicitly want an initial inventory.

Keep two storage views:

1. Immutable snapshots: Raw response metadata and extracted state for every successful observation.
2. Current state: The latest successful snapshot for each source.

This lets you reconstruct history without scanning every record during each run.

Protect the diff stage from bad inputs. Do not replace a good snapshot after an empty extraction, failed request, or partial parse. Use a unique source-and-run constraint to reject duplicate processing. If schedules overlap, acquire a source-level lock before comparing and updating state.

Generate a change-only report with fact labels and URL evidence

The model’s job is narrow: summarize verified differences. It should not decide what URLs were visited or invent missing fields.

Use structured input and output:

{
  "change_type": "modified",
  "field": "professional_plan_price",
  "before": "previous observed value",
  "after": "current observed value",
  "confidence": "high",
  "classification": "Fact",
  "observation_id": "generated-stable-id",
  "evidence_url": "https://example.com/pricing",
  "detected_at": "timestamp"
}

Require one of these labels:

• Fact: Directly observed in an approved source.
• Inference: A conclusion derived from observed facts.
• Unverified: The pipeline lacks enough evidence to validate the statement.

This mirrors the useful distinction between competitor claims, independent evidence, and supported inference described by Relevance AI’s competitive analyst.

A Markdown report can remain concise:

## Changed

- **Fact — Pricing:** The professional plan value changed.
  - Before: `previous observed value`
  - After: `current observed value`
  - Evidence: https://example.com/pricing

## Removed

- No verified removals.

## Unverified

- The feature section could not be extracted. The previous snapshot was retained.

## No changes detected

- News feed
- Positioning page

Useful monitored areas include pricing, feature launches, positioning, marketing activity, and customer sentiment. Keep the evidence type visible. A competitor pricing page proves what that competitor currently claims. It does not independently prove product quality or market impact.

If there are no verified differences, skip the model call and emit “No changes detected” deterministically.

Schedule, deploy, and make the job safe to rerun

Expose one command that works locally and in production:

python -m monitor run --config sources.yaml
python -m monitor run --config sources.yaml --dry-run

A dry run should fetch and diff without updating current state or sending notifications.

Provide secrets and runtime settings through the environment:

export DATABASE_URL="..."
export REPORT_WEBHOOK_URL="..."
export LOG_LEVEL="INFO"

Then schedule the command with cron or a managed scheduled-job service:

*/30 * * * * cd /app && python -m monitor run --config sources.yaml

That example matches a price-change objective measured in 30-minute windows. Your cadence should follow the source and the decision it supports. Pricing pages may run more frequently. News and RSS can run on a regular daily schedule. Feature pages often need less frequent collection. Slower market signals should not be polled like operational alerts.

Calendar schedules are not the only trigger model. Relevance AI also supports email and API triggers. For this workflow, a schedule remains easier to audit unless an authorized webhook or API event is available.

Make retries safe:

• Derive a stable run ID from the scheduled window and job name.
• Enforce unique source-and-run records.
• Upsert observations rather than appending duplicates.
• Lock each source during state comparison.
• Resume from completed stages.
• Never overwrite the last good snapshot after a failed fetch.
• Send a report only after its changes have been committed.
• Record notification delivery separately from collection.

These patterns are covered in the guide to idempotent scheduled Python jobs.

Before deployment, test redirect rejection, empty pages, changed selectors, repeated runs, concurrent starts, and notification failures. Alert on sustained collection or extraction errors. Define how long raw responses, snapshots, reports, and logs should be retained.

Track operational metrics that reflect the original goal. Datagrid suggests measures such as median price-change detection time. Add your own false-positive rate so you can see whether normalization and extraction rules create noisy reports.

Conclusion

A dependable competitor monitoring AI agent is mostly a controlled data pipeline. The model is only the final summarization step.

Start with approved sources. Preserve requested and final URLs. Store immutable observations. Diff only against the last successful snapshot. Report verified changes with explicit labels and evidence links. Then make the scheduled job idempotent, observable, and safe under partial failure.

Once the command works locally, a managed scheduled runtime such as HollowHost can provide the execution lifecycle around it without changing the core Python workflow.