Monitor web changes with LangGraph

The mental model: a deployed graph subscribes

A LangGraph agent usually runs as a service, so it fits the push pattern. Rather than polling pages inside a node, you subscribe to the Hypeline stream over Server-Sent Events and drive the graph once per real change. Each event is already extracted (main content only), deduplicated, and delivered on one stable schema, whether it came from a feed, a watched web page, or a streaming source. If the process restarts, you reconnect with the cursor of the last event you handled and resume exactly there.

Subscribe over SSE and drive the graph

Register the sources once, then let the stream drive the loop. One deduplicated event arrives per genuine change, and the cursor makes a restart a non-event.

# pip install hypeline langgraph
from hypeline import Hypeline
from langgraph.graph import StateGraph, START, END

hy = Hypeline("hype_...")

# Register the sources this agent should watch (idempotent).
for url in [
    "https://competitor.example/pricing",
    "https://competitor.example/changelog",
]:
    hy.create_watch(url, keywords=["price", "plan", "deprecat"])

# A deployed graph reacts to each genuinely-new change.
def triage(state):
    event = state["event"]
    # classify, summarize, decide, act...
    return {"note": f"{event.title}: {event.url}"}

builder = StateGraph(dict)
builder.add_node("triage", triage)
builder.add_edge(START, "triage")
builder.add_edge("triage", END)
graph = builder.compile()

# The stream drives the loop: one event per real change,
# resumable by cursor if the process restarts.
for event in hy.stream(keyword="price"):
    graph.invoke({"event": event})

The hypeline client is a thin wrapper over the HTTP API: create_watch is POST /v1/sources and stream() opens the GET /v1/stream Server-Sent Events endpoint, so you can subscribe with a plain curl or fetch call if you prefer.

Or wake the graph with a signed webhook

If you would rather not hold an open connection, add a webhook destination to an alert (POST /v1/alerts, then POST /v1/alerts/{alert_id}/destinations) pointed at your service. Every change arrives as an HMAC-signed POST you verify before invoking the graph, so an unauthenticated caller cannot spoof a change. This suits serverless deployments where a long-lived SSE connection is awkward.

from fastapi import FastAPI, Request, HTTPException
from hypeline import verify_signature  # Standard Webhooks HMAC

app = FastAPI()

@app.post("/hypeline")
async def on_change(request: Request):
    body = await request.body()
    if not verify_signature(body, request.headers, secret="whsec_..."):
        raise HTTPException(status_code=401)
    event = await request.json()
    graph.invoke({"event": event})
    return {"ok": True}

Why Hypeline, not a page-change checker

• Deduplicated on extracted content. A simhash fingerprint over the extracted main text separates real edits from template, ad, and timestamp churn, so your graph is not woken by noise.
• One schema across every source. Feeds, watched pages, and streaming sources all arrive as the same versioned event, so the graph has no per-source branching.
• Cursor resume. Reconnect with your last cursor after a deploy or crash and the stream replays in order. Nothing is lost while the agent was away.
• Signed events. Each event carries a content-layer Ed25519 signature, and webhooks are HMAC-signed, so the agent can trust what it acts on.