DEV Community

Cover image for Operate Your Browser with a Voice Agent
Rishi Raj Jain
Rishi Raj Jain

Posted on

Operate Your Browser with a Voice Agent

#ai #programming #tutorial #python

Ever wish you could stay on a call with your hands full (driving, cooking, or pacing through a tense review) and say “pull up yesterday’s spreadsheet and paste that number into Slack”, trusting something on the wire to move real tabs instead of narrating imaginary clicks? Picture the teammate you already use on video calls, the one who answers docs questions or summarizes threads, except they can open sites you’re logged into, skim what is visible, click the control you named, type where you pointed, and tell you plainly if something failed.

This walkthrough builds that browser-connected voice participant: speech in and speech out via a realtime model, with every navigate, read, click, or type routed through Python functions on Playwright.

We wire everything up with Vision Agents as the voice agent framework, Stream for WebRTC audio and video, OpenAI Realtime so the model listens and replies in speech, Playwright with persistent sessions in a cloned Chrome instance, and roughly a dozen explicit tools (from open_chrome through run_browser_sequence) so the assistant never improvises selectors in plain text alone.

Demo

Prerequisites

You will need the following to get going with the implementation:

  • Python 3.12 or later
  • uv as the Python package manager
  • Google Chrome installed (the code defaults to a standard macOS path)

Table Of Contents

Set up the application

Clone the code from the GitHub repository by running the following commands:

git clone https://github.com/rishi-raj-jain/control-agent
uv sync
Enter fullscreen mode Exit fullscreen mode

Now, create a .env file at the root of your project. You are going to add the items you will save from the section below.

Setting Up Environment Variables

To configure your application, set up the necessary environment variables for each integration. Follow the steps below for each provider:

Create a Stream application

  • Go to the Stream dashboard.
  • Choose + Create an App.
  • In the dialog, type a name for your application and pick the region for your edge-server location(s).
  • After creation, locate the API Key and Secret under Your Credentials.

Stream Application Overview

  • Add these credentials to your .env file as follows: 
   STREAM_API_KEY="your-api-key"
   STREAM_API_SECRET="your-api-secret"
Enter fullscreen mode Exit fullscreen mode

Configure OpenAI 

   OPENAI_API_KEY="your-openai-api-key"
Enter fullscreen mode Exit fullscreen mode

Ensure your organization can use the Realtime API and the model you pick in code (here, gpt-realtime-1.5).

Chrome executable and profiles (Optional)

Most setups work with defaults the code has for MacOS. Tune these only if Chrome lives in a non-standard path or you want to clone from a particular profile folder:

  • CHROME_PATH — full path to the Chrome binary when it is not the default Apple Silicon macOS bundle path.
  • CHROME_USER_DATA_DIR — source User Data directory Playwright seeds from (omit to detect the OS default Chrome profile directory).
  • CHROME_AGENT_USER_DATA_DIR — agent-isolated profile root (omit to use ~/.control-agent/chrome).
  • CHROME_PROFILE_DIRECTORY — which named profile folder under User Data to read (omit to infer last-used from Chrome’s Local State file).

Append to .env when needed:

## Optional Chrome (omit to use sensible defaults)

# CHROME_PATH="/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
# CHROME_USER_DATA_DIR=""
# CHROME_AGENT_USER_DATA_DIR="~/.control-agent/chrome"
# CHROME_PROFILE_DIRECTORY="Default"
Enter fullscreen mode Exit fullscreen mode

Putting it together, a minimal .env should look something like this:

# .env

## Stream
STREAM_API_KEY="..."
STREAM_API_SECRET="..."

## OpenAI (Realtime)
OPENAI_API_KEY="sk-proj-..."
Enter fullscreen mode Exit fullscreen mode

That's it for configuring environment variables. The sections that follow unpack Agent.instructions, the registered tool surface, Playwright-backed Chrome control, then the concrete flows in main.py.

Defining agent behavior with instructions

Realtime voice models are great at conversation, but browser state is not in the transcript. If you only say "be helpful" the model may narrate clicks or confuse closing a tab with quitting Chrome. Tight instructions on the Agent turns that vague intent into repeatable steps - always connect with open_chrome first, use read_page before claiming what is on screen, reserve go_back for history (not tab rotation), and batch steps with run_browser_sequence when the user asks for a short workflow.

