| name | custom-agent-creator |
| description | Create and integrate new modular agents for browseruse-bench (BaseAgent in browseruse_bench/agents/, registry import in __init__.py, entry in configs/agent_registry.yaml, runtime config in root config.yaml agents section). Use when a user asks to add/register a new agent, scaffold a custom agent, or implement a new agent based on external docs. |
Custom Agent Creation and Integration
Minimum Inputs
- Confirm Agent name (registry key used in configs/agent_registry.yaml and root config.yaml)
- Confirm Python module name (browseruse_bench/agents/.py)
- Confirm supported benchmarks (e.g., Online-Mind2Web / BrowseComp / LexBench-Browser)
- Collect external docs URL(s); read and extract usable SDK/API details
- Define config keys (model_id, api_key, base_url, timeout, SDK-specific options)
- Agent runtime config lives in root config.yaml under agents:<agent_name>; model/provider
parameters are read from agent_config and injected into SDK constructors directly
Required References
- browseruse_bench/agents/init.py (check existing imports before adding)
- configs/agent_registry.yaml (check existing agent registrations)
Load when needed (do NOT read upfront)
- references/custom-agent-integration.md — SDK integration patterns
- browseruse_bench/agents/base.py — BaseAgent method signatures
- browseruse_bench/agents/registry.py — @register_agent internals
- browseruse_bench/runner/agent_runner.py — runner entrypoint conventions
- browseruse_bench/cli/run.py — extra dependency mapping
- browseruse_bench/utils/venv.py — install_agent_dependencies (supports requirements_file)
- browseruse_bench/agents/browser_use.py — simple agent reference example
- browseruse_bench/agents/agent_tars.py — subprocess-based agent example
- browseruse_bench/agents/skyvern.py — async/CDP agent example
- docs/en/agents/custom-agent.mdx — user-facing docs template
- docs/zh/agents/custom-agent.mdx — Chinese docs template
Standard Workflow
-
Create module:
- browseruse_bench/agents/<agent_module>.py
-
Implement BaseAgent subclass:
- Add from future import annotations
- PEP8 import order: stdlib → third-party → local
- Use typing annotations (Dict, Any, Path)
- Do not use print(); use logger
- Do not use sys.path.insert
- Read all configurable values from agent_config (no hardcoding)
- Pass model/provider parameters from agent_config directly into SDK constructors (avoid relying solely on environment variables)
- If the upstream SDK has a hardcoded/whitelist model check, do not expose that limitation directly to browseruse-bench users when it can be avoided:
use a provider-native bootstrap/default model only for SDK initialization, and after initialization replace the SDK's runtime inference client/engine with your own wrapper that honors the configured
model_id, api_key, and base_url
- Resolve paths with REPO_ROOT when touching files
-
Register module import:
- Add
from browseruse_bench.agents import <agent_module> # noqa: F401 to browseruse_bench/agents/init.py
-
Register agent in configs/agent_registry.yaml:
- path: browseruse_bench/agents
- entrypoint: browseruse_bench/runner/agent_runner.py
- venv: .venv (or .venvs/ if it needs an isolated environment)
- supported_benchmarks: [...]
-
Add runtime config to root config.yaml under agents:<agent_name>:
- active_model: <default_model_key>
- models: { <model_key>: { model_id, api_key, base_url, ... } }
- browser: { browser_id, ... } (if the agent uses a browser)
- defaults: { timeout, max_steps, ... }
-
Add dependencies — two paths depending on whether the agent's deps conflict with core:
Path A — no conflict (happy path):
- Add extra group to pyproject.toml
[project.optional-dependencies]: <agent> = ["sdk>=x.y"]
- Map agent name → extra in browseruse_bench/cli/run.py (the
if/elif extra_name block)
- Registry entry uses only
venv: .venvs/<agent> (no requirements_file)
- Install:
uv venv .venvs/<agent> && uv pip install --python .venvs/<agent> -e ".[<agent>]"
Path B — deps conflict with core project (e.g. older openai, older pydantic):
How to detect a conflict before committing:
Run uv pip install --python .venvs/<agent> -e ".[<agent>]" in a scratch venv first.
If uv says "X and Y are incompatible", use Path B.
-
Optional sanity run:
- uv run bubench run --agent --data --mode first_n --count 1
Minimal Viable Variant (lean BaseAgent.run_task)
- Config keys from the root config.yaml
models/browser/defaults sections are not normalized
at load time — write lowercase in both the YAML and in Python (model_id, base_url, browser_id,
max_steps, timeout, and agent-specific flags like use_vision, flash_mode).
- For common config keys use BaseAgent helpers (defined in
browseruse_bench/agents/base.py,
load when needed): self.get_model_id(agent_config), self.get_timeout(agent_config),
self.get_max_steps(agent_config), self.get_api_key(agent_config, env_var="PROVIDER_API_KEY"),
self.get_base_url(agent_config, env_var=...). These handle key aliases (timeout/timeout_seconds,
max_steps/max_turns, etc.) and provide safe defaults. For agent-specific flags use
agent_config.get("flag_name") directly.
- LLM branch only on
model_type → ChatBrowserUse / ChatOpenAI / ChatGoogle; avoid large parameter maps.
- Browser session lifecycle must go through
open_browser_session(...) from browseruse_bench.browsers.
- Keep provider-specific session create/delete logic in
browseruse_bench/browsers/providers/*.py; do not duplicate this in agent modules.
- In agent code, branch by
session_context.transport (local / cdp / cloud_native) instead of hardcoding provider SDK logic.
- History handling: decode base64 screenshots to
trajectory/, collect basic actions, steps, end-to-end ms; keep result dict small.
- Trajectory screenshots are an expected artifact, not an optional extra:
AgentResult.screenshots
must list the files saved under trajectory/. If the agent framework can capture screenshots,
make the default rules ask for at least one after reaching the main page and one after finding
the answer (see the openclaw/hermes default rules). If capturing is deliberately disabled — for
example the only screenshot path also triggers a vision-model call the bench model cannot serve
(hermes browser_vision) — record that trade-off in the agent module's default rules or a module
comment so a run with zero screenshots is explainable, and still keep the collection code path
working for the day it is re-enabled.
- For provider semantics, routing, workspace, and missing-file edge cases, follow the guidance in
SDK Integration Pitfalls below instead of repeating those rules in the generated agent.
- Error handling: catch specific expected failures (
asyncio.TimeoutError, TimeoutError, ValueError, RuntimeError), set status/error; no bare except.
- Still obey repo rules: no hardcoded secrets/URLs, use
logger, no sys.path.insert, type hints, imports at top, no emojis, resolve paths via REPO_ROOT when touching files.
SDK Integration Pitfalls (learned from SeeAct-style integrations)
- Do not assume an SDK's documented provider support is sufficient for browseruse-bench needs.
Verify whether it also supports:
- custom
base_url
- non-default model IDs
- OpenAI-compatible gateways
- provider-specific API key env vars
- If the SDK reads credentials only from environment variables, isolate that env injection in a narrow setup/restore block.
Restore the previous environment values in
finally so one task run does not leak credentials into another.
- Prefer wrapping the SDK's inference engine/client instead of forking or patching the third-party package in-place.
Keep browseruse-bench-specific provider normalization in our agent module.
- Some SDKs assume task-local artifact subdirectories already exist (
screenshots/, dom/, traces, etc.).
If the SDK writes files under the task workspace, proactively create the expected subdirectories after
initialization and re-check them before repeated callbacks if the SDK is fragile.
- If you use LiteLLM (or a similar router), remember that it may infer the provider from the model name.
Example: a model name like
gemini-2.5-pro may be auto-routed to Vertex/Google even when the user explicitly
configured an OpenAI-compatible gateway. In those cases, set the provider explicitly (for example
custom_llm_provider="openai") instead of relying on model-name inference.
- When wiring official Gemini/Vertex-style multimodal paths, audit runtime-only dependencies required by the
router stack (for example
Pillow, vertexai, Google SDK packages). Add them to the isolated requirements
file when they are genuinely needed by the selected provider path.
- Conversely, do not install provider-specific dependencies just because the model name looks like that provider.
If the request is intentionally routed through an OpenAI-compatible gateway, force the router down the
OpenAI-compatible path instead of allowing it to import unrelated provider SDKs.
- Audit third-party SDK loggers for duplicate terminal output. Some SDKs attach their own
StreamHandler and
also propagate to the root logger, causing every line to appear twice. In our wrapper layer, disable
propagation or remove duplicate console handlers while keeping file handlers when needed.
- After provider routing changes, run a real smoke task per provider path you claim to support
(for example one OPENAI run, one OPENAI_COMPATIBLE run, one GEMINI run if direct Gemini is supported).
Unit tests are necessary but not sufficient for SDK/router integrations.
- Add focused tests for:
- runtime config resolution (
model_type, model_id, api/base selection)
- bootstrap model fallback for SDKs with model whitelists
- OpenAI-compatible/custom gateway routing
- explicit provider override when the model name would otherwise trigger router auto-detection
- task-local artifact directory creation
- missing screenshot/file fallback behavior
- duplicate logger handler deduplication
Adding a New Browser Backend (required checklist)
- Add provider implementation:
browseruse_bench/browsers/providers/<provider>.py- implement
open(agent_name, agent_config) and close(session_context)
- Register backend id:
- update
browseruse_bench/browsers/registry.py
- Dependency and runtime:
- optional provider SDK imports must be
ImportError-safe
- fail only when selected backend is used
- Tests:
- add/extend
tests/browseruse_bench/test_browsers.py
- include open-success, missing-credential/dependency, cleanup-failure-tolerated
- Docs/config updates:
- update
docs/en/agents/*.mdx, docs/zh/agents/*.mdx
- update quickstart env docs if new env vars are required
BaseAgent Template (replace <agent_name> and <agent_module>)
from __future__ import annotations
import logging
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Dict
from browseruse_bench.agents.base import BaseAgent
from browseruse_bench.agents.registry import register_agent
from browseruse_bench.schemas import AgentMetrics, AgentResult
logger = logging.getLogger(__name__)
@register_agent
class <AgentName>(BaseAgent):
name = "<agent_name>"
def run_task(
self,
task_info: Dict[str, Any],
agent_config: Dict[str, Any],
task_workspace: Path,
) -> AgentResult:
task_id = task_info.get("task_id", "unknown")
task_prompt = self.build_task_prompt(task_info)
timeout = self.get_timeout(agent_config)
logger.info("Starting task %s", task_id)
t0 = time.monotonic()
return AgentResult(
task_id=task_id,
timestamp=datetime.now(UTC),
env_status="success",
agent_done="done",
agent_success=True,
answer="",
metrics=AgentMetrics(end_to_end_ms=int((time.monotonic() - t0) * 1000), steps=0),
)
Choose Your Pattern
The BaseAgent Template above covers the sync/SDK case. Two more integration patterns are common; pick one before coding:
| Pattern | Base class | When to use | Example agents |
|---|
| SDK embedding (sync) | BaseAgent | Agent is a Python SDK you call directly, synchronous API | — |
| SDK embedding (async) | BaseAgent + _run_task_async | Agent SDK is async (Playwright, httpx-based) | browser_use.py, openai_cua.py, skyvern.py |
| CLI subprocess | CLIAgent | Agent is an external executable (npm, binary) | agent_tars.py, claude_code.py |
Templates B and C below cover the non-sync variants.
AgentResult Validation Rules
AgentResult uses extra="forbid" — unknown fields cause validation errors.
Constraints enforced by model_validator:
agent_success MUST be None when agent_done != "done"error MUST be None when env_status == "success" AND agent_done == "done"env_status: "success" | "failed"agent_done: "done" | "timeout" | "max_steps" | "error"
Status-setting pattern (use in all templates):
if error_msg and "Timeout" in error_msg:
env_status, agent_done = "success", "timeout"
elif error_msg:
env_status, agent_done = "failed", "error"
elif steps >= max_steps:
env_status, agent_done = "success", "max_steps"
else:
env_status, agent_done = "success", "done"
agent_success = None
if agent_done == "done":
agent_success = bool(final_answer)
Error Handling — SDK Exception Catch (critical)
Templates only catch generic exceptions (RuntimeError, OSError, ValueError, TimeoutError). You MUST also catch the agent SDK's base API exception class, or a 401 / rate-limit / bad-request will crash the task runner and skip result.json.
import <sdk_module>
try:
except <sdk_module>.APIError as exc:
error_msg = f"<SDK> API error: {exc}"
logger.error("Task %s <SDK> API error: %s", task_id, exc)
except (RuntimeError, OSError, ValueError) as exc:
error_msg = str(exc)
logger.error("Task %s error: %s", task_id, exc)
Check the SDK's exception hierarchy (usually <sdk>.APIError or <sdk>.Error as base) and catch the broadest SDK-specific class.
Template B — Async BaseAgent
For SDKs that are async (Playwright, httpx, modern browser-use SDKs). Wrap asyncio.run() in run_task(), put the real logic in _run_task_async().
from __future__ import annotations
import asyncio
import logging
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from browseruse_bench.agents.base import BaseAgent
from browseruse_bench.agents.registry import register_agent
from browseruse_bench.browsers import BrowserSessionContext, open_browser_session
from browseruse_bench.schemas import AgentMetrics, AgentResult, AgentUsage
logger = logging.getLogger(__name__)
@register_agent
class <AgentName>(BaseAgent):
name = "<agent-name>"
def run_task(
self,
task_info: dict[str, Any],
agent_config: dict[str, Any],
task_workspace: Path,
) -> AgentResult:
timeout = self.get_timeout(agent_config)
browser_id = agent_config.get("browser_id", "Chrome-Local")
with open_browser_session(
browser_id=browser_id,
agent_name=self.name,
agent_config=agent_config,
) as session_context:
return asyncio.run(
self._run_task_async(
task_info, agent_config, task_workspace,
timeout, session_context,
)
)
async def _run_task_async(
self,
task_info: dict[str, Any],
agent_config: dict[str, Any],
task_workspace: Path,
timeout: int,
session_context: BrowserSessionContext,
) -> AgentResult:
task_id = task_info["task_id"]
task_prompt = self.build_task_prompt(task_info)
model_id = self.get_model_id(agent_config) or ""
max_steps = self.get_max_steps(agent_config)
trajectory_dir = task_workspace / "trajectory"
trajectory_dir.mkdir(parents=True, exist_ok=True)
t0 = time.monotonic()
error_msg: str | None = None
final_answer = ""
steps = 0
total_input_tokens = 0
total_output_tokens = 0
try:
pass
except (asyncio.TimeoutError, TimeoutError):
error_msg = f"Timeout after {timeout} seconds"
logger.error("Task %s timed out after %d seconds", task_id, timeout)
except (RuntimeError, OSError, ValueError) as exc:
error_msg = str(exc)
logger.error("Task %s error: %s", task_id, exc)
end_to_end_ms = int((time.monotonic() - t0) * 1000)
if error_msg and "Timeout" in error_msg:
env_status, agent_done = "success", "timeout"
elif error_msg:
env_status, agent_done = "failed", "error"
elif steps >= max_steps:
env_status, agent_done = "success", "max_steps"
else:
env_status, agent_done = "success", "done"
agent_success = None
if agent_done == "done":
agent_success = bool(final_answer)
usage_obj = None
if total_input_tokens + total_output_tokens > 0:
usage_obj = AgentUsage(
total_prompt_tokens=total_input_tokens,
total_completion_tokens=total_output_tokens,
)
return AgentResult(
task_id=task_id,
task=task_prompt,
timestamp=datetime.now(UTC),
env_status=env_status,
agent_done=agent_done,
agent_success=agent_success,
answer=final_answer if not error_msg else f"[Task Failed: {error_msg}]",
model_id=model_id,
browser_id=session_context.backend_id,
metrics=AgentMetrics(
end_to_end_ms=end_to_end_ms,
steps=steps,
usage=usage_obj,
),
error=error_msg if env_status == "failed" else None,
)
Template C — CLIAgent Subprocess
For agents that ship as an external executable (npm package, binary). Inherit CLIAgent instead of BaseAgent to get _run_subprocess and _map_exit_status helpers.
from __future__ import annotations
import logging
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from browseruse_bench.agents.cli_agent import CLIAgent
from browseruse_bench.agents.registry import register_agent
from browseruse_bench.browsers import open_browser_session
from browseruse_bench.schemas import AgentMetrics, AgentResult
logger = logging.getLogger(__name__)
@register_agent
class <AgentName>(CLIAgent):
name = "<agent-name>"
def run_task(
self,
task_info: dict[str, Any],
agent_config: dict[str, Any],
task_workspace: Path,
) -> AgentResult:
task_id = task_info["task_id"]
task_prompt = self.build_task_prompt(task_info)
timeout = self.get_timeout(agent_config)
model_id = self.get_model_id(agent_config) or ""
browser_id = agent_config.get("browser_id", "Chrome-Local")
cmd = ["<executable>", "<subcommand>"]
with open_browser_session(
browser_id=browser_id,
agent_name=self.name,
agent_config=agent_config,
) as session_context:
cmd_args: list[str] = []
cmd_args.extend(["--input", task_prompt])
full_cmd = cmd + cmd_args
try:
returncode, stdout_lines, execution_error = self._run_subprocess(
full_cmd,
timeout=timeout,
task_workspace=task_workspace,
)
except FileNotFoundError:
return AgentResult(
task_id=task_id,
timestamp=datetime.now(UTC),
env_status="failed",
agent_done="error",
error=f"Executable '{cmd[0]}' not found.",
metrics=AgentMetrics(end_to_end_ms=0, steps=0),
)
env_status, agent_done = self._map_exit_status(returncode, execution_error)
final_answer = ""
steps = 0
agent_success = None
if agent_done == "done":
agent_success = bool(final_answer)
return AgentResult(
task_id=task_id,
task=task_prompt,
timestamp=datetime.now(UTC),
env_status=env_status,
agent_done=agent_done,
agent_success=agent_success,
answer=final_answer if not execution_error else f"[Task Failed: {execution_error}]",
model_id=model_id,
browser_id=session_context.backend_id,
metrics=AgentMetrics(end_to_end_ms=0, steps=steps),
error=execution_error if env_status == "failed" else None,
)
CLIAgent provides:
_run_subprocess(cmd, timeout, task_workspace, ...) → (returncode, stdout_lines, execution_error); drains stdout/stderr to files, handles timeout (SIGTERM → SIGKILL), raises FileNotFoundError if executable not found_map_exit_status(returncode, execution_error) → (env_status, agent_done)
Constraints
- No print(); use logger
- No bare except or except Exception; catch specific exceptions and log ERROR
- No hardcoded timeout, URL, API key, or model values
