id: ai-deep-research-market-intelligence
namespace: company.strategy
description: |
Autonomous competitor intelligence and deep web research agent.
Orchestrates live web retrieval via Tavily and Google Gemini to extract
competitor product moves, pricing shifts, and strategic threats into a
cited executive briefing with automated Slack escalation on high-threat findings.
triggers:
- id: weekly_research_digest
type: io.kestra.plugin.core.trigger.Schedule
description: Weekly Monday sweep to audit competitor announcements and market
moves. Shipped disabled by default.
cron: "0 9 * * 1"
disabled: true
- id: adhoc_research_webhook
type: io.kestra.plugin.core.trigger.Webhook
description: Authenticated webhook to trigger ad-hoc competitive research on demand.
key: "{{ secret('WEBHOOK_KEY') }}"
inputs:
- id: target_entity
type: STRING
defaults: OpenAI
description: The competitor company, brand, or technology to investigate.
- id: focus_areas
type: STRING
defaults: Product releases, enterprise pricing changes, API updates, and
regulatory filings
description: Specific domain angles or strategic areas to prioritize during research.
- id: max_sources
type: INT
defaults: 5
description: Maximum number of search sources to retrieve and synthesize.
tasks:
- id: gather_web_intelligence
type: io.kestra.plugin.ai.agent.AIAgent
description: Deploys an AI research agent to query live web sources via Tavily
and synthesize findings into structured intelligence.
provider:
type: io.kestra.plugin.ai.provider.GoogleGemini
apiKey: "{{ secret('GEMINI_API_KEY') }}"
modelName: gemini-2.5-flash
configuration:
temperature: 0.1
maxToken: 8192
responseFormat:
type: JSON
jsonSchema:
type: object
required:
- target_entity
- market_threat_level
- executive_summary
- key_findings
- swot_analysis
- strategic_recommendations
properties:
target_entity:
type: string
market_threat_level:
type: string
enum:
- LOW
- MEDIUM
- HIGH
- CRITICAL
executive_summary:
type: string
key_findings:
type: array
items:
type: object
required:
- title
- source_url
- category
- impact_summary
properties:
title:
type: string
source_url:
type: string
category:
type: string
impact_summary:
type: string
swot_analysis:
type: object
required:
- strengths
- weaknesses
- opportunities
- threats
properties:
strengths:
type: array
items:
type: string
weaknesses:
type: array
items:
type: string
opportunities:
type: array
items:
type: string
threats:
type: array
items:
type: string
strategic_recommendations:
type: array
items:
type: string
contentRetrievers:
- type: io.kestra.plugin.ai.retriever.TavilyWebSearch
apiKey: "{{ secret('TAVILY_API_KEY') }}"
maxResults: "{{ inputs.max_sources }}"
systemMessage: |
You are a Principal Corporate Strategy and Market Intelligence Director.
Your mandate is to research competitor activities with objective, evidence-based rigor.
Guidelines:
1. Always search for primary sources: official press releases, developer blogs, documentation, and pricing pages.
2. Classify market threat levels:
- CRITICAL: Direct assault on our core product, massive price cut, or major enterprise acquisition.
- HIGH: Major feature launch matching our roadmap, key executive defection, or disruptive open-source model release.
- MEDIUM: Minor product enhancements, beta programs, or marketing campaigns.
- LOW: Routine bug fixes, minor documentation updates, or speculative rumors.
3. Maintain a neutral, analytical tone. Do not use marketing adjectives or speculation.
prompt: |
Execute deep research on the target entity: {{ inputs.target_entity }}
Focus Areas: {{ inputs.focus_areas }}
Search the live web for the latest developments, synthesize findings, conduct a thorough SWOT analysis, and output the required JSON schema.
- id: compile_executive_briefing
type: io.kestra.plugin.scripts.python.Script
description: Formats structured agent output into a publication-ready executive
briefing markdown artifact.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
intelligence.json: "{{ outputs.gather_web_intelligence.textOutput }}"
outputFiles:
- competitor-market-briefing.md
script: |
import json
from datetime import datetime, timezone
from kestra import Kestra
with open("intelligence.json", "r", encoding="utf-8") as f:
data = json.load(f)
target = data.get("target_entity", "Unknown Competitor")
threat_level = data.get("market_threat_level", "LOW")
summary = data.get("executive_summary", "No executive summary provided.")
findings = data.get("key_findings", [])
swot = data.get("swot_analysis", {})
recommendations = data.get("strategic_recommendations", [])
threat_emoji = {
"CRITICAL": "🚨 CRITICAL",
"HIGH": "⚠️ HIGH",
"MEDIUM": "ℹ️ MEDIUM",
"LOW": "✅ LOW"
}.get(threat_level, threat_level)
md_lines = [
f"# Executive Market Intelligence Briefing: {target}",
f"**Date:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')} | **Threat Level:** {threat_emoji}",
"",
"## Executive Summary",
summary,
"",
"## Key Market Findings",
"| Source / Title | Category | Strategic Impact |",
"| :--- | :--- | :--- |"
]
for f_item in findings:
title = f_item.get('title', 'Untitled')
url = f_item.get('source_url', '#')
category = f_item.get('category', 'General')
impact = f_item.get('impact_summary', '')
md_lines.append(f"| [{title}]({url}) | {category} | {impact} |")
md_lines.extend([
"",
"## SWOT Matrix",
"",
"### Strengths",
"\n".join([f"- {s}" for s in swot.get("strengths", [])]) or "- None reported",
"",
"### Weaknesses",
"\n".join([f"- {w}" for w in swot.get("weaknesses", [])]) or "- None reported",
"",
"### Opportunities",
"\n".join([f"- {o}" for o in swot.get("opportunities", [])]) or "- None reported",
"",
"### Threats",
"\n".join([f"- {t}" for t in swot.get("threats", [])]) or "- None reported",
"",
"## Strategic Countermeasures",
"\n".join([f"1. {r}" for r in recommendations]) or "1. Continue periodic market observation."
])
report_content = "\n".join(md_lines)
with open("competitor-market-briefing.md", "w", encoding="utf-8") as out:
out.write(report_content)
print(f"Generated competitor-market-briefing.md for {target} with threat level {threat_level}")
Kestra.outputs({
"threat_level": threat_level,
"target": target,
"summary": summary[:200] + "..." if len(summary) > 200 else summary,
"findings_count": len(findings)
})
- id: threat_gate
type: io.kestra.plugin.core.flow.If
description: Evaluates whether the competitor development poses a HIGH or
CRITICAL market threat.
condition: "{{ outputs.compile_executive_briefing.vars.threat_level == 'HIGH' or
outputs.compile_executive_briefing.vars.threat_level == 'CRITICAL' }}"
then:
- id: alert_leadership_channel
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alerts executive strategy and leadership channels on high-threat
competitor moves.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#market-intelligence",
"text": "🚨 *High-Threat Market Move Detected: {{ inputs.target_entity }}* 🚨\n\n*Threat Level:* `{{ outputs.compile_executive_briefing.vars.threat_level }}`\n*Sources Analyzed:* `{{ outputs.compile_executive_briefing.vars.findings_count }}`\n\n*Executive Summary:*\n{{ outputs.compile_executive_briefing.vars.summary }}\n\nView the full cited briefing in Kestra execution `{{ execution.id }}`."
}
else:
- id: log_routine_briefing
type: io.kestra.plugin.core.log.Log
description: Logs standard execution pass-through for routine or low-threat
market updates.
message: "Routine market research completed for {{ inputs.target_entity }}.
Threat level: {{ outputs.compile_executive_briefing.vars.threat_level
}}. Zero escalations triggered."
errors:
- id: alert_on_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alerts technical on-call if web search or agent inference fails.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#platform-alerts",
"text": "❌ *Market Intelligence Pipeline Failed* ❌\nWorkflow `{{ flow.id }}` failed during research on `{{ inputs.target_entity }}` (Execution `{{ execution.id }}`)."
}