The string passed into Agent(..., instructions=...) below names the tools, separates close_tab from close_chrome, tells the model to normalize spoken site names to URLs, and embeds a copy-pasteable JSON example for multi-step flows so the schema is not invented on the fly.

async def create_agent(kwargs) -> Agent:
    agent = Agent(
        edge=getstream.Edge(),
        llm=openai.Realtime(model="gpt-realtime-1.5", voice="ash"),
        agent_user=User(name="Browser Assistant", id="agent"),
        instructions=(
            "You are a voice assistant in a live call that controls the user's Chrome browser.\n"
            "When the user asks for browser work, call the browser tools instead of guessing.\n"
            "If Chrome is not connected yet, call open_chrome first.\n"
            "Use close_chrome to close the agent Chrome window and reopen_chrome to open a fresh one.\n"
            "Use close_tab only to close a single website tab, not the whole Chrome window.\n"
            "For spoken website names, turn them into real URLs before opening tabs.\n"
            "Use read_page to fetch the rendered page text, input fields, and clickable labels.\n"
            "Use type_into_field to type into a search box or other input, then set submit=true to press Enter.\n"
            "click_on_page reads the page first and clicks the closest visible label to what the user said.\n"
            "Use go_back for browser history, not switch_tab.\n"
            "For multi-step requests, use run_browser_sequence with ordered steps.\n"
            "Example: open Gmail and compose -> "
            '[{"action":"open_tab","url":"https://mail.google.com"},'
            '{"action":"click","label":"Compose","domain":"mail.google.com"}].\n'
            "Confirm what you did briefly after each action."
        ),
    )
Enter fullscreen mode Exit fullscreen mode

Now, let's move to the next section to learn about how browser is being controlled by the AI agent.

Exposing browser work as registered tools

A common trap is STT → LLM → TTS where the model describes what it would click. Tool calling flips that so that the model emits named operations with arguments (label, url, optional domain), your Python runs Playwright, and the returned string becomes the model’s ground truth for what to say next (success, error, or “closest matches” when a label is vague).

Step Responsibility
1 User speaks, Stream carries audio and then OpenAI Realtime handles speech understanding and replies.
2 The model issues structured tool calls mapped to @agent.llm.register_function handlers.
3 Playwright mutates a real BrowserContext (tabs, history, focus).
4 Each tool returns a str: confirmation, failure, or candidate labels when confidence is low.

Vision Agents exposes these functions to the agent with @agent.llm.register_function (routing hints for the model) and thin async wrappers that delegate to BrowserController:

    @agent.llm.register_function(
        description="Launch a dedicated Chrome window for the agent with the default profile and connect to it."
    )
    async def open_chrome() -> str:
        return await browser.open_chrome()

    @agent.llm.register_function(description="Close the agent Chrome window.")
    async def close_chrome() -> str:
        return await browser.close_chrome()

    @agent.llm.register_function(
        description="Close the agent Chrome window and open a fresh one with the default profile."
    )
    async def reopen_chrome() -> str:
        return await browser.reopen_chrome()

    @agent.llm.register_function(description="List the open Chrome tabs.")
    async def list_tabs() -> str:
        return await browser.list_tabs()

    @agent.llm.register_function(description="Open a new Chrome tab at the spoken URL or site name.")
    async def open_tab(url: str) -> str:
        return await browser.open_tab(url)

    @agent.llm.register_function(description="Close an open tab that matches the spoken domain.")
    async def close_tab(domain: str) -> str:
        return await browser.close_tab(domain)

    @agent.llm.register_function(
        description="Switch tabs by domain, title, tab number, or say next or previous."
    )
    async def switch_tab(target: str) -> str:
        return await browser.switch_tab(target)

    @agent.llm.register_function(
        description="Read the rendered page text, input fields, and clickable labels on the current or matching tab."
    )
    async def read_page(domain: str | None = None) -> str:
        return await browser.read_page(domain)

    @agent.llm.register_function(
        description="Type into the closest matching input field such as a search box on the current or matching tab."
    )
    async def type_into_field(
        text: str,
        field: str,
        domain: str | None = None,
        submit: bool = False,
    ) -> str:
        return await browser.type_into_field(text, field, domain, submit)

    @agent.llm.register_function(
        description="Click the closest visible label to what the user said on the current or matching tab."
    )
    async def click_on_page(label: str, domain: str | None = None) -> str:
        return await browser.click_on_page(label, domain)

    @agent.llm.register_function(
        description="Go back one page in the current tab's browser history."
    )
    async def go_back(domain: str | None = None) -> str:
        return await browser.go_back(domain)

    @agent.llm.register_function(
        description="Run an ordered list of browser actions such as open Gmail then click Compose."
    )
    async def run_browser_sequence(steps: list[dict[str, str]]) -> str:
        return await browser.run_browser_sequence(steps)
