LlamaIndex AgentWorkflow + Anchor: Event-Driven Browser Agents in Python

LlamaIndex shipped AgentWorkflow — a pre-wired, event-driven agent loop built on top of their Workflows primitives. Unlike frameworks that hide the execution graph, AgentWorkflow surfaces every step as a typed event you can inspect, replay, and test independently. Combined with Anchor's managed cloud browsers, you get session isolation, stealth fingerprinting, and proxy routing out of the box — without touching infrastructure.

This guide builds a competitive intelligence pipeline: it searches the web, fetches pricing pages, and synthesises findings into a typed Pydantic report using two cooperating agents.

Why LlamaIndex AgentWorkflow?

1. Event-driven step graph — each agent action is a typed Event object, making debugging and replay straightforward without extra instrumentation.
2. First-class multi-agent handoffs — a root agent can pass control to specialised sub-agents, letting you route expensive reasoning to capable models and cheap parsing to faster ones.
3. Native tool validation — tools return typed objects that the planner acts on without string munging, which eliminates an entire class of agent hallucination.

Setup

pip install llama-index-core llama-index-llms-openai anchor-browser playwright
playwright install chromium

export ANCHOR_API_KEY="your_anchor_key"
export OPENAI_API_KEY="your_openai_key"

Defining Browser Tools

Each tool spins up its own isolated Anchor session. Session isolation means cookies, localStorage, and browser fingerprints never bleed across tasks — critical when hitting the same domain from different logical agents.

import os
import asyncio
from anchor_browser import AnchorClient
from llama_index.core.tools import FunctionTool

anchor = AnchorClient(api_key=os.environ["ANCHOR_API_KEY"])


async def fetch_page_text(url: str) -> str:
    """Fetch a URL and return the visible body text (first 6 000 chars)."""
    async with anchor.sessions.create(
        proxy_type="datacenter",
        iso_country_code="US",
        isolated=True,
    ) as session:
        page = await session.get_page()
        await page.goto(url, wait_until="networkidle")
        # innerText strips tags — cuts token usage ~80% vs page.content()
        text = await page.evaluate("document.body.innerText")
        return text[:6000]


async def search_web(query: str) -> list[str]:
    """Search Google and return the top 5 non-Google result URLs."""
    async with anchor.sessions.create(
        proxy_type="datacenter",
        iso_country_code="US",
        isolated=True,
    ) as session:
        page = await session.get_page()
        await page.goto(
            f"https://www.google.com/search?q={query}",
            wait_until="networkidle",
        )
        urls: list[str] = await page.evaluate("""
            Array.from(document.querySelectorAll('a[href^="http"]'))
                .map(a => a.href)
                .filter(h => !h.includes('google.com'))
                .slice(0, 5)
        """)
        return urls


fetch_tool  = FunctionTool.from_defaults(async_fn=fetch_page_text)
search_tool = FunctionTool.from_defaults(async_fn=search_web)

A few things worth noting:

• isolated=True gives each session a clean browser state — no shared fingerprint carryover between agent steps.
• document.body.innerText instead of page.content() reduces HTML noise and cuts LLM token costs significantly on content-heavy pages.
• Anchor handles TLS fingerprinting and user-agent rotation automatically, so no stealth plugin configuration is needed.

Your First AgentWorkflow

from llama_index.core.agent.workflow import AgentWorkflow
from llama_index.llms.openai import OpenAI

research_agent = AgentWorkflow.from_tools(
    tools=[fetch_tool, search_tool],
    llm=OpenAI(model="gpt-4o", temperature=0),
    system_prompt=(
        "You are a competitive intelligence researcher. "
        "Given a company name, search for their pricing page, fetch it, "
        "and return a structured summary of plans and prices. "
        "Always verify data by fetching the source URL before reporting."
    ),
)

async def main():
    result = await research_agent.run(
        "Research pricing for Notion, Linear, and Figma. "
        "For each: find the pricing URL, fetch the page, "
        "and list each plan name with its monthly price."
    )
    print(result)

asyncio.run(main())

Multi-Agent Handoffs

AgentWorkflow's ReActAgent class supports can_handoff_to — a declarative list of agents the current one may delegate to. Use this to keep browsing and reasoning on separate cost tiers:

from llama_index.core.agent.workflow import AgentWorkflow, ReActAgent

# Fast, cheap model handles all browser I/O
browser_agent = ReActAgent(
    name="browser_agent",
    description=(
        "Fetches URLs and returns raw page content. "
        "Hands off to analyst_agent once data is collected."
    ),
    tools=[fetch_tool, search_tool],
    llm=OpenAI(model="gpt-4o-mini"),
    can_handoff_to=["analyst_agent"],
)

# Capable model handles synthesis — no browser access needed
analyst_agent = ReActAgent(
    name="analyst_agent",
    description="Analyses collected web content and produces structured reports.",
    tools=[],
    llm=OpenAI(model="gpt-4o"),
    can_handoff_to=["browser_agent"],
)

pipeline = AgentWorkflow(
    agents=[browser_agent, analyst_agent],
    root_agent="browser_agent",
)

async def run_pipeline():
    return await pipeline.run(
        "Compare Linear and Jira for a 20-person engineering team. "
        "Include pricing, GitHub integration depth, and any AI features."
    )

The browser agent pays gpt-4o-mini rates for the I/O-heavy steps; the analyst pays gpt-4o rates only for the reasoning step. On a 10-page research task, this typically cuts inference cost by 60–70%.

Structured Output with Pydantic

from pydantic import BaseModel

class PricingTier(BaseModel):
    plan_name: str
    monthly_price_usd: float | None
    key_features: list[str]

class CompetitorReport(BaseModel):
    company: str
    pricing_url: str
    tiers: list[PricingTier]

# AgentWorkflow returns a typed object when output_cls is set
result: CompetitorReport = await research_agent.run(
    "Fetch Notion pricing and return structured data.",
    output_cls=CompetitorReport,
)

for tier in result.tiers:
    print(f"{tier.plan_name}: ${tier.monthly_price_usd}/mo")

Production Tips

• Cap concurrency with a semaphore — wrap tool calls in asyncio.Semaphore(5) to stay within your Anchor plan's concurrent session limit.
• Scope sessions to the tool call — the async with anchor.sessions.create() pattern above is correct; never hold sessions open between agent steps.
• Geo-target with iso_country_code — pricing pages often serve different numbers by region; match the country to your users.
• Add retry logic — wrap page.goto in a tenacity retry with exponential backoff; JS-heavy SPAs occasionally time out on first load.

What's Next

The natural extension is running this pipeline against a list of 50 competitors in parallel. LlamaIndex's Workflows engine supports parallel step branches natively, and Anchor scales sessions horizontally without any infrastructure changes on your side.

Deploy the pipeline as a scheduled job and email the structured report every Monday morning — you'll have a live competitive intelligence feed with under 200 lines of Python.

Try Anchor free and run your first LlamaIndex browser agent in minutes →