Building a Real-Time Market Research Agent with Claude — Automate Competitor Analysis, Price Monitoring & Trend Detection

// src/tools/competitorTools.js
export const competitorTools = [
  {
    name: "web_search",
    description: "Search the web for up-to-date information. Use this to find competitor press releases, news, and product updates.",
    input_schema: {
      type: "object",
      properties: {
        query: {
          type: "string",
          description: "Search query. Including company names, product names, and date ranges improves accuracy."
        },
        max_results: {
          type: "number",
          description: "Maximum number of results to return (default: 5)"
        }
      },
      required: ["query"]
    }
  },
  {
    name: "save_competitor_data",
    description: "Saves competitor analysis data locally for delta comparison on the next run.",
    input_schema: {
      type: "object",
      properties: {
        competitor_name: { type: "string" },
        data: {
          type: "object",
          description: "Analysis data to store (pricing, features, news, etc.)"
        },
        timestamp: { type: "string" }
      },
      required: ["competitor_name", "data", "timestamp"]
    }
  },
  {
    name: "load_previous_data",
    description: "Loads previously saved competitor analysis data for comparison.",
    input_schema: {
      type: "object",
      properties: {
        competitor_name: { type: "string" }
      },
      required: ["competitor_name"]
    }
  }
];

// src/agents/competitorAgent.js
import Anthropic from "@anthropic-ai/sdk";
import fs from "fs/promises";
import path from "path";
 
const client = new Anthropic();
const DATA_DIR = "./data/competitors";
 
async function executeTool(toolName, toolInput) {
  switch (toolName) {
    case "save_competitor_data": {
      await fs.mkdir(DATA_DIR, { recursive: true });
      const filePath = path.join(DATA_DIR, `${toolInput.competitor_name}.json`);
      await fs.writeFile(filePath, JSON.stringify(toolInput, null, 2));
      return { success: true, message: `Saved data for ${toolInput.competitor_name}` };
    }
 
    case "load_previous_data": {
      const filePath = path.join(DATA_DIR, `${toolInput.competitor_name}.json`);
      try {
        const raw = await fs.readFile(filePath, "utf8");
        return JSON.parse(raw);
      } catch {
        return { exists: false, message: "No previous data found (first run)" };
      }
    }
 
    default:
      return { error: `Unknown tool: ${toolName}` };
  }
}
 
export async function runCompetitorAnalysis(competitors) {
  const systemPrompt = `You are a market research specialist. For each competitor provided,
conduct a thorough investigation covering:
 
1. Latest press releases and news (past 7 days)
2. Product and feature changes
3. Pricing updates
4. Social media and community sentiment
5. Delta from previous data (what changed)
 
Guidelines:
- Clearly distinguish facts from inferences
- Rank findings by importance (high / medium / low)
- Include 1–2 sentences of business implication per finding
- Output in English`;
 
  const userMessage = `Analyze these competitors: ${competitors.join(", ")}
 
For each company: use web_search to gather the latest information, call load_previous_data to compare
with prior state, and save updated data with save_competitor_data.
Finish with a structured summary for all companies.`;
 
  let messages = [{ role: "user", content: userMessage }];
 
  while (true) {
    const response = await client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 8192,
      system: systemPrompt,
      tools: competitorTools,
      messages
    });
 
    messages.push({ role: "assistant", content: response.content });
 
    if (response.stop_reason === "end_turn") {
      const textBlock = response.content.find(b => b.type === "text");
      return textBlock?.text ?? "Analysis complete (no text output)";
    }
 
    if (response.stop_reason === "tool_use") {
      const toolResults = [];
      for (const block of response.content) {
        if (block.type !== "tool_use") continue;
 
        console.log(`[Tool call] ${block.name}:`, block.input);
        const result = await executeTool(block.name, block.input);
        console.log(`[Tool result]:`, result);
 
        toolResults.push({
          type: "tool_result",
          tool_use_id: block.id,
          content: JSON.stringify(result)
        });
      }
      messages.push({ role: "user", content: toolResults });
    }
  }
}

// src/agents/priceMonitorAgent.js
import Anthropic from "@anthropic-ai/sdk";
import fs from "fs/promises";
import path from "path";
 