Enter fullscreen mode Exit fullscreen mode

Now, let's move to the next section to learn about what's actually empowering the control of Chrome to the AI agent, i.e. Playwright.

Playwright session and persistent Chrome

Playwright drives a real Chromium session with async APIs that match Vision Agents’ tool handlers, and it exposes semantic locators such as page.goto, go_back, get_by_role, get_by_label, fill, click, instead of brittle “click at pixel (x,y)” hacks. Here Chromium is launched with launch_persistent_context an agent-only user-data directory (by default under ~/.control-agent/chrome) so cookies survive across runs, seeded by copying selective files from the user’s Chrome profile (_prepare_agent_profile, _ensure_agent_local_state, PROFILE_SYNC_FILES earlier in the file).

BrowserController owns one BrowserContext and an asyncio.Lock so concurrent realtime tool calls cannot double-launch Chrome and _ensure_connected reopens when the handle is stale.

    async def _launch_agent_chrome(self) -> str:
        await self._close_agent_chrome()

        agent_user_data_dir = _agent_chrome_user_data_dir()
        source_user_data_dir = CHROME_USER_DATA_DIR or _default_chrome_user_data_dir()
        if source_user_data_dir:
            await asyncio.to_thread(
                _prepare_agent_profile,
                agent_user_data_dir,
                source_user_data_dir,
            )

        playwright = await self._ensure_playwright()
        launch_args = [
            "--no-first-run",
            "--no-default-browser-check",
            f"--profile-directory={AGENT_PROFILE_DIRECTORY}",
            "--remote-allow-origins=*",
        ]
        launch_kwargs: dict[str, object] = {
            "user_data_dir": agent_user_data_dir,
            "headless": False,
            "args": launch_args,
        }
        if os.path.exists(CHROME_PATH):
            launch_kwargs["executable_path"] = CHROME_PATH
        else:
            launch_kwargs["channel"] = "chrome"

        try:
            self._browser_context = await playwright.chromium.launch_persistent_context(
                launch_kwargs
            )
        except Exception as exc:
            return f"Could not open Chrome: {exc}"

        return "Opened a new Chrome window with your default profile and connected."
Enter fullscreen mode Exit fullscreen mode

Now, let's move to the next section to understand how is the browser being operated on such as closing and opening of tabs.

Launch Chrome, manage tabs, go back, and click

Open / close run under the controller lock, either confirming an existing context or invoking _launch_agent_chrome, versus closing the BrowserContext.

    async def open_chrome(self) -> str:
        async with self._lock:
            if self._is_agent_chrome_connected():
                try:
                    _ = self._browser_context.pages
                    return "Agent Chrome is already open and connected."
                except Exception:
                    self._browser_context = None

            return await self._launch_agent_chrome()

    async def close_chrome(self) -> str:
        async with self._lock:
            if not self._is_agent_chrome_connected():
                return "Agent Chrome is not open."

            await self._close_agent_chrome()
            return "Closed the agent Chrome window."
Enter fullscreen mode Exit fullscreen mode

A new tab is context.new_page(), goto with domcontentloaded, bring_to_front, matching what a human expects as the focused surface.

    async def open_tab(self, url: str) -> str:
        try:
            target = _normalize_url(url)
        except ValueError as error:
            return str(error)

        try:
            context = await self._active_context()
            page = await context.new_page()
            await page.goto(target, wait_until="domcontentloaded", timeout=60000)
            await page.bring_to_front()
            title = await page.title()
            return f"Opened {target} in a new tab ({title})."
        except RuntimeError as error:
            return str(error)
        except Exception as error:
            return f"Could not open {url}: {error}"
Enter fullscreen mode Exit fullscreen mode

