How to Build a Stateful Web Research Agent with ZenRows and LangGraph

Imagine a LangGraph research agent monitoring technology news for live web data. The scrape node retrieves an article, the extraction node pulls the key facts, and the synthesis node turns them into a briefing. This workflow looks reliable until the scrape node hits a protected site. Instead of article content, it returns a challenge page, and LangGraph stores that response in state as if it were valid content. That is the problem this tutorial solves.

In this tutorial, you'll build a stateful web research agent with LangGraph and ZenRows. The agent will scrape news articles into graph state, retry failed retrievals through a conditional branch, extract structured fields, and generate a briefing from multiple sources. ZenRows retrieves valid content from protected, JavaScript-rendered sites, reducing how often the retry branch needs to run. By the end, you'll have a three-node research graph that routes retrieval failures into a retry branch before bad data reaches downstream nodes.

Why a Failed Web Scraping Node is Worse Inside a Graph Than in a Script

A failed scraping node corrupts every downstream step in a LangGraph workflow. In a traditional scraping script, a failed request fails at one spot. You read the traceback, fix the error, and rerun. A LangGraph node reads from a shared state object and writes its output back into that same state, so when a scraping node retrieves a challenge page instead of the expected content, the graph does not stop.

Downstream nodes read the corrupted state as valid input, run their LLM (Large Language Model) calls and tool actions, and produce output that amplifies the original mistake. By the time the failure surfaces, tracing it back is difficult because the chain of causation now spans several nodes. Without a retry branch, the graph checkpoints the bad node as successful, and one scraping failure becomes a total workflow failure.

This tutorial adds a fallback conditional edge that catches bad output before it propagates and makes the agent reliable in production. ZenRows provides reliable web access that reduces how often the retry branch fires, since premium proxies and JavaScript rendering handle most bot-protected pages on the first call.

Adding ZenRows as a Scraping Node

The simplest way to add web scraping to a LangGraph agent is to wrap a ZenRows retrieval call inside a graph node and write the returned Markdown into state. This section builds that retrieval layer. You'll define the graph state, create a ZenRows scraping node, and compile a graph that stores the returned Markdown in the results field.

Prerequisites

To follow along, you'll need to create a ZenRows account to get your API key and an OpenAI account with billing enabled for LLM calls.

Step 1: Install the required packages

Install LangGraph, OpenAI, ZenRows LangChain integration, and Python dotenv.

pip install langgraph langchain-openai langchain-zenrows python-dotenv

Step 2: Add your ZenRows and OpenAI API keys

Create a .env file in your project root so LangGraph can read both your ZenRows key for retrieval and your OpenAI key for the later LLM nodes.

ZENROWS_API_KEY=<YOUR_ZENROWS_API_KEY>
OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>

Step 3: Define the graph state

Every LangGraph node in this tutorial reads from and writes to the same state object. Define a ResearchState TypedDict with one field per data point the graph needs to track across nodes.

Create a state.py file:

from typing import TypedDict


class ResearchState(TypedDict):
    urls: list[str]
    results: list[str]
    extracted: list[dict]
    briefing: str
    current_url: str
    status: str
    retry_count: int

Here's what each field in the ResearchState schema does:

• urls stores the URLs the agent needs to research.
• results stores the Markdown ZenRows returns.
• extracted stores structured article data.
• briefing stores the final synthesized report.
• current_url tracks the URL currently being processed.
• status records whether retrieval succeeded.
• retry_count tracks how many times the graph has retried a failed retrieval.

Keeping retrieval metadata in state makes debugging easier because every node can see exactly what happened during the previous step.

Step 4: Set up the ZenRows scraping node

This node uses ZenRowsUniversalScraper from langchain-zenrows as a LangChain tool inside LangGraph. You'll activate the premium_proxy and js_render parameters to access dynamic contentdynamic content. The node retrieves a page with ZenRows, converts it to Markdown, and writes the result back into state.

Create a nodes.py file:

from dotenv import load_dotenv
from langchain_zenrows import ZenRowsUniversalScraper

from state import ResearchState

# load environment variables
load_dotenv()

