DEV Community

ZenRows profile image Tiioluwani
Tiioluwani for ZenRows

Posted on Originally published at zenrows.com

Build a Web Research Multi-Agent System with AG2 and Zenrows

#ai #python #agents #tutorial

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/web-research-multi-agent-ag2-zenrows

This guide shows you how to register Zenrows Fetch as a typed tool inside an AG2 group chat pipeline (researcher, analyst, and critic) so the model gets live page content instead of a fetch failure circulating through the shared history. You'll need Python 3.10+, AG2 v0.12.2, an OpenAI API key, and a Zenrows API key.

All the code is on GitHub.

Before you start

Install AG2 with its OpenAI client:

pip install "ag2[openai]==0.12.2"
Enter fullscreen mode Exit fullscreen mode

Two things before you run anything: you install as ag2 but import as autogen in v0.12.2, so import ag2 raises a ModuleNotFoundError. And AG2 doesn't bundle the OpenAI client, which is why [openai] is required.

Load keys from .env:

import os
from dotenv import load_dotenv

import autogen
from autogen import AssistantAgent, GroupChat, GroupChatManager, UserProxyAgent

load_dotenv()

llm_config = {
    "config_list": [{"model": "gpt-4o-mini", "api_key": os.getenv("OPENAI_API_KEY")}],
    "temperature": 0,
}
Enter fullscreen mode Exit fullscreen mode

Set up the agent pipeline

Define four agents. user_proxy executes tools and terminates on TERMINATE. The other three divide the research task.

# executes tools and triggers termination
user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config=False,
    is_termination_msg=lambda msg: "TERMINATE" in (msg.get("content") or ""),
)

# fetches and summarizes web content
researcher = AssistantAgent(
    name="researcher",
    system_message=(
        "You are a research agent. When given a URL, fetch its web content and "
        "summarize what you find. Be factual and note anything that looks "
        "incomplete or missing from the page. Pass your summary to the analyst."
    ),
    llm_config=llm_config,
)

# extracts structured fields from researcher output
analyst = AssistantAgent(
    name="analyst",
    system_message=(
        "You are a data analyst. Extract structured fields, such as plan names "
        "and prices, from the content the researcher provides. Present the "
        "result as a clean list. If the researcher's content has no usable "
        "data, say so explicitly instead of guessing."
    ),
    llm_config=llm_config,
)

# validates analyst output and terminates
critic = AssistantAgent(
    name="critic",
    system_message=(
        "You are a critic. Validate the analyst's structured output against "
        "the researcher's original content. Flag any gaps, missing fields, or "
        "inconsistencies. When the output checks out or no further progress is "
        "possible, say so and end your message with TERMINATE."
    ),
    llm_config=llm_config,
)
Enter fullscreen mode Exit fullscreen mode

Wire them into a group chat:

group_chat = GroupChat(
    agents=[user_proxy, researcher, analyst, critic],
    messages=[],
    max_round=6,
)

manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config)

chat_result = user_proxy.initiate_chat(
    manager,
    message=(
        "Fetch the product listing from this page and extract the "
        "product name and price: https://www.walmart.com/ip/AirPods-Pro-3/17835006350"
    ),
)
Enter fullscreen mode Exit fullscreen mode

For accurate token counts, use autogen.gather_usage_summary(). chat_result.cost always reports zero for group chats.

usage_summary = autogen.gather_usage_summary(
    [user_proxy, researcher, analyst, critic, manager]
)
print(usage_summary)
Enter fullscreen mode Exit fullscreen mode

What happens without a fetch tool

The researcher has no registered tool that can reach a URL. It admits this. That admission enters the shared history, and every agent that follows reads and forwards it:

researcher: I'm unable to access external URLs directly to fetch content.
analyst: I cannot access external URLs or fetch content from them.
critic: Both parties acknowledge the inability to access external URLs. TERMINATE

Total tokens: 990 | Cost: $0.000252
Enter fullscreen mode Exit fullscreen mode

Low cost because nothing was retrieved. The next step changes that.

Register Zenrows as a typed AG2 tool

AG2 builds the tool schema from type hints. Annotated[type, description] on every parameter is required; without it, registration fails.

import requests
from typing import Annotated
from autogen.tools import Tool

ZENROWS_API_KEY = os.getenv("ZENROWS_API_KEY")
ZENROWS_ENDPOINT = "https://api.zenrows.com/v1/"

def fetch_page_content(
    url: Annotated[str, "The target URL to fetch through Zenrows using adaptive stealth mode."],
) -> str:
    """Fetch a URL through Zenrows Fetch and return clean Markdown."""
    params = {
        "url": url,
        "apikey": ZENROWS_API_KEY,
        # mode=auto: Zenrows picks the config each site needs
        # — JS rendering and premium proxies only when required
        "mode": "auto",
        # response_type=markdown: strips HTML so the model reads
        # clean text, not raw markup
        "response_type": "markdown",
    }
    response = requests.get(ZENROWS_ENDPOINT, params=params, timeout=60)
    response.raise_for_status()
    return response.text
Enter fullscreen mode Exit fullscreen mode

Wrap it and register with both agents:

fetch_tool = Tool(
    name="fetch_page_content",
    description=(
        "Fetch a URL through Zenrows and return clean Markdown content from "
        "JavaScript-rendered and anti-bot-protected pages."
    ),
    func_or_tool=fetch_page_content,
)