go_back is per tab browser history, not cycling tabs which matches the instructions that contrast it with switch_tab.

    async def go_back(self, domain: str | None = None) -> str:
        try:
            page = await self._page_for(domain)
        except RuntimeError as error:
            return str(error)

        await page.bring_to_front()
        try:
            await page.go_back(wait_until="domcontentloaded", timeout=5000)
        except Exception:
            return "There is no previous page to go back to."

        title = await page.title()
        return f"Went back to {title} ({page.url})."
Enter fullscreen mode Exit fullscreen mode

click_on_page scores visible labels gathered in-page against what the speaker said (_score_label_match + CLICK_MATCH_THRESHOLD guardrails), then invokes _click_target, which prefers get_by_role (button, link, …) and falls back to get_by_text.

    async def _click_target(self, page: Page, text: str) -> None:
        for role in ("button", "link", "menuitem", "tab", "option"):
            locator = page.get_by_role(role, name=re.compile(re.escape(text), re.IGNORECASE))
            if await locator.count():
                await locator.first.click()
                return

        locator = page.get_by_text(re.compile(re.escape(text), re.IGNORECASE))
        if await locator.count():
            await locator.first.click()
            return

        raise RuntimeError(f'Could not click "{text}".')
Enter fullscreen mode Exit fullscreen mode 
    async def click_on_page(self, label: str, domain: str | None = None) -> str:
        try:
            page = await self._page_for(domain)
        except RuntimeError as error:
            return str(error)

        await page.bring_to_front()
        targets = await self._gather_click_targets(page)
        if not targets:
            return f'Could not find "{label}" because no clickable text was found on {page.url}.'

        scored = sorted(
            ((target, _score_label_match(label, target["text"])) for target in targets),
            key=lambda item: item[1],
            reverse=True,
        )
        best_target, best_score = scored[0]
        if best_score < CLICK_MATCH_THRESHOLD:
            closest = ", ".join(target["text"] for target, _ in scored[:5])
            return (
                f'Could not find a confident match for "{label}" on {page.url}. '
                f"Closest options: {closest}"
            )

        matched_text = best_target["text"]
        try:
            await self._click_target(page, matched_text)
        except RuntimeError as error:
            return str(error)

        if matched_text.lower() == label.strip().lower():
            return f'Clicked "{matched_text}" on {page.url}.'
        return (
            f'Clicked "{matched_text}" on {page.url}. '
            f'That was the closest match to "{label}".'
        )
Enter fullscreen mode Exit fullscreen mode

Tab routing, scraping the DOM, optional domain filters

Optional domain funnels read_page, click_on_page, type_into_field, and go_back through _page_for, among open tabs it picks the last whose URL satisfies _domain_matches, limiting the "wrong tab" mistakes when Gmail and Slack both stay open.

    async def _page_for(self, domain: str | None = None) -> Page:
        pages = await self._pages()
        if not pages:
            raise RuntimeError("No open tabs were found.")

        if domain:
            matches = [page for page in pages if _domain_matches(page.url, domain)]
            if not matches:
                raise RuntimeError(f"No open tab matched {domain}.")
            return matches[-1]
        return pages[-1]
Enter fullscreen mode Exit fullscreen mode

switch_tab rotates which tab has focus (next, previous, numbered index, or domain/title match), orthogonal to go_back.

    async def switch_tab(self, target: str) -> str:
        pages = await self._pages()
        if not pages:
            return "No open tabs were found."

        cleaned = target.strip().lower()
        if cleaned in {"next", "forward"}:
            current = pages[-1]
            index = pages.index(current)
            page = pages[(index + 1) % len(pages)]
        elif cleaned in {"previous", "back", "prior"}:
            current = pages[-1]
            index = pages.index(current)
            page = pages[(index - 1) % len(pages)]
        elif cleaned.isdigit():
            index = int(cleaned) - 1
            if index < 0 or index >= len(pages):
                return f"There is no tab number {cleaned}."
            page = pages[index]
        else:
            matches = [
                page
                for page in pages
                if _domain_matches(page.url, target)
                or target.lower() in (await page.title()).lower()
            ]
            if not matches:
                return f"No tab matched {target}."
            page = matches[-1]

        await page.bring_to_front()
        title = await page.title()
        return f"Switched to {title} ({page.url})."