# initialize the scraper
scraper = ZenRowsUniversalScraper()

def scrape_node(state: ResearchState) -> ResearchState:
    # retrieve the current page
    content = scraper.invoke({
        "url": state["current_url"],
        "js_render": True,
        "premium_proxy": True,
        "response_type": "markdown",
    })

    # write the result into graph state
    return {
        **state,
        "results": state["results"] + [content],
        "status": "ok",
    }

The node always returns status: "ok". It has no way yet to tell a real article apart from a blocked page.

Step 5: Build the graph

Create a graph.py file for the first version of the workflow. Register the scrape node as the entry point, and connect it straight to END to verify the retrieval step before adding more branches.

from langgraph.graph import END
from langgraph.graph import StateGraph

from nodes import scrape_node
from state import ResearchState

# create the graph object
graph = StateGraph(ResearchState)

# register nodes
graph.add_node("scrape", scrape_node)

# define execution flow
graph.set_entry_point("scrape")
graph.add_edge("scrape", END)

# compile the graph
app = graph.compile()

At this point, the graph contains one node and one transition.

Step 6: Run the graph

Invoke the graph with a target URL.

result = app.invoke({
    "urls": [],
    "results": [],
    "current_url": "https://www.scrapingcourse.com/ecommerce/",
    "status": "",
})
print(result)

Then run it:

python graph.py

State after execution:

{
    'urls': [],
    'results': [
        "[Skip to navigation]...\n\n# Shop\n\nShowing 1-16 of 188 results...\n\n- **Abominable Hoodie** $69.00..."
    ],
    'current_url': 'https://www.scrapingcourse.com/ecommerce/',
    'status': 'ok'
}

The output is truncated for brevity. Congratulations! The retrieval node works. Next, you'll add retrieval validation and a retry branch so the graph can recover when a target page returns blocked content instead of usable data.

Adding a Retry Branch for Failed Retrievals in AI Agents

A retry branch prevents bad retrievals from corrupting the entire graph. The retrieval node you built earlier assumes every request succeeds. However, a page can return a CAPTCHA, a Cloudflare challenge, or an empty response even when the request itself completes successfully. From LangGraph's perspective, that response is still data. Unless you validate the retrieval result, the graph has no way to distinguish an article from a challenge page. The solution is to make retrieval quality part of the graph's routing logic.

Step 1: Detect blocked pages

Define a list of phrases that signal a blocked page. Then create a function in nodes.py that checks for them.

BLOCK_SIGNALS = [
    "Checking your browser",
    "Enable JavaScript and cookies to continue",
    "Verify you are human",
]



def is_blocked(content: str) -> bool:
    # empty responses are unusable
    if not content.strip():
        return True

    # common challenge page signals
    return any(signal in content for signal in BLOCK_SIGNALS)

is_blocked() doesn't yet change what the scrape node does with that information.

Step 2: Update the scrape node

Update the scrape node to call is_blocked() and store the result as a status, rather than assuming every retrieval succeeded.

def scrape_node(state: ResearchState) -> ResearchState:
    # scrape the current URL 
    content = scraper.invoke(
        {
            "url": state["current_url"],
            "js_render": True,
            "premium_proxy": True,
            "response_type": "markdown",
        }
    )

    status = "blocked" if is_blocked(content) else "ok"

    return {
        **state,
        "results": state["results"] + [content],
        "status": status,
    }

The only change to the scrape node is the status field. It gives the graph enough information to make routing decisions later.

Step 3: Create a retry node

Create a dedicated retry node in nodes.py. The node will retry with a different proxy origin instead of repeating the identical request. If the second retrieval still returns blocked content, the node logs the URL before advancing so you can identify which source introduced low-quality data into the workflow.

def retry_node(state: ResearchState) -> ResearchState:
    # retry the failed retrieval with a different proxy origin
    content = scraper.invoke(
        {
            "url": state["current_url"],
            "js_render": True,
            "premium_proxy": True,
            "proxy_country": "us",
            "response_type": "markdown",
        }
    )
    status = "blocked" if is_blocked(content) else "ok"

    if status == "blocked":
        print(
            f"Retry failed for {state['current_url']}. "
            "Advancing with blocked content."
        )

    existing = list(state["results"])
    if existing:
        existing[-1] = content
    else:
        existing.append(content)

    return {
        **state,
        "results": existing,
        "status": status,
        "retry_count": state["retry_count"] + 1,
    }