# exposes the schema to the researcher so it knows to call this tool
fetch_tool.register_for_llm(researcher)
# user_proxy is always the execution agent in a group chat
fetch_tool.register_for_execution(user_proxy)
Enter fullscreen mode Exit fullscreen mode

Update the researcher's system message to name the tool explicitly:

researcher = AssistantAgent(
    name="researcher",
    system_message=(
        "You are a research agent. When given a URL, use the fetch_page_content "
        "tool to retrieve its web content, then summarize what you find. Be "
        "factual and note anything that looks incomplete or missing from the "
        "page. Pass your summary to the analyst."
    ),
    llm_config=llm_config,
)
Enter fullscreen mode Exit fullscreen mode

Full pipeline with Zenrows integrated 

import os
from typing import Annotated

import requests
from dotenv import load_dotenv

import autogen
from autogen import AssistantAgent, GroupChat, GroupChatManager, UserProxyAgent
from autogen.tools import Tool

load_dotenv()

ZENROWS_API_KEY = os.getenv("ZENROWS_API_KEY")
ZENROWS_ENDPOINT = "https://api.zenrows.com/v1/"

llm_config = {
    "config_list": [{"model": "gpt-4o-mini", "api_key": os.getenv("OPENAI_API_KEY")}],
    "temperature": 0,
}

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config=False,
    is_termination_msg=lambda msg: "TERMINATE" in (msg.get("content") or ""),
)

researcher = AssistantAgent(
    name="researcher",
    system_message=(
        "You are a research agent. When given a URL, use the fetch_page_content "
        "tool to retrieve its web content, then summarize what you find. Be "
        "factual and note anything that looks incomplete or missing from the "
        "page. Pass your summary to the analyst."
    ),
    llm_config=llm_config,
)

analyst = AssistantAgent(
    name="analyst",
    system_message=(
        "You are a data analyst. Extract structured fields, such as plan names "
        "and prices, from the content the researcher provides. Present the "
        "result as a clean list. If the researcher's content has no usable "
        "data, say so explicitly instead of guessing."
    ),
    llm_config=llm_config,
)

critic = AssistantAgent(
    name="critic",
    system_message=(
        "You are a critic. Validate the analyst's structured output against "
        "the researcher's original content. Flag any gaps, missing fields, or "
        "inconsistencies. When the output checks out or no further progress is "
        "possible, say so and end your message with TERMINATE."
    ),
    llm_config=llm_config,
)

def fetch_page_content(
    url: Annotated[str, "The target URL to fetch through Zenrows using adaptive stealth mode."],
) -> str:
    """Fetch a URL through Zenrows Fetch and return clean Markdown."""
    params = {
        "url": url,
        "apikey": ZENROWS_API_KEY,
        "mode": "auto",
        "response_type": "markdown",
    }
    response = requests.get(ZENROWS_ENDPOINT, params=params, timeout=60)
    response.raise_for_status()
    return response.text

fetch_tool = Tool(
    name="fetch_page_content",
    description=(
        "Fetch a URL through Zenrows and return clean Markdown content from "
        "JavaScript-rendered and anti-bot-protected pages."
    ),
    func_or_tool=fetch_page_content,
)

fetch_tool.register_for_llm(researcher)
fetch_tool.register_for_execution(user_proxy)

group_chat = GroupChat(
    agents=[user_proxy, researcher, analyst, critic],
    messages=[],
    max_round=6,
)

manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config)

chat_result = user_proxy.initiate_chat(
    manager,
    message=(
        "Fetch the product listing from this page and extract the "
        "product name and price: https://www.walmart.com/ip/AirPods-Pro-3/17835006350"
    ),
)

print("\n===== USAGE SUMMARY (ALL AGENTS) =====")
usage_summary = autogen.gather_usage_summary(
    [user_proxy, researcher, analyst, critic, manager]
)
print(usage_summary)
Enter fullscreen mode Exit fullscreen mode

Results 

researcher: ***** Suggested tool call: fetch_page_content *****
Arguments: {"url":"https://www.walmart.com/ip/AirPods-Pro-3/17835006350"}
>>>>>>>> EXECUTING FUNCTION fetch_page_content...

analyst:
- Product Name: Apple AirPods Pro 3
- Price: $189.99 (was $249.00, you save $59.01)

critic: The analyst's output is consistent with the researcher's content. TERMINATE

Total tokens: 7,236 | Cost: $0.001147
Enter fullscreen mode Exit fullscreen mode
Run Products extracted Total tokens Cost
Baseline (no fetch tool) No 990 $0.000252
Fetch with mode=auto Yes 7,236 $0.001147

The working run spends more tokens because it retrieved actual page content. The baseline spent almost nothing because it retrieved nothing.

Keeping costs predictable

Content volume in the shared history is the main cost driver. Every agent turn re-reads and forwards whatever is in the history, so page size multiplies across pipeline turns.

  • Keep max_round conservative and raise it only when the pipeline consistently needs more turns.
  • Cache the Zenrows response within a session if the same URL is fetched more than once.
  • Use Zenrows Batch for pipelines that research several URLs in one run. It handles concurrency and retries without you managing fetch queues.

What's next

Top comments (0)