Enter fullscreen mode Exit fullscreen mode

read_page brings that tab forward, pulls body text (truncated at MAX_PAGE_TEXT_CHARS), then appends enumerated inputs and clickable labels assembled by page.evaluate scripts (_gather_input_targets, _gather_click_targets) that skip invisible or zero-area nodes, supplying discrete strings the voice model can reuse on subsequent tool calls.

    async def read_page(self, domain: str | None = None) -> str:
        try:
            page = await self._page_for(domain)
        except RuntimeError as error:
            return str(error)

        await page.bring_to_front()
        title = await page.title()
        body = _collapse_whitespace(await page.locator("body").inner_text())
        if len(body) > MAX_PAGE_TEXT_CHARS:
            body = body[:MAX_PAGE_TEXT_CHARS] + "..."

        targets = await self._gather_click_targets(page)
        inputs = await self._gather_input_targets(page)
        sections = [
            f"Title: {title}",
            f"URL: {page.url}",
            f"Rendered page text:\n{body or '[No visible text found]'}",
        ]
        if inputs:
            input_lines = []
            for target in inputs:
                label = self._input_target_label(target)
                details = []
                if target.get("placeholder"):
                    details.append(f'placeholder="{target["placeholder"]}"')
                if target.get("type"):
                    details.append(f'type="{target["type"]}"')
                if target.get("value"):
                    details.append(f'value="{target["value"]}"')
                suffix = f" ({', '.join(details)})" if details else ""
                input_lines.append(f"- {label}{suffix}")
            sections.append("Input fields:\n" + "\n".join(input_lines))
        if targets:
            target_lines = "\n".join(f"- {target['text']}" for target in targets)
            sections.append(f"Clickable text:\n{target_lines}")
        return "\n\n".join(sections)
Enter fullscreen mode Exit fullscreen mode

Now, let's move to the next section to learn how the inputs are actually being filled by the AI agent with the help of Playwright locators.

Fill Out Forms and Input Fields

type_into_field gathers the same _gather_input_targets list, scores it with _best_input_target, and refuses to type when INPUT_MATCH_THRESHOLD is not met, surfacing closest fields for the user to disambiguate by voice.

    async def type_into_field(
        self,
        text: str,
        field: str,
        domain: str | None = None,
        submit: bool = False,
    ) -> str:
        try:
            page = await self._page_for(domain)
        except RuntimeError as error:
            return str(error)

        await page.bring_to_front()
        inputs = await self._gather_input_targets(page)
        if not inputs:
            return f'No input fields were found on {page.url}.'

        match = self._best_input_target(field, inputs)
        if match is None:
            return f'No input fields were found on {page.url}.'

        best_target, best_score = match
        if best_score < INPUT_MATCH_THRESHOLD:
            closest = ", ".join(self._input_target_label(target) for target in inputs[:5])
            return (
                f'Could not find a confident match for "{field}" on {page.url}. '
                f"Closest fields: {closest}"
            )

        matched_label = self._input_target_label(best_target)
        try:
            await self._fill_input_target(page, best_target, text, submit)
        except Exception as error:
            return str(error)

        action = " and submitted it" if submit else ""
        if matched_label.lower() == field.strip().lower():
            return f'Typed "{text}" into "{matched_label}" on {page.url}{action}.'
        return (
            f'Typed "{text}" into "{matched_label}" on {page.url}{action}. '
            f'That was the closest match to "{field}".'
        )
Enter fullscreen mode Exit fullscreen mode