This graph uses a single retry attempt. The retry node replaces the last retrieval, or appends if there's nothing yet to replace, so the graph never ends up with two conflicting results for the same URL.

Step 4: Add a routing function

Define a check_retrieval function that decides whether the graph should retry or continue.

def check_retrieval(state: ResearchState) -> str:
    if state["status"] == "blocked" and state["retry_count"] < 1:
        return "retry"

    return "continue"

Step 5: Register the conditional edge

Update the graph.py file. Register the retry node, then add a conditional edge that routes the scrape node's output through check_retrieval so the graph can branch based on the previous step's decision.

from langgraph.graph import StateGraph
from langgraph.graph import END

from state import ResearchState


from nodes import (
    scrape_node,
    retry_node,
    check_retrieval,
)

graph = StateGraph(ResearchState)

graph.add_node("scrape", scrape_node)
graph.add_node("retry", retry_node)

graph.set_entry_point("scrape")

# route to retry or end based on what check_retrieval returns
graph.add_conditional_edges(
    "scrape",
    check_retrieval,
    {
        "retry": "retry",
        "continue": END,
    },
)

graph.add_edge("retry", END)

app = graph.compile()

The graph now contains two possible paths. Instead of treating retrieval quality as an implementation detail inside a function, the graph treats it as part of the execution flow.

Step 6: Test the retry path

Run the graph against a protected target.

result = app.invoke(
    {
        "urls": [],
        "results": [],
        "current_url": "https://www.scrapingcourse.com/cloudflare-challenge",
        "status": "",
        "retry_count": 0,
    }
)
print(result)

Run it:

python graph.py

Cloudflare's challenge page may not trigger on every request because ZenRows resolves most challenges automatically. The example code block below shows what a blocked retrieval looks like when it occurs, so you can see what is_blocked() checks for.

State after scrape_node blocked retrieval:

#Illustrative blocked response, not a ZenRows result
{
    "urls": [],
    "results": [
        "Enable JavaScript and cookies to continue"
    ],
    "current_url": "https://www.scrapingcourse.com/cloudflare-challenge",
    "status": "blocked",
    "retry_count": 0
}

When the retry node runs and fails again, the graph updates its state before advancing. You will get this output:

Retry failed for https://www.scrapingcourse.com/cloudflare-challenge. Advancing with blocked content.

{
    "urls": [],
    "results": [
        "Enable JavaScript and cookies to continue"
    ],
    "current_url": "https://www.scrapingcourse.com/cloudflare-challenge",
    "status": "blocked",
    "retry_count": 1
}

Here's what running the graph against the target actually returned. ZenRows resolved the challenge on the first attempt, so status came back "ok" and the retry node never had to run.

State after scrape_node successful retrieval:

{
    'urls': [],
    'results': [
        '[![](https://www.scrapingcourse.com/assets/images/logo.svg) Scraping Course](http://www.scrapingcourse.com/)\n\n# Cloudflare Challenge\n\n![](https://www.scrapingcourse.com/assets/images/challenge.svg)\n\n## You bypassed the Cloudflare challenge! :D'
    ],
    'current_url': 'https://www.scrapingcourse.com/cloudflare-challenge',
    'status': 'ok',
    'retry_count': 0
}

The graph now distinguishes between a successful request and a successful retrieval. Next, you'll connect this retrieval logic to extraction and synthesis nodes so the graph can turn raw content into a briefing.

Building the Full Three-Node Research Graph

The graph you've built so far can retrieve content and recover from blocked pages. In this section, you'll build a news monitoring workflow with three nodes:

• Scrape retrieves article content with ZenRows Universal Scraper API.
• Extract uses an AI model to extract structured data from each article.
• Synthesize combines those fields into a briefing.

