Most browser agent tutorials show you a single agent completing a task. Production systems look nothing like that. Real workflows branch, retry, hand off between specialized sub-agents, and run dozens of sessions in parallel. LangGraph gives you the graph-based control flow to build these systems; Anchor gives you the browser infrastructure to run them reliably at scale.
This post walks through a concrete pattern: a LangGraph workflow where multiple Anchor browser sessions fan out in parallel, with conditional routing based on what each agent finds.
Why LangGraph for Browser Agents?
LangGraph models your agent workflow as a directed graph where nodes do work and edges route between them. This maps naturally to browser automation patterns:
- A routing node decides which sub-agents to spawn based on the task
- Browser nodes each own an isolated Anchor session and execute a specific goal
- A reducer node aggregates results across parallel branches
- Conditional edges handle retries, fallbacks, and early exits
Compare this to a flat Python async loop: with LangGraph you get built-in state management, resumability, and a visual graph you can inspect — without writing orchestration boilerplate yourself.
Setup
pip install langgraph langchain-anthropic anchor-browser playwrightSet your API keys:
export ANCHOR_API_KEY="..."
export ANTHROPIC_API_KEY="..."Defining the Graph State
LangGraph passes a shared state object between nodes. For a browser agent graph, we track tasks, results, and errors:
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
class BrowserAgentState(TypedDict):
tasks: list[dict] # Goals for each browser agent
results: Annotated[list, operator.add] # Merged across parallel branches
errors: Annotated[list, operator.add]
retry_count: int
final_report: str
The Annotated[list, operator.add] tells LangGraph to merge lists from parallel branches rather than overwrite them — essential when multiple browser nodes write results simultaneously.
The Browser Node
Each browser node creates an isolated Anchor session, executes its task, and returns structured results:
import os
import asyncio
from anchor_browser import AnchorClient
from langchain_anthropic import ChatAnthropic
anchor = AnchorClient(api_key=os.environ["ANCHOR_API_KEY"])
llm = ChatAnthropic(model="claude-opus-4-8")
async def browser_node(config: dict) -> dict:
task = config["task"]
session = anchor.sessions.create(
proxy_country="us",
options={"adblock": True}
)
try:
playwright = session.get_playwright()
page = playwright.chromium.connect_over_cdp(
session.ws_endpoint
).contexts[0].pages[0]
await page.goto(task["url"])
content = await page.inner_text("body")
response = await llm.ainvoke([{
"role": "user",
"content": f"Task: {task['goal']}
Page content:
{content[:5000]}"
}])
return {
"results": [{"url": task["url"], "output": response.content}],
"errors": []
}
except Exception as e:
return {
"results": [],
"errors": [{"url": task["url"], "error": str(e)}]
}
finally:
session.close()
Wiring the Graph with Fan-Out
LangGraph's Send API lets you dynamically spawn one node per task — all running in parallel Anchor sessions:
from langgraph.constants import Send
def route_tasks(state: BrowserAgentState):
"""Fan out: one browser_node per task, all run in parallel."""
return [
Send("browser_agent", {"task": task})
for task in state["tasks"]
]
def check_errors(state: BrowserAgentState) -> str:
"""Retry on failures, up to 2 times."""
has_failures = len(state.get("errors", [])) > 0
under_limit = state.get("retry_count", 0) < 2
return "retry" if has_failures and under_limit else "aggregate"
async def aggregate_results(state: BrowserAgentState) -> dict:
"""Summarize all browser agent outputs into one report."""
combined = "
".join(
f"[{r['url']}]
{r['output']}" for r in state["results"]
)
summary = await llm.ainvoke([{
"role": "user",
"content": f"Synthesize these research findings into a concise report:
{combined}"
}])
return {"final_report": summary.content}
# Build the graph
graph = StateGraph(BrowserAgentState)
graph.add_node("browser_agent", browser_node)
graph.add_node("aggregate", aggregate_results)
graph.set_conditional_entry_point(route_tasks)
graph.add_conditional_edges(
"browser_agent",
check_errors,
{"retry": "browser_agent", "aggregate": "aggregate"}
)
graph.add_edge("aggregate", END)
app = graph.compile()
Running It
import asyncio
async def main():
result = await app.ainvoke({
"tasks": [
{"url": "https://competitor-a.com/pricing", "goal": "Extract all pricing tiers and features"},
{"url": "https://competitor-b.com/pricing", "goal": "Extract all pricing tiers and features"},
{"url": "https://competitor-c.com/pricing", "goal": "Extract all pricing tiers and features"},
],
"results": [],
"errors": [],
"retry_count": 0,
"final_report": ""
})
print(result["final_report"])
asyncio.run(main())
Three Anchor sessions spin up in parallel, each visits a different URL, and the results merge into a single competitive-pricing report. The whole flow completes in roughly the time of the slowest individual task — not the sum of all three.
What This Pattern Gives You
LangGraph handles control flow, state, and agent coordination. Anchor handles browser reliability, session isolation, anti-bot resilience, and fingerprint management. Neither tries to be the other.
In practice this means your graph logic stays clean — you write nodes and edges, not retry-on-bot-detection code. When Anchor's infrastructure handles a Cloudflare challenge or an unexpected redirect, your graph just sees a successful result. When a session genuinely fails, the conditional edge retries it automatically without restarting the full graph.
For teams moving beyond single-task scripts, this separation of concerns is what holds up under real production workloads — dozens of agents, diverse targets, and complex branching logic all managed cleanly in one place.
Build your first multi-agent browser workflow — start with Anchor for free →