_fill_input_target walks a fixed ladder of Playwright locators (get_by_label, aria-label, placeholder, get_by_role, name, #id, then input[type="search"]), focuses the control, fills, and optionally press("Enter") when submit is true.

    async def _fill_input_target(
        self, page: Page, target: dict[str, str], text: str, submit: bool
    ) -> None:
        locator: Locator | None = None
        label = target.get("label", "")
        placeholder = target.get("placeholder", "")
        name = target.get("name", "")
        field_id = target.get("id", "")
        role = target.get("role", "")
        input_type = target.get("type", "")

        if label:
            by_label = page.get_by_label(label, exact=False)
            if await by_label.count():
                locator = by_label.first
            else:
                by_label = page.get_by_label(label, exact=True)
                if await by_label.count():
                    locator = by_label.first
            if locator is None:
                locator = await _locator_by_attribute(page, "aria-label", label)

        if locator is None and placeholder:
            locator = await _locator_by_attribute(page, "placeholder", placeholder)

        if locator is None and role in {"searchbox", "combobox", "textbox"}:
            accessible_name = label or placeholder or name
            if accessible_name:
                by_role = page.get_by_role(role, name=accessible_name, exact=False)
                if await by_role.count():
                    locator = by_role.first
                else:
                    by_role = page.get_by_role(role, name=accessible_name, exact=True)
                    if await by_role.count():
                        locator = by_role.first

        if locator is None and name:
            locator = await _locator_by_attribute(page, "name", name)

        if locator is None and field_id:
            field_locator = page.locator(f"#{field_id}")
            if await field_locator.count():
                locator = field_locator.first

        if locator is None and input_type == "search":
            search_locator = page.locator('input[type="search"]')
            if await search_locator.count():
                locator = search_locator.first

        if locator is None:
            raise RuntimeError(f'Could not find the input field "{self._input_target_label(target)}".')

        await locator.click()
        await locator.fill(text)
        if submit:
            await locator.press("Enter")
Enter fullscreen mode Exit fullscreen mode

Finally, let's move to the next section to learn how multiple steps as shown in the demo are being handled in sequence to accomodate various actions from a single voice instruction.

Multi-step batches with run_browser_sequence

When you want fewer round-trips across the realtime session, run_browser_sequence executes an ordered list, branching on action and forwarding to the same methods as standalone tools such as the Gmail + Compose example in instructions is exactly the payload shape consumed here.

    async def run_browser_sequence(self, steps: list[dict[str, str]]) -> str:
        results: list[str] = []
        for step in steps:
            action = (step.get("action") or "").strip().lower()
            if action == "open_chrome":
                results.append(await self.open_chrome())
            elif action == "close_chrome":
                results.append(await self.close_chrome())
            elif action == "reopen_chrome":
                results.append(await self.reopen_chrome())
            elif action == "open_tab":
                results.append(await self.open_tab(step.get("url", "")))
            elif action == "close_tab":
                results.append(await self.close_tab(step.get("domain", "")))
            elif action == "switch_tab":
                results.append(await self.switch_tab(step.get("target", "")))
            elif action == "click":
                results.append(
                    await self.click_on_page(
                        step.get("label", ""),
                        step.get("domain"),
                    )
                )
            elif action == "list_tabs":
                results.append(await self.list_tabs())
            elif action == "read_page":
                results.append(await self.read_page(step.get("domain")))
            elif action == "type_into_field":
                results.append(
                    await self.type_into_field(
                        step.get("text", ""),
                        step.get("field", ""),
                        step.get("domain"),
                        (step.get("submit", "false").lower() == "true"),
                    )
                )
            elif action == "go_back":
                results.append(await self.go_back(step.get("domain")))
            else:
                results.append(f"Unknown action: {action or 'missing'}")
        return " ".join(results)
Enter fullscreen mode Exit fullscreen mode

How to Run

To launch the service locally, use the following command in your terminal:

uv run main.py run
Enter fullscreen mode Exit fullscreen mode

It will open a demo at demo.visionagents.ai and automatically join the session for you. You can then operate your browser with voice in real time, as demonstrated in the demo.

The terminal will display the following output prior to the agent joining the call, confirming that the browser tools (open_chrome, list_tabs, open_tab, read_page, click_on_page, type_into_field, run_browser_sequence, …) have been automatically registered:

# Illustrative log snippet

% uv run main.py run

... | INFO     | 🚀 Launching agent...
... | INFO     | [Agent: agent] | 🤖 Agent joining call: ...
+ ... | INFO     | Added 12 tools to session config: ['open_chrome', 'close_chrome', …]
Enter fullscreen mode Exit fullscreen mode

Ending Thoughts

While real-time voice powered by Stream and OpenAI Realtime makes browser interaction conversational, actual control of the browser relies on registered Python functions that precisely map spoken instructions to Playwright actions, ensuring consistent and reliable results rather than leaving execution up to the model's interpretation. These times are exciting to see what all is possible with AI including writing code to mostly not touch your keys again for basic operations.

Top comments (0)