Competitive intelligence used to mean hiring an analyst to read reports and visit competitor websites once a quarter. Now it means running an AI agent continuously, surfacing changes the moment they happen. This guide walks through building a practical competitive intelligence agent using current AI tooling — one that monitors competitor pricing, job postings, product updates, and press coverage without requiring constant human oversight.
What a Competitive Intelligence Agent Actually Does
The agent’s job is to answer four questions on a continuous basis:
- What changed on competitor websites? Product pages, pricing pages, feature lists, documentation.
- What are competitors publishing? Blog posts, press releases, changelog entries, case studies.
- What signals are they sending? Job postings (what skills they’re hiring for implies product roadmap), patent filings, conference appearances.
- What is being said about them? News coverage, review sites, Reddit, Hacker News, LinkedIn.
A human analyst doing this manually works quarterly at best. An AI agent can check daily or hourly and only surfaces changes worth reading about — eliminating the noise while preserving the signal.
Architecture Overview
The agent follows a three-stage pipeline:
Scheduled trigger → Collection (web search + scraping) → Analysis (LLM) → Output (digest / alert)Collection layer: Fetches raw content from competitor URLs, search results, and news feeds. This is where MCP tools for web search and Firecrawl-style scraping fit in.
Analysis layer: An LLM compares new content against previous state, extracts meaningful changes, and generates structured summaries. This is where you route to a reasoning model for complex comparisons and a cheaper model for routine checks.
Output layer: Writes results to a Notion database, sends a Slack digest, or triggers an alert when a pricing change is detected.
Setting Up the Collection Layer
Using Exa for Structured News Search
Exa’s search API is well-suited for competitor monitoring because it returns full document content, not just snippets, and supports date filtering to retrieve only recent results.
import exa_py
from datetime import datetime, timedelta
exa = exa_py.Exa(api_key="your-exa-api-key")
def search_competitor_news(competitor_name: str, days_back: int = 7) -> list[dict]:
cutoff = datetime.now() - timedelta(days=days_back)
results = exa.search_and_contents(
f"{competitor_name} product update OR pricing OR announcement OR launch",
num_results=10,
start_published_date=cutoff.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
use_autoprompt=True,
text={"max_characters": 2000}
)
return [
{
"title": r.title,
"url": r.url,
"published_date": r.published_date,
"content": r.text,
"score": r.score
}
for r in results.results
]Monitoring Competitor Web Pages for Changes
For direct page monitoring, you want to store a hash or text snapshot of each page and detect diffs. A simple approach using Firecrawl:
import hashlib
import json
from pathlib import Path
import firecrawl
app = firecrawl.FirecrawlApp(api_key="your-firecrawl-key")
SNAPSHOT_DIR = Path("competitor_snapshots")
SNAPSHOT_DIR.mkdir(exist_ok=True)
def check_page_for_changes(url: str, competitor: str) -> dict:
result = app.scrape_url(url, params={"formats": ["markdown"]})
current_content = result.get("markdown", "")
current_hash = hashlib.sha256(current_content.encode()).hexdigest()
snapshot_path = SNAPSHOT_DIR / f"{competitor}_{hashlib.md5(url.encode()).hexdigest()}.json"
changed = False
previous_content = ""
if snapshot_path.exists():
snapshot = json.loads(snapshot_path.read_text())
if snapshot["hash"] != current_hash:
changed = True
previous_content = snapshot["content"]
snapshot_path.write_text(json.dumps({
"hash": current_hash,
"content": current_content,
"url": url,
"last_checked": datetime.now().isoformat()
}))
return {
"url": url,
"changed": changed,
"current_content": current_content,
"previous_content": previous_content
}Using MCP Tools for Real-Time Search
If your agent framework supports MCP, the Brave Search or Exa MCP servers handle the web search calls natively. This means your agent can trigger searches as tool calls within its reasoning loop rather than requiring pre-scheduled fetches:
# With LangChain + MCP tools
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_anthropic import ChatAnthropic
async def run_intelligence_sweep(competitors: list[str]) -> str:
async with MultiServerMCPClient({
"exa": {
"command": "npx",
"args": ["-y", "exa-mcp-server"],
"env": {"EXA_API_KEY": "your-key"}
},
"firecrawl": {
"command": "npx",
"args": ["-y", "firecrawl-mcp"],
"env": {"FIRECRAWL_API_KEY": "your-key"}
}
}) as client:
tools = await client.get_tools()
llm = ChatAnthropic(model="claude-opus-4-7")
agent = llm.bind_tools(tools)
prompt = f"""
You are a competitive intelligence analyst. Research these competitors: {competitors}
For each competitor:
1. Search for recent news, product launches, and announcements from the last 7 days
2. Check their pricing page if you know the URL
3. Look for recent job postings that signal product direction
Summarize the key intelligence findings in a structured report.
"""
response = await agent.ainvoke(prompt)
return response.contentThe Analysis Layer: Extracting Signal from Noise
Raw content changes are mostly noise. A pricing page might have a CSS class name change. A blog page might update a footer timestamp. You need the LLM to distinguish meaningful changes from cosmetic ones.
Structured Change Detection
from anthropic import Anthropic
import json
client = Anthropic()
CHANGE_ANALYSIS_PROMPT = """
You are analyzing changes to a competitor's web page.
Previous content (abbreviated):
{previous}
Current content (abbreviated):
{current}
Identify ONLY changes that are strategically significant:
- Pricing changes (new tiers, price increases/decreases, feature tier changes)
- New product features or product launches
- Removal of features
- New partnerships or integrations announced
- Changes to target customer messaging
- New case studies or customer additions
For each significant change found, output JSON with:
{{
"change_type": "pricing|product|messaging|partnership|customer",
"description": "Brief description of what changed",
"significance": "high|medium|low",
"raw_evidence": "The exact text that shows this change"
}}
If no strategically significant changes exist, return an empty array [].
"""
def analyze_page_change(previous_content: str, current_content: str) -> list[dict]:
response = client.messages.create(
model="claude-opus-4-7",
max_tokens=1024,
messages=[{
"role": "user",
"content": CHANGE_ANALYSIS_PROMPT.format(
previous=previous_content[:3000],
current=current_content[:3000]
)
}]
)
try:
return json.loads(response.content[0].text)
except json.JSONDecodeError:
return []Job Posting Analysis for Roadmap Signals
Job postings are one of the most reliable leading indicators of a competitor’s direction. A company hiring five ML engineers for a feature they haven’t announced yet is advertising their roadmap.
def analyze_job_postings(job_listings: list[dict]) -> dict:
combined_jobs = "\n\n".join([
f"Role: {job['title']}\nDescription: {job['description'][:500]}"
for job in job_listings[:20]
])
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""
Analyze these job postings from a competitor. Identify:
1. Technical areas they're investing in (AI/ML, security, specific languages/platforms)
2. Product areas under active development (new features, new markets)
3. Go-to-market signals (new sales regions, enterprise vs. SMB focus)
4. Organizational scaling patterns
Job postings:
{combined_jobs}
Output a structured intelligence summary.
"""
}]
)
return {"analysis": response.content[0].text, "job_count": len(job_listings)}Scheduling and Orchestration
For production use, you want the agent to run on a schedule and only escalate when something meaningful is found. Two approaches work well here:
Approach 1: Script-Gated Scheduling
The most resource-efficient pattern is a script that runs frequently, performs the raw checks (page hash comparison, RSS feed parsing), and only triggers the full LLM analysis when a raw change is detected.
# check_competitors.py — runs every hour via cron
import json
import sys
from competitor_monitor import check_page_for_changes
WATCH_URLS = {
"acme-corp": [
"https://acmecorp.com/pricing",
"https://acmecorp.com/features",
],
"rival-io": [
"https://rival.io/pricing",
]
}
changes_detected = []
for competitor, urls in WATCH_URLS.items():
for url in urls:
result = check_page_for_changes(url, competitor)
if result["changed"]:
changes_detected.append({
"competitor": competitor,
"url": url,
"previous": result["previous_content"],
"current": result["current_content"]
})
# Output consumed by the scheduling layer
print(json.dumps({
"wakeAgent": len(changes_detected) > 0,
"data": {"changes": changes_detected}
}))Approach 2: LangGraph with Persistent State
For more complex workflows with conditional branching, LangGraph lets you define the intelligence pipeline as a stateful graph:
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class IntelState(TypedDict):
competitors: list[str]
raw_changes: Annotated[list, operator.add]
analyzed_changes: Annotated[list, operator.add]
digest: str
def collect_news(state: IntelState) -> dict:
all_changes = []
for competitor in state["competitors"]:
news = search_competitor_news(competitor)
all_changes.extend(news)
return {"raw_changes": all_changes}
def analyze_changes(state: IntelState) -> dict:
if not state["raw_changes"]:
return {"analyzed_changes": []}
analyzed = []
for change in state["raw_changes"]:
analysis = analyze_page_change("", change.get("content", ""))
if analysis:
analyzed.extend(analysis)
return {"analyzed_changes": analyzed}
def generate_digest(state: IntelState) -> dict:
if not state["analyzed_changes"]:
return {"digest": "No significant changes detected this run."}
changes_text = json.dumps(state["analyzed_changes"], indent=2)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1500,
messages=[{
"role": "user",
"content": f"Create a concise competitive intelligence digest from these findings:\n{changes_text}"
}]
)
return {"digest": response.content[0].text}
def should_continue(state: IntelState) -> str:
return "analyze" if state["raw_changes"] else END
graph = StateGraph(IntelState)
graph.add_node("collect", collect_news)
graph.add_node("analyze", analyze_changes)
graph.add_node("digest", generate_digest)
graph.set_entry_point("collect")
graph.add_conditional_edges("collect", should_continue)
graph.add_edge("analyze", "digest")
graph.add_edge("digest", END)
app = graph.compile()Output and Alerting
The agent is only useful if its findings reach the right people at the right time. Three output patterns cover most cases:
Slack digest (weekly): A formatted summary of all significant changes detected in the past 7 days. Useful for team-wide awareness without creating alert fatigue.
Slack alert (immediate): Triggered only for high-significance changes — pricing updates, major product launches, competitor acquisitions. Keep this threshold high.
Notion / Confluence database: Write all findings, including medium and low significance, to a structured database that product managers and marketers can search. This becomes a searchable competitive intelligence archive over time.
def send_slack_digest(digest: str, channel: str = "#competitive-intel") -> None:
import httpx
httpx.post(
"https://slack.com/api/chat.postMessage",
headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"},
json={
"channel": channel,
"text": "Competitive Intelligence Digest",
"blocks": [
{"type": "header", "text": {"type": "plain_text", "text": "Competitive Intelligence Digest"}},
{"type": "section", "text": {"type": "mrkdwn", "text": digest[:3000]}}
]
}
)Practical Considerations
Rate limiting and robots.txt: Only scrape pages that permit it. Most competitor product pages don’t restrict crawling, but check robots.txt and respect crawl delays. For aggressive monitoring, use official APIs where available — Crunchbase, LinkedIn, G2, and Glassdoor all have API access. Exa and similar services handle rate limiting and caching on your behalf.
Data freshness vs. cost: Running LLM analysis on every fetched page is expensive and unnecessary. The hash-based change detection pattern means you only invoke the LLM when raw content has actually changed, which dramatically reduces API costs. On a 20-competitor watchlist, raw checks cost almost nothing; LLM analysis might trigger a handful of times per week.
False positive tuning: Competitive intelligence agents learn what matters through use. Start with a high significance threshold, review flagged changes for a few weeks, and tune the analysis prompt based on what was actually useful versus noise. Adding your product’s positioning context to the prompt — “we compete on ease of use, not price, so messaging changes matter more than pricing changes” — significantly improves signal quality.
The underlying pattern here isn’t complicated. A scheduled script checks whether anything changed. If something changed, an LLM analyzes whether the change matters. If it matters, someone gets notified. The AI adds value in the “does this matter?” step — the judgment that previously required a human analyst sitting down with a spreadsheet every quarter.