The execution flow looks like this:

Step 1: Update nodes.py

Add the model and JSON imports to nodes.py and initialize the model. This tutorial uses gpt-4o-mini. You can swap it with your preferred LLM.

import json

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_zenrows import ZenRowsUniversalScraper

from state import ResearchState

load_dotenv()

# initialize the scraper and model
scraper = ZenRowsUniversalScraper()
llm = ChatOpenAI(model="gpt-4o-mini")

Step 2: Update the ZenRows scraping node

The first version of the graph scraped a single URL. A research workflow needs to process a list of sources. You'll update scrape_node to loop through urls and keep each result in state.

def scrape_node(state: ResearchState) -> ResearchState:
    results = []

    for url in state["urls"]:
        content = scraper.invoke(
            {
                "url": url,
                "js_render": True,
                "premium_proxy": True,
                "response_type": "markdown",
            }
        )

        results.append(content)

    status = (
        "blocked"
        if any(is_blocked(content) for content in results)
        else "ok"
    )

    return {
        **state,
        "results": results,
        "status": status,
    }

The graph now works through the full urls list, and current_url only matters to the single-page version you built earlier.

Step 3: Update the retry node

Update the retry node to loop through each result and retry only the ones is_blocked() flags, instead of retrying the entire batch.

def retry_node(state: ResearchState) -> ResearchState:
    results = list(state["results"])

    for index, content in enumerate(results):
        if not is_blocked(content):
            continue

        retried_content = scraper.invoke(
            {
                "url": state["urls"][index],
                "js_render": True,
                "premium_proxy": True,
                "proxy_country": "us",
                "response_type": "markdown",
            }
        )

        # log failed retries before advancing
        if is_blocked(retried_content):
            print(
                f"Retry failed for {state['urls'][index]}. "
                "Advancing with blocked content."
            )

        results[index] = retried_content

    status = (
        "blocked"
        if any(is_blocked(content) for content in results)
        else "ok"
    )

    return {
        **state,
        "results": results,
        "status": status,
        "retry_count": state["retry_count"] + 1,
    }

The retry node now retries selectively.

Step 4: Create the extraction node

The extraction node converts Markdown into structured data. For each article, extract_node asks the model to return a fixed set of fields in structured JSON format. LLMs sometimes wrap their JSON response in markdown fences even when told not to. The cleaning step strips those out before parsing. Add this node to the same nodes.py file.