const client = new Anthropic();
const PRICE_DATA_DIR = "./data/prices";
 
const priceTools = [
  {
    name: "extract_pricing_from_page",
    description: "Extracts pricing information from a URL and returns structured data.",
    input_schema: {
      type: "object",
      properties: {
        url: { type: "string", description: "URL of the pricing page" },
        product_name: { type: "string", description: "Product or company name" }
      },
      required: ["url", "product_name"]
    }
  },
  {
    name: "compare_prices",
    description: "Compares current pricing data against the previously saved snapshot and detects changes.",
    input_schema: {
      type: "object",
      properties: {
        product_name: { type: "string" },
        current_prices: { type: "object" }
      },
      required: ["product_name", "current_prices"]
    }
  }
];
 
async function executePriceTool(toolName, toolInput) {
  if (toolName === "compare_prices") {
    const filePath = path.join(PRICE_DATA_DIR, `${toolInput.product_name}.json`);
    let previous = null;
    try {
      previous = JSON.parse(await fs.readFile(filePath, "utf8"));
    } catch { /* First run */ }
 
    await fs.mkdir(PRICE_DATA_DIR, { recursive: true });
    const saveData = {
      product_name: toolInput.product_name,
      prices: toolInput.current_prices,
      updated_at: new Date().toISOString()
    };
    await fs.writeFile(filePath, JSON.stringify(saveData, null, 2));
 
    if (!previous) {
      return { change_detected: false, message: "Initial snapshot saved (nothing to compare)" };
    }
 
    const changes = [];
    for (const [plan, price] of Object.entries(toolInput.current_prices)) {
      const prevPrice = previous.prices?.[plan];
      if (prevPrice !== undefined && prevPrice !== price) {
        const diff = typeof price === "number" ? price - prevPrice : "changed";
        const direction = typeof diff === "number" ? (diff > 0 ? "price increase" : "price decrease") : "";
        changes.push({ plan, prev: prevPrice, current: price, diff, direction });
      }
    }
 
    return {
      change_detected: changes.length > 0,
      changes,
      previous_updated: previous.updated_at
    };
  }
 
  return { error: `Unknown tool: ${toolName}` };
}
 
export async function runPriceMonitoring(targets) {
  const systemPrompt = `You are a price monitoring agent. Analyze the pricing pages of SaaS and
e-commerce sites and accurately report any differences from the previous snapshot.
 
Output format:
- If changes found: product name, plan name, previous price, current price, percent change, business implication
- If no changes: state "No changes detected" briefly
- Use original currency (USD, JPY, EUR, etc.)`;
 
  const userMessage = `Investigate pricing for the following products:\n${
    targets.map(t => `- ${t.name}: ${t.url}`).join("\n")
  }\n\nUse web_search to fetch current pricing, then compare_prices to diff against the previous snapshot.`;
 
  let messages = [{ role: "user", content: userMessage }];
 
  while (true) {
    const response = await client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 4096,
      system: systemPrompt,
      tools: priceTools,
      messages
    });
 
    messages.push({ role: "assistant", content: response.content });
 
    if (response.stop_reason === "end_turn") {
      const textBlock = response.content.find(b => b.type === "text");
      return textBlock?.text ?? "Price monitoring complete";
    }
 
    if (response.stop_reason === "tool_use") {
      const toolResults = [];
      for (const block of response.content) {
        if (block.type !== "tool_use") continue;
        const result = await executePriceTool(block.name, block.input);
        toolResults.push({
          type: "tool_result",
          tool_use_id: block.id,
          content: JSON.stringify(result)
        });
      }
      messages.push({ role: "user", content: toolResults });
    }
  }
}

