Building a web-research AutoGen team with tool use

Building an autogen web research agent team tool use pipeline forces you to confront real engineering problems: tool schemas, agent handoffs, and model reliability. This guide walks through a working implementation using AutoGen’s AgentChat API, live web tools, and a verifiable run loop you can drop into a service.

Step 1: Install the correct AutoGen packages

AutoGen split into autogen-core, autogen-agentchat, and autogen-ext in the 0.4 line. Install the chat and extension layers plus a search client:

pip install autogen-agentchat autogen-ext ddgs beautifulsoup4 requests

Pin versions in production. The AgentChat API is stable enough for internal tooling but still moves between minors.

Step 2: Define web research tools with explicit schemas

AutoGen tools are just async functions wrapped in FunctionTool. The model sees the function signature and docstring. Keep the docstring precise—vague descriptions produce garbage arguments.

from autogen_core.tools import FunctionTool
from typing import List
import asyncio

async def web_search(query: str, max_results: int = 3) -> str:
    """Search the public web for a query and return a list of title/url/snippet dicts.
    
    Args:
        query: The search string.
        max_results: Number of results to return (1-5).
    """
    from ddgs import DDGS
    with DDGS() as d:
        results = d.text(query, max_results=max_results)
    return str(results)

async def fetch_page(url: str) -> str:
    """Download a URL and return cleaned text content (max 5000 chars)."""
    import requests
    from bs4 import BeautifulSoup
    r = requests.get(url, timeout=10, headers={"User-Agent": "research-bot/1.0"})
    r.raise_for_status()
    soup = BeautifulSoup(r.text, "html.parser")
    for tag in soup(["script", "style"]):
        tag.decompose()
    text = " ".join(soup.get_text().split())
    return text[:5000]

search_tool = FunctionTool(web_search, description="Search the web")
fetch_tool = FunctionTool(fetch_page, description="Fetch and clean a web page")

The FunctionTool wrapper auto-generates the JSON schema from type hints. If you pass max_results as a string from the model, AutoGen coerces it—but only if the hint is correct.

Step 3: Configure an OpenAI-compatible model client

AutoGen’s OpenAIChatCompletionClient speaks the standard OpenAI API. Point it at any compliant gateway. For example, n4n.ai exposes one OpenAI-compatible endpoint that fronts 240+ models and handles provider fallback when a backend is rate-limited, so you avoid writing retry logic per provider.

from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",
    api_key="YOUR_KEY",  # use env var in real code
    temperature=0.2,
)

If you run this against OpenAI directly, swap base_url to the default. The rest of the code is identical.

Step 4: Build the agent team with role separation

A research team works best when one agent gathers and another synthesizes. Use RoundRobinGroupChat to alternate speakers. The researcher gets the tools; the writer does not.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    tools=[search_tool, fetch_tool],
    system_message=(
        "You are a meticulous researcher. Use web_search to find sources, "
        "then fetch_page to read them. Cite URLs in your response. "
        "When you have enough, hand off to the writer with 'TERMINATE'."
    ),
)

writer = AssistantAgent(
    name="writer",
    model_client=model_client,
    system_message=(
        "You are a technical writer. Take the researcher's findings and "
        "produce a 3-bullet summary with source links. End with 'TERMINATE'."
    ),
)

termination = TextMentionTermination("TERMINATE")
team = RoundRobinGroupChat([researcher, writer], termination_condition=termination)

The TextMentionTermination stops the loop when either agent emits the token. Without it, the round-robin runs forever.

Step 5: Run a task and stream messages

AutoGen supports run_stream for incremental output. Consume the stream and log every message type—you need this for debugging tool calls.

import asyncio

async def main():
    task = "What are the latest open-source LLM inference engines released in 2025?"
    stream = team.run_stream(task=task)
    async for message in stream:
        if message.type == "ToolCallRequestEvent":
            print(f"[toolcall] {message.content}")
        elif message.type == "ToolCallExecutionEvent":
            print(f"[toolresult] {message.content[:200]}...")
        elif message.type == "TextMessage":
            print(f"[{message.source}] {message.content}")

asyncio.run(main())

Expect the researcher to emit a ToolCallRequestEvent for web_search, receive results, possibly call fetch_page, then yield a text handoff. The writer then responds and terminates.

Step 6: Verify success programmatically

A manual print is not a test. Wrap the run in an assertion that tools were actually called and the writer produced a summary.

async def verify_run():
    stream = team.run_stream(task="Summarize three recent vector DB benchmarks")
    tool_called = False
    writer_spoke = False
    async for msg in stream:
        if msg.type == "ToolCallExecutionEvent":
            tool_called = True
        if msg.type == "TextMessage" and msg.source == "writer":
            writer_spoke = True
            assert "http" in msg.content, "writer must cite sources"
    assert tool_called, "researcher never called a tool"
    assert writer_spoke, "writer never produced output"

# asyncio.run(verify_run())

If tool_called is false, the researcher ignored its tools—usually because the system prompt was too weak or the model temperature too high. Drop temperature to 0.0 and re-test.

Step 7: Handle failures and rate limits

Web tools fail. Wrap fetch_page in try/except and return a structured error string so the model can recover:

async def fetch_page(url: str) -> str:
    try:
        # ... same as before
    except Exception as e:
        return f"ERROR fetching {url}: {type(e).__name__}"

The model will see the error and can pick another source. For model-level limits, the gateway fallback mentioned in Step 3 covers provider outages, but you should still catch ModelClientError around team.run_stream and retry with a different model string if your client supports it.

Step 8: Production notes

Run the team inside a worker queue, not a synchronous request handler. AutoGen’s agents are async; block them on a thread and you lose concurrency. Persist the ToolCallExecutionEvent payloads to your audit log—they are the only record of what the agent actually retrieved.

If you need per-token cost tracking, the gateway’s usage metering (available on the n4n.ai endpoint) returns usage in the standard OpenAI response shape; read it from model_client after each turn.

The autogen web research agent team tool use pattern scales to more agents—add a fact-checker that re-fetches claims, or a router that picks the search tool based on query language. The scaffolding above is the minimum that works end to end.

Verify your setup

You have a complete pipeline when:

1. pip install completes without conflicts.
2. verify_run() passes—tool called, writer cites URLs.
3. Streaming logs show ToolCallRequestEvent followed by ToolCallExecutionEvent.
4. Swapping base_url to another OpenAI-compatible endpoint requires no code changes.

Anything less means the agent is hallucinating or the schema is wrong. Fix the tool docstring first.