ComputerUseAgent — Browser Automation with Live Streaming

from pathlib import Path
import sys

from pathlib import Path
import sys

ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


from examples.run_multi_tool_agent import build_llm_from_env
from shipit_agent import Agent
from shipit_agent.computer_use import (
    ComputerUseAgent, MockBrowserSession, parse_action,
)

!playwright install chromium

llm = build_llm_from_env("bedrock")

from shipit_agent.llms import BedrockChatLLM, BedrockGemmaChatLLM

gemma3 = BedrockChatLLM(model="bedrock/google.gemma-3-27b-it")

eu = BedrockChatLLM(model="google.gemma-4-26b-a4b", aws_region_name="eu-central-1")
print("eu-central-1 base_url    :", eu._mantle_delegate.client_kwargs["base_url"])

import textwrap
import json

def _short(val, n=220):
    s = val if isinstance(val, str) else json.dumps(val, default=str)
    return textwrap.shorten(s.replace("\n", " "), width=n, placeholder=" …")


def pretty_event(ev):
    t, msg, p = ev.type, ev.message or "", ev.payload or {}
    if t == "run_started":
        return f'[status] run started :: {_short(p.get("prompt", ""), 140)}'
    if t == "planning_started":
        return f'[thinking] planning started :: {_short(p.get("prompt", ""), 140)}'
    if t == "planning_completed":
        return f'[thinking] planning completed :: {_short(p.get("output", ""))}'
    if t == "reasoning_started":
        return f'[thinking] 🧠 model reasoning started (iteration {p.get("iteration", "?")})'
    if t == "reasoning_completed":
        return f'[thinking] 🧠 reasoning :: {_short(p.get("content", ""), 320)}'
    if t == "step_started":
        return f'[step] LLM completion started :: iteration={p.get("iteration", "?")}, tools={p.get("tool_count", "?")}'
    if t == "tool_called":
        return f'[tool] ▶ calling `{msg}` :: args={_short(p.get("arguments", {}))}'
    if t == "tool_completed":
        return f'[tool] ✔ completed `{msg}` :: {_short(p.get("output", ""))}'
    if t == "tool_failed":
        return f'[tool] ✖ failed `{msg}` :: {_short(p.get("error", ""))}'
    if t == "interactive_request":
        return f"[interactive] {msg} :: {_short(p)}"
    if t == "run_completed":
        return f'[final] {_short(p.get("output", ""))}'
    return f"[{t}] {msg} :: {_short(p)}"

llm = BedrockChatLLM(model="google.gemma-4-31b")

import os

LIVE = bool(os.getenv("AWS_BEARER_TOKEN_BEDROCK"))
if not LIVE:
    try:  # generate a short-term bearer token from standard AWS credentials
        from aws_bedrock_token_generator import provide_token
        os.environ["AWS_BEARER_TOKEN_BEDROCK"] = provide_token(region="us-east-1")
        os.environ.setdefault("AWS_REGION_NAME", "us-east-1")
        LIVE = True
        print("✓ short-term Bedrock token generated from AWS credentials")
    except Exception as e:
        print(f"offline — no Bedrock credentials available ({type(e).__name__})")
else:
    print("✓ using AWS_BEARER_TOKEN_BEDROCK from the environment")

llm = BedrockChatLLM(model="google.gemma-4-31b", region="us-east-1")

live_agent = Agent(
        llm=llm,
        auto_use_skills=False,
    )

from IPython.display import clear_output
from IPython.display import Markdown, display
import json

lines: list[str] = []
print("Waiting for first event...", flush=True)

for event in live_agent.stream("Write a python function that takes a list of numbers and returns the sum of the even numbers in the list."):
    lines.append(f"`{event.type}` — {pretty_event(event)}")
    clear_output(wait=True)
    display(Markdown("## Live Stream\n\n" + "\n".join(lines)))

class ScriptedLLM:
    def __init__(self, replies):
        self.replies = replies
        self._i = 0
    def complete(self, *, messages, **_):
        text = self.replies[min(self._i, len(self.replies) - 1)]
        self._i += 1
        return text

price_llm = ScriptedLLM([
    "ACTION: navigate https://www.apple.com/iphone-15-pro/",
    "ACTION: scroll 0,800",
    "ACTION: done The iPhone 15 Pro starts at $999.",
])

browser = MockBrowserSession()
agent = ComputerUseAgent(
    llm=price_llm,
    browser=browser,
    goal="Find the starting price of the iPhone 15 Pro",
    max_iterations=8,
)
result = agent.run()

print(f'status: {result.status}')
print(f'final answer: {result.final_text}')
print(f'iterations: {result.iterations}')
print()
print('action history:')
for r in result.action_history:
    print(f'  {r.iteration}: {r.action.kind.value:<10} {r.action.args}')