// src/agents/trendAgent.js
import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
export async function runTrendDetection(keywords, timeRange = "past_7_days") {
  const systemPrompt = `You are a trend analysis specialist. For the keywords provided,
conduct thorough research across multiple sources and synthesize insights.
 
Analysis areas:
1. Rapidly rising topics (explain *why* they are gaining traction now)
2. Top 3 news stories with brief summaries (prioritize authoritative sources)
3. Community reaction on X (Twitter), Reddit, and Hacker News
4. Relationship to competitor activities uncovered elsewhere
5. Short-term predictions for the next 1–2 weeks (be explicit about confidence level)
 
Output structure:
## 🔥 Rising Trends
## 📰 Key News (Top 3)
## 💬 Community Sentiment
## 🔮 Short-Term Forecast
## 📊 Action Items (Business Implications)`;
 
  const response = await client.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 6144,
    system: systemPrompt,
    messages: [
      {
        role: "user",
        content: `Run a comprehensive trend analysis for the following keywords over the ${timeRange} period.
 
Keywords: ${keywords.join(", ")}
 
Run multiple web_search calls from different angles before synthesizing your findings.`
      }
    ],
    tools: [
      {
        name: "web_search",
        description: "Search the web for trend information.",
        input_schema: {
          type: "object",
          properties: {
            query: { type: "string" },
            time_filter: {
              type: "string",
              enum: ["past_day", "past_week", "past_month"]
            }
          },
          required: ["query"]
        }
      }
    ]
  });
 
  const textBlock = response.content.find(b => b.type === "text");
  return textBlock?.text ?? "Trend analysis complete";
}

// src/pipeline/reportPipeline.js
import Anthropic from "@anthropic-ai/sdk";
import { runCompetitorAnalysis } from "../agents/competitorAgent.js";
import { runPriceMonitoring } from "../agents/priceMonitorAgent.js";
import { runTrendDetection } from "../agents/trendAgent.js";
 
const client = new Anthropic();
 
const CONFIG = {
  competitors: ["OpenAI", "Google Gemini", "Mistral AI"],
  priceTargets: [
    { name: "OpenAI ChatGPT Plus", url: "https://openai.com/pricing" },
    { name: "Google Gemini Advanced", url: "https://one.google.com/about/ai-premium" }
  ],
  trendKeywords: ["Claude AI", "LLM 2026", "AI agents", "generative AI pricing"]
};
 
async function sendToSlack(report) {
  if (!process.env.SLACK_BOT_TOKEN) return;
 
  await fetch("https://slack.com/api/chat.postMessage", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": `Bearer ${process.env.SLACK_BOT_TOKEN}`
    },
    body: JSON.stringify({
      channel: process.env.SLACK_CHANNEL_ID,
      text: `📊 Market Research Report — ${new Date().toLocaleDateString("en-US")}`,
      blocks: [
        {
          type: "section",
          text: { type: "mrkdwn", text: `*📊 Market Research Report — ${new Date().toLocaleDateString("en-US")}*` }
        },
        { type: "divider" },
        { type: "section", text: { type: "mrkdwn", text: report.slice(0, 2900) } }
      ]
    })
  });
}
 
