id: developer-mention-radar
namespace: company.team
description: |
Watch Hacker News, Stack Overflow, GitHub and dev.to for mentions of your
project, skip anything already seen, let an LLM sort each new mention into
security, bug, question, comparison, praise or noise, and post a digest to
Slack grouped by urgency.
labels:
team: devrel
use-case: mention-monitoring
triggers:
- id: every_six_hours
type: io.kestra.plugin.core.trigger.Schedule
description: Check every six hours. None of these sites can push events, so the
flow polls and only looks at what is new since its last successful check.
cron: "0 */6 * * *"
inputs:
- id: keyword
type: STRING
displayName: Keyword
description: Exact term to watch, usually your project's name.
defaults: kestra
- id: devto_tag
type: STRING
displayName: dev.to tag
description: dev.to can only be searched by tag.
defaults: kestra
- id: exclude_github_org
type: STRING
displayName: GitHub org to exclude
description: Your own organization, so your own issues and pull requests are not
reported as mentions.
defaults: kestra-io
- id: first_run_lookback_hours
type: INT
displayName: First-run lookback (hours)
description: How far back to search before any state exists.
defaults: 24
- id: max_mentions
type: INT
displayName: Max mentions per run
description: Caps how many new mentions are sent to the model in one run, which
bounds the cost. The rest are picked up next run.
defaults: 40
- id: notify
type: BOOL
displayName: Post to Slack and save state
description: Set to false for a dry run that searches and classifies but posts
nothing and changes no state.
defaults: true
variables:
claude_model: claude-sonnet-5-5
tasks:
- id: checkpoints
type: io.kestra.plugin.core.output.OutputValues
description: Work out, per source, where the last successful check stopped.
Before the first run, look back first_run_lookback_hours.
values:
hacker_news: "{{ ((kv('mention-radar-' ~ (inputs.keyword | slugify),
errorOnMissing=false) ?? {}).sources ?? {}).hacker_news ??
(execution.startDate | dateAdd(0 - inputs.first_run_lookback_hours,
'HOURS')) }}"
stack_overflow: "{{ ((kv('mention-radar-' ~ (inputs.keyword | slugify),
errorOnMissing=false) ?? {}).sources ?? {}).stack_overflow ??
(execution.startDate | dateAdd(0 - inputs.first_run_lookback_hours,
'HOURS')) }}"
github: "{{ ((kv('mention-radar-' ~ (inputs.keyword | slugify),
errorOnMissing=false) ?? {}).sources ?? {}).github ??
(execution.startDate | dateAdd(0 - inputs.first_run_lookback_hours,
'HOURS')) }}"
devto: "{{ ((kv('mention-radar-' ~ (inputs.keyword | slugify),
errorOnMissing=false) ?? {}).sources ?? {}).devto ??
(execution.startDate | dateAdd(0 - inputs.first_run_lookback_hours,
'HOURS')) }}"
seen: "{{ ((kv('mention-radar-' ~ (inputs.keyword | slugify),
errorOnMissing=false) ?? {}).seen ?? []) | toJson }}"
- id: search
type: io.kestra.plugin.core.flow.Parallel
description: Query every source at once. A source that is down does not stop the
others; its checkpoint stays put so it is searched again next run.
tasks:
- id: hacker_news
type: io.kestra.plugin.core.http.Download
description: Stories and comments on Hacker News containing the exact keyword,
through the free Algolia search API.
uri: "https://hn.algolia.com/api/v1/search_by_date?query={{ inputs.keyword |
urlencode }}&tags={{ '(story,comment)' | urlencode
}}&typoTolerance=false&hitsPerPage=100&numericFilters={{
('created_at_i>' ~ (outputs.checkpoints.values.hacker_news |
timestamp)) | urlencode }}"
allowFailure: true
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: stack_overflow
type: io.kestra.plugin.core.http.Download
description: New Stack Overflow questions mentioning the keyword, through the
free Stack Exchange API.
uri: "https://api.stackexchange.com/2.3/search/advanced?site=stackoverflow&sort\
=creation&order=desc&pagesize=100&q={{ inputs.keyword | urlencode
}}&fromdate={{ outputs.checkpoints.values.stack_overflow | timestamp
}}"
allowFailure: true
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: github
type: io.kestra.plugin.core.http.Download
description: Issues and pull requests in other organizations' repositories that
mention the keyword.
uri: "https://api.github.com/search/issues?sort=created&order=desc&per_page=100\
&q={{ (inputs.keyword ~ ' created:>' ~
(outputs.checkpoints.values.github |
date(\"yyyy-MM-dd'T'HH:mm:ss'Z'\", timeZone='UTC')) ~ ' -org:' ~
inputs.exclude_github_org) | urlencode }}"
headers:
Accept: application/vnd.github+json
Authorization: "Bearer {{ secret('GITHUB_TOKEN') }}"
User-Agent: kestra-developer-mention-radar
allowFailure: true
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: devto
type: io.kestra.plugin.core.http.Download
description: Latest dev.to articles with the tag. dev.to has no date filter, so
older articles are dropped in the next step.
uri: "https://dev.to/api/articles/latest?per_page=50&tag={{ inputs.devto_tag |
urlencode }}"
allowFailure: true
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: normalize
type: io.kestra.plugin.scripts.python.Script
description: Put every source into one format, neutralize template markers in
the untrusted text, drop mentions already seen, cap the batch, and compute
the new checkpoints. Only sources that answered move their checkpoint
forward.
dependencies:
- kestra
inputFiles:
hacker_news.json: "{{ outputs.hacker_news.uri ?? '' }}"
stack_overflow.json: "{{ outputs.stack_overflow.uri ?? '' }}"
github.json: "{{ outputs.github.uri ?? '' }}"
devto.json: "{{ outputs.devto.uri ?? '' }}"
checkpoints.json: "{{ outputs.checkpoints.values | toJson }}"
outputFiles:
- new_mentions.json
script: |
import html, json, re
from datetime import datetime, timezone
from kestra import Kestra
checkpoints = json.load(open("checkpoints.json"))
seen = set(json.loads(checkpoints["seen"]))
run_started = "{{ execution.startDate }}"
max_mentions = int({{ inputs.max_mentions }})
def load(name):
text = open(f"{name}.json").read().strip()
return json.loads(text) if text else None
def parse(value):
return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
def iso(value):
# One timestamp format for every source, so checkpoints always parse the same way.
moment = datetime.fromtimestamp(value, timezone.utc) if isinstance(value, int) else parse(value)
return moment.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def clean(text, limit=400):
return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", text or ""))).strip()[:limit]
def defuse(value):
# Mention text is untrusted. Break template markers so Kestra never evaluates it later.
return re.sub(r"([{}])(?=[{}%#])", r"\1 ", value) if isinstance(value, str) else value
mentions, answered = [], []
data = load("hacker_news")
if data is not None:
answered.append("hacker_news")
for hit in data.get("hits", []):
mentions.append({"id": f"hn:{hit['objectID']}", "source": "Hacker News",
"title": hit.get("title") or f"Comment on: {hit.get('story_title') or 'a story'}",
"snippet": clean(hit.get("comment_text") or hit.get("story_text") or hit.get("title")),
"url": f"https://news.ycombinator.com/item?id={hit['objectID']}",
"author": hit.get("author"), "created_at": iso(hit["created_at"])})
data = load("stack_overflow")
if data is not None and "items" in data:
answered.append("stack_overflow")
for item in data["items"]:
mentions.append({"id": f"so:{item['question_id']}", "source": "Stack Overflow",
"title": html.unescape(item["title"]), "snippet": ", ".join(item.get("tags", [])),
"url": item["link"], "author": item.get("owner", {}).get("display_name"),
"created_at": iso(item["creation_date"])})
data = load("github")
if data is not None and "items" in data:
answered.append("github")
for item in data["items"]:
if item["user"]["type"] == "Bot":
continue # dependency bumps and translation bots are not people talking about you
mentions.append({"id": f"gh:{item['id']}", "source": "GitHub",
"title": item["title"], "snippet": clean(item.get("body")),
"url": item["html_url"], "author": item["user"]["login"], "created_at": iso(item["created_at"])})
data = load("devto")
if isinstance(data, list):
answered.append("devto")
for article in data:
if parse(article["published_at"]) <= parse(checkpoints["devto"]):
continue
mentions.append({"id": f"devto:{article['id']}", "source": "dev.to",
"title": article["title"], "snippet": clean(article.get("description")),
"url": article["url"], "author": article.get("user", {}).get("username"),
"created_at": iso(article["published_at"])})
# Oldest first, so mentions beyond the cap are the newest ones and the next run picks them up.
new = sorted((m for m in mentions if m["id"] not in seen), key=lambda m: m["created_at"])
batch, deferred = new[:max_mentions], new[max_mentions:]
source_of = {"hn": "hacker_news", "so": "stack_overflow", "gh": "github", "devto": "devto"}
sources = {s: checkpoints[s] for s in source_of.values()}
for source in answered:
if any(source_of[m["id"].split(":")[0]] == source for m in deferred):
# Only move up to the newest mention actually sent; the seen list stops it being sent twice.
sent = [m["created_at"] for m in batch if source_of[m["id"].split(":")[0]] == source]
sources[source] = max(sent) if sent else sources[source]
else:
sources[source] = iso(run_started)
batch = [{k: defuse(v) for k, v in m.items()} for m in batch]
# Remember recent IDs so overlapping searches never report the same mention twice.
new_seen = ([m["id"] for m in batch] + json.loads(checkpoints["seen"]))[:1000]
json.dump(batch, open("new_mentions.json", "w"))
Kestra.outputs({
"new_count": len(batch),
"left_for_next_run": len(deferred),
"answered": answered,
"failed_sources": [s for s in sources if s not in answered],
"new_state": {"sources": sources, "seen": new_seen},
})
- id: triage
type: io.kestra.plugin.core.flow.If
description: Classify and report only when there is something new, so quiet runs
cost nothing.
condition: "{{ outputs.normalize.vars.new_count > 0 }}"
then:
- id: classify
type: io.kestra.plugin.ai.completion.ChatCompletion
description: One model call for the whole batch. Each mention gets a category, a
relevance flag, a one-line summary and a suggested action.
provider:
type: io.kestra.plugin.ai.provider.Anthropic
apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
modelName: "{{ vars.claude_model }}"
configuration:
maxToken: 16000
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
messages:
- type: SYSTEM
content: |
You triage public mentions of the software project "{{ inputs.keyword }}" for its maintainers and developer relations team. Respond with a JSON array only: no code fences, no commentary.
- type: USER
content: |
For each mention below, return {"id", "category", "relevant", "summary", "action"}:
- id: copy it exactly.
- category: one of security, bug, question, comparison, praise, other.
security = a vulnerability, exploit or CVE. bug = something broken. question = someone asking for help.
comparison = the project compared with or chosen over or against another tool. praise = a positive experience.
- relevant: false if the mention is not about this software project (for example another product or a word with the same name), otherwise true.
- summary: one sentence of at most 25 words saying what the person said.
- action: one short suggestion for the team, such as "reply with the docs link for X" or "share on social".
Mentions:
{{ read(outputs.normalize.outputFiles['new_mentions.json']) }}
- id: route
type: io.kestra.plugin.scripts.python.Script
description: Join the classification back onto the mentions by ID, so links
always come from the sources and never from the model, then build the
Slack digest grouped by urgency.
dependencies:
- kestra
inputFiles:
mentions.json: "{{ outputs.normalize.outputFiles['new_mentions.json'] }}"
classification.txt: "{{ outputs.classify.textOutput }}"
outputFiles:
- mentions.jsonl
script: |
import json, re
from kestra import Kestra
mentions = {m["id"]: m for m in json.load(open("mentions.json"))}
raw = re.sub(r"^```(json)?|```$", "", open("classification.txt").read().strip(), flags=re.M).strip()
try:
labels = {c["id"]: c for c in json.loads(raw) if c.get("id") in mentions}
except (ValueError, AttributeError, TypeError, KeyError):
raise SystemExit(f"The model did not return a JSON array of labels. It returned: {raw[:200]}")
order = [("security", "🔒 Security"), ("bug", "🐛 Bugs"), ("question", "❓ Questions"),
("comparison", "⚔️ Comparisons"), ("praise", "🎉 Praise")]
rows, counts = [], {key: 0 for key, _ in order + [("other", "")]}
with open("mentions.jsonl", "w") as out:
for mid, mention in mentions.items():
label = labels.get(mid, {"category": "other", "relevant": True, "summary": mention["title"], "action": "review"})
row = {**mention, **{k: label.get(k) for k in ("category", "relevant", "summary", "action")}}
out.write(json.dumps(row) + "\n")
if row["relevant"]:
row["category"] = row["category"] if row["category"] in counts else "other"
counts[row["category"]] += 1
rows.append(row)
listed = [r for r in rows if r["category"] != "other"]
lines = [f"*{len(listed)} new mention(s) of {{ inputs.keyword }}* ({counts['other']} minor, {len(mentions) - len(rows)} unrelated, not listed)"]
for key, heading in order:
group = [r for r in listed if r["category"] == key]
if group:
lines.append(f"\n*{heading}*")
lines += [f"• <{r['url']}|{r['source']}>: {r['summary']} _{r['action']}_" for r in group]
names = {"hacker_news": "Hacker News", "stack_overflow": "Stack Overflow", "github": "GitHub", "devto": "dev.to"}
failed = [names[s] for s in {{ outputs.normalize.vars.failed_sources | toJson }}]
if failed:
lines.append(f"\n_Not reachable this run, will be retried next run: {', '.join(failed)}_")
Kestra.outputs({"listed_count": len(listed), "counts": counts, "digest": "\n".join(lines)})
- id: notify_team
type: io.kestra.plugin.core.flow.If
description: Post only when notify is on and at least one mention is worth the
team's attention.
condition: "{{ inputs.notify and outputs.route.vars.listed_count > 0 }}"
then:
- id: post_digest
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Post the digest, most urgent categories first.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "{{ outputs.route.vars.digest }}"
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: save_state
type: io.kestra.plugin.core.flow.If
description: Save checkpoints and seen IDs only after everything above
succeeded, so a failed run is simply retried next time without losing or
repeating mentions.
condition: "{{ inputs.notify }}"
then:
- id: store_state
type: io.kestra.plugin.core.kv.Set
description: Persist the per-source checkpoints and the recently seen mention IDs.
key: "mention-radar-{{ inputs.keyword | slugify }}"
kvType: JSON
value: "{{ outputs.normalize.vars.new_state | toJson }}"
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alert the team with the failed task, the error and a link to the
execution. State was not saved, so the next run retries the same mentions.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Developer mention radar FAILED* in flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}.
Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the execution logs' }}
{{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
No state was saved, so the next run retries the same mentions.
outputs:
- id: new_mentions
type: INT
displayName: New mentions
description: New mentions found in this run, before the relevance filter.
value: "{{ outputs.normalize.vars.new_count }}"
- id: mentions_file
type: STRING
displayName: Classified mentions file
description: Internal storage URI of mentions.jsonl, one classified mention per
line, when there were new mentions.
value: "{{ outputs.route.outputFiles['mentions.jsonl'] ?? 'none' }}"