form_llm = ScriptedLLM([
    "ACTION: click 400,220",
    "ACTION: type you",
    "ACTION: click 400,300",
    "ACTION: type [email protected]",
    "ACTION: click 400,380",
    "ACTION: type Hello, I would like to discuss our partnership.",
    "ACTION: click 400,460",
    "ACTION: done Form submitted successfully.",
])

form_browser = MockBrowserSession()
form_agent = ComputerUseAgent(
    llm=form_llm,
    browser=form_browser,
    goal="Fill out the contact form with name, email, and message",
    max_iterations=15,
)
form_result = form_agent.run()
print(form_result.final_text)
print()
print(f'browser ran {len(form_browser.calls)} commands:')
for c2 in form_browser.calls:
    print(f'  {c2[0]:<10} {c2[1]}')

class FlakyBrowser(MockBrowserSession):
    def __init__(self):
        super().__init__()
        self._click_count = 0
    def click(self, x, y):
        self._click_count += 1
        if self._click_count == 1:
            raise RuntimeError('Element not visible at (100, 100)')
        return super().click(x, y)

recover_llm = ScriptedLLM([
    'ACTION: click 100,100',  # will fail
    'ACTION: scroll 0 400',   # scroll down to bring element into view
    'ACTION: click 100,100',  # try again
    'ACTION: done Successfully clicked after scrolling.',
])

recover_browser = FlakyBrowser()
recover_agent = ComputerUseAgent(
    llm=recover_llm, browser=recover_browser, goal='click submit', max_iterations=8,
)
recover_result = recover_agent.run()

print(f'status: {recover_result.status}')
print(f'final: {recover_result.final_text}')
print()
print('action history with errors:')
for r in recover_result.action_history:
    err = f' ✗ {r.error}' if r.error else ' ✓'
    print(f'  {r.iteration}: {r.action.kind.value} {r.action.args}{err}')

anthropic_block = {
    'type': 'tool_use',
    'name': 'computer',
    'input': {
        'action': 'left_click',
        'coordinate': [320, 180],
        'rationale': 'Clicking the search icon in the top-right.',
    },
}
action = parse_action(anthropic_block)
print(f'kind: {action.kind.value}')
print(f'args: {action.args}')
print(f'rationale: {action.rationale}')

os.environ["SHIPIT_RUN_LIVE_BROWSER"] = "1"

import os

# Set SHIPIT_RUN_LIVE_BROWSER=1 to actually open the window and run this.
if os.getenv("SHIPIT_RUN_LIVE_BROWSER"):
    from shipit_agent import StreamRenderer
    from shipit_agent.computer_use import PlaywrightBrowserSession

    # headless=False → a real Chrome window opens so you can WATCH the agent
    # work, while StreamRenderer narrates every action in the terminal.
    with PlaywrightBrowserSession.launch(
        headless=False,                # visible browser (True for CI/servers)
        viewport_size=(1280, 720),
        slow_mo=200,                   # slow every op so you can SEE it move
        storage_state=".shipit_workspace/browser_state.json",  # remembers consent
    ) as live_browser:
        cu_agent = ComputerUseAgent(
            llm=llm,                   # vision-capable model recommended
            browser=live_browser,
            goal='Find the cheapest direct flight from SFO to JFK on May 20',
            max_iterations=15,
        )
        live_renderer = StreamRenderer(style="rich")
        for event in cu_agent.stream():   # live: browser window + terminal cards
            live_renderer.feed(event)
        live_renderer.close()
        live_browser.save_storage_state()  # next run skips consent walls
else:
    print("Set SHIPIT_RUN_LIVE_BROWSER=1 to open a visible Chrome window and")
    print("watch the agent browse live while the terminal streams tool cards.")
    print("(agent.run() still works unchanged if you prefer the blocking style.)")

from shipit_agent import StreamRenderer
from shipit_agent.computer_use import ComputerUseAgent, MockBrowserSession
from shipit_agent.llms.base import LLMResponse

class FlightSearchLLM:
    """Scripted vision LLM: navigate → type → Enter → scroll → done."""
    def __init__(self):
        self.acts = iter([
            "ACTION: navigate https://flights.example.com",
            "ACTION: type SFO to JFK May 20",
            "ACTION: key Enter",
            "ACTION: scroll 0,600",
            "ACTION: done Cheapest direct: $214 on United UA 1543.",
        ])
    def complete(self, *, messages, **_):
        return LLMResponse(content=next(self.acts))

demo_agent = ComputerUseAgent(
    llm=FlightSearchLLM(),
    browser=MockBrowserSession(),
    goal="Find the cheapest direct flight SFO→JFK on May 20",
    max_iterations=10,
)

renderer = StreamRenderer(style="plain")   # style="rich" for ANSI ⏺/⎿ cards
for event in demo_agent.stream():
    renderer.feed(event)
renderer.close()