def extract_node(state: ResearchState) -> ResearchState:
    extracted = []

    for i, content in enumerate(state["results"]):
        response = llm.invoke(
            f"""
Extract the following fields from this article.

Return valid JSON only. No markdown fences, no preamble.

{{
  "headline": "...",
  "source": "...",
  "date": "...",
  "summary": "..."
}}

Article:

{content[:4000]}
"""
        )

        cleaned = (
            response.content
            .replace("```

json", "")
            .replace("

```", "")
            .strip()
        )

        try:
            extracted.append(json.loads(cleaned))
        except json.JSONDecodeError:
            print(
                f"JSON parse failed for article {i}. "
                f"Raw response:\n{cleaned[:300]}"
            )
            extracted.append({
                "headline": "Parse error",
                "source": state["urls"][i] if i < len(state["urls"]) else "unknown",
                "date": "",
                "summary": "Extraction failed. The model returned malformed JSON.",
            })

    return {
        **state,
        "extracted": extracted,
    }

Each article becomes a dictionary of structured fields for the synthesis node to read. The prompt only sends the model the first 4,000 characters of each article, since large news pages are full of navigation menus, related links, and footer content that add tokens without improving extraction quality. A cleaning step strips markdown fences before parsing, and any response that still fails to parse becomes a placeholder naming the source URL instead of crashing the run.

Step 5: Create the synthesis node

The synthesis node extracts the structured fields from each article and compiles them into a briefing. Create it in the same nodes.py.

def synthesize_node(state: ResearchState) -> ResearchState:
    response = llm.invoke(
        f"""
Write a concise news briefing.

Requirements:
- One bullet per story
- Preserve the key facts
- Keep each bullet under 40 words

Articles:

{json.dumps(state["extracted"], indent=2)}
"""
    )

    return {
        **state,
        "briefing": response.content,
    }

The scraping, extraction, and synthesis nodes are independent, so the agent is easier to inspect.

Step 6: Assemble the graph

Now update graph.py to build your complete LangGraph web research agent. Register the new nodes and connect them to the retrieval workflow.

from langgraph.graph import StateGraph
from langgraph.graph import END

from state import ResearchState


from nodes import (
    scrape_node,
    retry_node,
    check_retrieval,
    extract_node,
    synthesize_node,
)

graph = StateGraph(ResearchState)

# register nodes
graph.add_node("scrape", scrape_node)
graph.add_node("retry", retry_node)
graph.add_node("extract", extract_node)
graph.add_node("synthesize", synthesize_node)

# define entry point 
graph.set_entry_point("scrape")

# route blocked retrievals through retry branch
graph.add_conditional_edges(
    "scrape",
    check_retrieval,
    {
        "retry": "retry",
        "continue": "extract",
    },
)

# continue graph execution
graph.add_edge("retry", "extract")
graph.add_edge("extract", "synthesize")
graph.add_edge("synthesize", END)

# compile graph
app = graph.compile()

Retrieval failures remain isolated from the rest of the workflow. The graph only advances to extraction after the retrieval phase finishes.

Step 7: Run the research agent

Test the graph against three technology-focused news sources: CNET, Ars Technica, and TechCrunch.

result = app.invoke(
    {
        "urls": [
            "https://www.cnet.com",
            "https://arstechnica.com",
            "https://techcrunch.com",
        ],
        "results": [],
        "extracted": [],
        "briefing": "",
        "current_url": "",
        "status": "",
        "retry_count": 0,
    }
)

print("\nExtracted fields:")
print(result["extracted"])

print("\nFinal briefing:")
print(result["briefing"])

print("\nStatus:", result["status"])
Run it:
python graph.py

Let's look at what the state contains after the extraction and synthesis stages.

State after extraction:

[{'headline': 'Wearables Have a Huge Weakness. What Can We Do?', 'source': 'CNET', 'date': '2026', 'summary': 'This article discusses the significant weaknesses of wearable technology and explores potential solutions to address these issues.'}, {'headline': 'How a massive password-cracking operation spilled credentials for thousands of orgs', 'source': 'Ars Technica', 'date': '2026-06-16', 'summary': 'A password-cracking operation has compromised the credentials of thousands of organizations, including major companies like Oracle, Lenovo, and FedEx.'}, {'headline': 'The slowtech revolution is here to kill your phone addiction and rescue your attention span', 'source': 'TechCrunch', 'date': '2026-06-17', 'summary': "The article discusses the emerging slowtech movement which aims to combat smartphone addiction and improve users' attention spans."}]

State after synthesis:

- A CNET article highlights the significant weaknesses of wearable technology and suggests potential solutions to address these vulnerabilities.
- Ars Technica reports a major password-cracking operation has compromised credentials of thousands of organizations, including Oracle, Lenovo, and FedEx.
- TechCrunch discusses the slowtech movement, aimed at reducing smartphone addiction and enhancing users' attention spans.

Your output will differ depending on the articles available when you run the agent. However, the state transition should remain consistent. Congratulations! You now have a stateful LangGraph research agent that retrieves web content with ZenRows, routes around retrieval failures, gathers information, and produces a final briefing from multiple websites.

Wrapping Up

This step-by-step tutorial covered how to build a three-node LangGraph web research agent with ZenRows that handles its own failures. The pattern you built here is the foundation of any production research or market monitoring agent that depends on live web data.

Now, you know:

• How to add a ZenRows scraping node to a LangGraph StateGraph
• How to detect a blocked retrieval and route it through a retry branch with add_conditional_edges
• How to build a multi-node graph that scrapes, extracts, and synthesizes real news sources into a briefing

The retry branch you built catches bad data before it spreads. ZenRows handles the access layer, so the graph only processes valid data. Try ZenRows for free and see how your research agent performs on protected pages.