async function saveToNotion(title, content) {
  if (!process.env.NOTION_TOKEN) return;
 
  await fetch("https://api.notion.com/v1/pages", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${process.env.NOTION_TOKEN}`,
      "Notion-Version": "2022-06-28",
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      parent: { database_id: process.env.NOTION_DATABASE_ID },
      properties: {
        title: { title: [{ text: { content: title } }] },
        Date: { date: { start: new Date().toISOString().split("T")[0] } }
      },
      children: [
        {
          object: "block",
          type: "paragraph",
          paragraph: {
            rich_text: [{ type: "text", text: { content: content.slice(0, 2000) } }]
          }
        }
      ]
    })
  });
}
 
async function generateIntegratedReport(competitorReport, priceReport, trendReport) {
  const response = await client.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 4096,
    messages: [
      {
        role: "user",
        content: `Synthesize the following three reports into a concise executive summary.
 
## Competitor Analysis Report
${competitorReport}
 
## Price Monitoring Report
${priceReport}
 
## Trend Report
${trendReport}
 
Your integrated report must include:
1. **Top 3 Insights This Week**
2. **Urgent Alerts** (anything requiring immediate action)
3. **Action Items for Next Week**
4. **Medium- to Long-Term Implications**
 
Target audience: business owners. Prioritize information that drives decisions.`
      }
    ]
  });
 
  return response.content.find(b => b.type === "text")?.text ?? "";
}
 
export async function runMarketResearchPipeline() {
  console.log("🚀 Market research pipeline started:", new Date().toLocaleString("en-US"));
 
  try {
    // Run all three agents in parallel for maximum throughput
    const [competitorReport, priceReport, trendReport] = await Promise.all([
      runCompetitorAnalysis(CONFIG.competitors),
      runPriceMonitoring(CONFIG.priceTargets),
      runTrendDetection(CONFIG.trendKeywords)
    ]);
 
    console.log("✅ All three agents done — generating integrated report...");
 
    const integratedReport = await generateIntegratedReport(
      competitorReport, priceReport, trendReport
    );
 
    const title = `Market Research Report — ${new Date().toLocaleDateString("en-US")}`;
 
    await Promise.allSettled([
      sendToSlack(integratedReport),
      saveToNotion(title, integratedReport)
    ]);
 
    console.log("✅ Report delivered");
    return integratedReport;
 
  } catch (error) {
    console.error("❌ Pipeline error:", error);
    throw error;
  }
}

// src/scheduler.js
import cron from "node-cron";
import { runMarketResearchPipeline } from "./pipeline/reportPipeline.js";
 
// Every Monday at 8:00 AM JST
cron.schedule("0 8 * * 1", runMarketResearchPipeline, {
  timezone: "Asia/Tokyo"
});
 
console.log("⏰ Market research scheduler running");

# Start the scheduler
node src/scheduler.js
 
# One-off test run
node -e "import('./src/pipeline/reportPipeline.js').then(m => m.runMarketResearchPipeline())"

async function callWithRetry(fn, maxRetries = 3, delayMs = 2000) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await fn();
    } catch (err) {
      if (err.status === 529 && i < maxRetries - 1) {
        console.warn(`Rate limit hit — retrying in ${delayMs * (i + 1)}ms`);
        await new Promise(r => setTimeout(r, delayMs * (i + 1)));
      } else {
        throw err;
      }
    }
  }
}
 
const report = await callWithRetry(() => runCompetitorAnalysis(competitors));

const MAX_STEPS = 20;
let steps = 0;
 
while (steps < MAX_STEPS) {
  steps++;
  const response = await client.messages.create({ ... });
  // ...
  if (response.stop_reason === "end_turn") break;
}
 
if (steps >= MAX_STEPS) {
  console.warn("⚠️ Max steps reached — forcing exit");
}

// ❌ Wrong — passing an object directly will cause an error
content: result
 
// ✅ Correct — always stringify
content: JSON.stringify(result)

function chunkText(text, maxLength = 2800) {
  const chunks = [];
  for (let i = 0; i < text.length; i += maxLength) {
    chunks.push(text.slice(i, i + maxLength));
  }
  return chunks;
}

import { chromium } from "playwright";
 
async function screenshotAndAnalyze(url) {
  const browser = await chromium.launch();
  const page = await browser.newPage();
  await page.goto(url, { waitUntil: "networkidle" });
  const buf = await page.screenshot({ fullPage: true });
  await browser.close();
 
  const response = await client.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 2048,
    messages: [
      {
        role: "user",
        content: [
          {
            type: "image",
            source: { type: "base64", media_type: "image/png", data: buf.toString("base64") }
          },
          { type: "text", text: "Extract all pricing plan details from this screenshot and return as JSON." }
        ]
      }
    ]
  });
 
  return response.content.find(b => b.type === "text")?.text;
}

// workers/market-research.js
export default {
  async scheduled(event, env, ctx) {
    const { runMarketResearchPipeline } = await import("./pipeline/reportPipeline.js");
    ctx.waitUntil(runMarketResearchPipeline(env));
  }
};

# wrangler.toml
name = "market-research-agent"
main = "workers/market-research.js"
compatibility_date = "2026-04-01"
 
[[triggers.crons]]
crons = ["0 0 * * 1"]

const INDUSTRY_PROMPTS = {
  saas: "Pay special attention to pricing model shifts (per-seat, usage-based, flat rate) and trial/freemium strategy changes.",
  ecommerce: "Focus on shipping costs, loyalty point changes, and seasonal promotion strategies.",
  ai: "Prioritize benchmark performance, API rate limits, context window upgrades, and model deprecations."
};
 
const systemPrompt = BASE_PROMPT + "\n\n" + (INDUSTRY_PROMPTS[industry] ?? "");

import { createHash } from "node:crypto";
 
function urlSetSimHash(urls) {
  const tokens = urls.flatMap((u) => new URL(u).hostname.split("."));
  const bits = new Array(64).fill(0);
  for (const token of tokens) {
    const hash = createHash("md5").update(token).digest();
    for (let i = 0; i < 64; i++) {
      const bit = (hash[Math.floor(i / 8)] >> (i % 8)) & 1;
      bits[i] += bit ? 1 : -1;
    }
  }
  return BigInt("0b" + bits.map((b) => (b > 0 ? "1" : "0")).join(""));
}
 
function jaccardSimilarity(setA, setB) {
  const a = new Set(setA);
  const b = new Set(setB);
  const intersection = [...a].filter((x) => b.has(x)).length;
  const union = new Set([...a, ...b]).size;
  return intersection / union;
}
 
export function shouldRunLlmDiff(prevUrls, currUrls) {
  const sim = jaccardSimilarity(prevUrls, currUrls);
  return sim < 0.7;  // run the LLM only when sim drops below 0.7
}

[[queues.producers]]
binding = "MARKET_RESEARCH_QUEUE"
queue = "market-research-jobs"
 
[[queues.consumers]]
queue = "market-research-jobs"
max_batch_size = 1
max_batch_timeout = 10

export default {
  async scheduled(event, env) {
    await env.MARKET_RESEARCH_QUEUE.send({ type: "competitor", at: Date.now() });
    await env.MARKET_RESEARCH_QUEUE.send({ type: "pricing", at: Date.now() });
    await env.MARKET_RESEARCH_QUEUE.send({ type: "trends", at: Date.now() });
  },
  async queue(batch, env) {
    for (const msg of batch.messages) {
      const job = msg.body;
      if (job.type === "competitor") await runCompetitorAgent(env);
      else if (job.type === "pricing") await runPricingAgent(env);
      else if (job.type === "trends") await runTrendsAgent(env);
      msg.ack();
    }
  }
};

function chunkForSlack(report, chunkSize = 2500) {
  const blocks = [];
  for (let i = 0; i < report.length; i += chunkSize) {
    blocks.push({
      type: "section",
      text: { type: "mrkdwn", text: report.slice(i, i + chunkSize) }
    });
  }
  return blocks;
}

function getPollingInterval(date) {
  const day = date.getDay();        // 0 = Sunday
  const dom = date.getDate();
  const month = date.getMonth() + 1;
  const isQuarterEnd = [3, 6, 9, 12].includes(month) && dom >= 25;
  const isMonthEnd = dom >= 28;
  const isWeekend = day === 0 || day === 6 || day === 5;
  if (isQuarterEnd) return 30 * 60 * 1000;                  // 30-minute interval
  if (isMonthEnd || isWeekend) return 2 * 60 * 60 * 1000;   // 2-hour interval
  return 6 * 60 * 60 * 1000;                                 // 6-hour interval on weekdays
}

class RateLimitedQueue {
  constructor(maxPerMin = 12) {
    this.maxPerMin = maxPerMin;
    this.timestamps = [];
  }
  async waitForSlot() {
    const now = Date.now();
    this.timestamps = this.timestamps.filter((t) => now - t < 60_000);
    if (this.timestamps.length >= this.maxPerMin) {
      const wait = 60_000 - (now - this.timestamps[0]);
      await new Promise((r) => setTimeout(r, wait + 100));
      return this.waitForSlot();
    }
    this.timestamps.push(Date.now());
  }
}

const PRIORITY_PROMPT = `
Classify changes by the following priority:
- P0: Feature segmentation boundary change (Free ↔ Paid, Tier 1 ↔ Tier 2)
- P1: New feature launches — especially AI/automation features
- P2: Major UI/UX overhauls
- P3: Pricing changes
- P4: Marketing copy changes (usually no notification needed)
`;