FastCDP API

Source and API details

fastcdp is built from Chrome’s own protocol description: the bundled browser_protocol.json and js_protocol.json list every domain, command, event, and parameter, with descriptions. Loading them once here is what lets the library expose the whole protocol with real signatures and docs, rather than wrapping a hand-picked subset. (The __file__ shuffle covers running this notebook interactively, where nbdev hasn’t set it.)

len(_cdp_domains), [d['domain'] for d in _cdp_domains[:5]]

(55, ['Accessibility', 'Animation', 'Audits', 'Autofill', 'BackgroundService'])

The protocol is large, so discovery is a search problem: cdp_search matches command and event names and descriptions, answering “what’s the CDP command for X?” without leaving the session.

cdp_search

def cdp_search(
    q:str
):

Search CDP domains and commands by name or description

cdp_search('target')[:100]

'Audits.checkFormsIssues: Runs the form issues check for the target page. Found issues are reported\nu'

cdp_conninfo

def cdp_conninfo(
    p:str=None, # Contents of a `DevToolsActivePort` file
    d:str | pathlib.Path=None, # Profile dir whose `DevToolsActivePort` to read
):

Connection info from contents p, profile dir d, or the default Chrome profile

Chrome (146+) can expose CDP from your everyday browser: enable Allow remote debugging in chrome://inspect/#remote-debugging. Chrome then accepts WebSocket connections on the endpoint recorded in its profile’s DevToolsActivePort file – port on the first line, WebSocket path on the second – and asks you to approve each newly connecting client. CDP.connect() waits up to 60 seconds for this approval. Only that WebSocket endpoint is served (the /json/* HTTP interface belongs to the debug-instance mechanism described later). cdp_conninfo reads the file, and CDP.connect() uses it by default.

On MacOS, the connection info is stored in ~/Library/Application Support/Google/Chrome/DevToolsActivePort (or ~/Library/Application Support/Chromium/DevToolsActivePort for Chromium).

conninfo = cdp_conninfo()
conninfo

'9222/devtools/browser/36601b5b-29af-474d-aa2d-2820c1e44169'

CDP is one websocket and two routing tables. _pending maps command ids to reply futures, and _events maps event names to subscriber queues; the read loop resolves the one and fans out to the other. __call__ sends any protocol command by name, bounded by command_timeout, raising protocol errors as RuntimeError and unwrapping single-value results. A background task pings every 30 seconds to keep the connection alive.

CDP

def CDP(
    wsconn:str=None, debug:bool=False, command_timeout:float=10
):

Chrome DevTools Protocol connection with event support

The transport is pluggable. _send (command frame in, reply frame out) and _dispatch (route an event frame to subscribed queues) are the only two methods that touch the websocket protocol flow, so an alternative transport subclasses CDP, overrides _send and the connection lifecycle, and feeds incoming events to _dispatch. Everything else (domain proxies, helpers, event buffers, Page) is inherited unchanged. solvecdp does exactly this, relaying frames through a solveit server to a Chrome extension. Every protocol command is bounded by command_timeout (10 seconds by default), so a lost reply cannot leave a helper waiting forever.

# await cdp.close()

CDP.remote

async def remote(
    port:int=9223, # Remote debugging port of the running Chrome
    debug:bool=None, # Print each event as it arrives?
):

Connect via Chrome remote debugging HTTP endpoint

remote targets the other mechanism: a separate “debug Chrome” started with a remote debugging port. fastcdp-setup creates a “CDP Chrome” launcher for one on remote’s default port 9223 (not 9222, which a main browser with built-in debugging enabled already holds). Since Chrome 136 the debugging switches are ignored for your everyday profile – a debug instance must point --user-data-dir at a non-standard directory, so that scripts can’t reach your real profile’s (differently-encrypted) cookies. To start one by hand on macOS:

'/Applications/Google Chrome.app/Contents/MacOS/Google Chrome' \
  --remote-debugging-port=9223 --user-data-dir=$HOME/.cache/fastcdp/cdp-chrome

That instance serves the classic HTTP metadata endpoints, and remote reads webSocketDebuggerUrl from /json/version to connect – no approval prompts, since the profile is disposable.

Launching Chrome

The third connection option: start the user’s installed Chrome ourselves, CDP-ready – no manual setup, no approval popups. launch runs it on a separate profile directory (Chrome requires a non-default one for debugging) with an ephemeral debug port, and quit shuts it down again. One launched instance per profile dir: a second launch on the same dir would just signal the running instance and exit.

We launch headless=True here so running this notebook’s tests doesn’t pop up a browser window; drop it when you want to watch.

chrome_bin

def chrome_bin():

Path of the installed Chrome/Chromium binary ($FASTCDP_CHROME overrides)

CDP.launch

async def launch(
    user_data_dir:str | pathlib.Path=None, # Profile dir; `~/.cache/fastcdp/profile` if None
    headless:bool=False, # Run without a visible window?
    debug:bool=None, # Print protocol events?
    timeout:int=10, # Seconds to wait for the debug endpoint
    reuse:bool=True, # Connect to an instance already running on this profile? (Else raise)
):

Launch the installed Chrome CDP-ready on its own profile dir, and connect to it

quit is launch’s counterpart: as well as closing the browser and connection it waits for the process to exit (or the debug port to free), so an immediately following launch on the same profile can’t collide with a half-dead instance.

CDP.quit

async def quit():

Quit the browser, wait until it has released its debug port, and close the connection

lcdp = await CDP.launch(headless=True)
lcdp.port

A second launch on the same profile connects to the running instance instead of failing (reuse=False to make it an error). The reattached handle has no proc, since the process belongs to whoever launched it, and quit works regardless. This is the recovery path when a kernel restart orphans a launched browser.

A quit Chrome leaves DevToolsActivePort behind, so launch trusts the file only after checking its port with is_port_free: a free port proves the file stale, and launch removes it and starts fresh.

rcdp = await CDP.launch()
assert rcdp.port == lcdp.port and rcdp.proc is None
await rcdp.close()

cdp = await CDP.remote(port=lcdp.port)

pages lists the open tabs as attribute-access rows, one line per target with * marking attached ones – the ids it shows are what attach and the Page helpers below take:

CDP.pages

async def pages()->Targets: # Rows with `targetId`, `url`, `title`, `attached`, ...; `*` marks attached

The browser’s open page targets

Targets

def Targets(
    *args, **kwargs
):

Page targets as attribute-access rows, one line per target

await cdp('Target.createTarget', url='https://example.com')

'C670B0A075CA07FDD84FE718250ACE76'

ps = await cdp.pages
pg = first(p for p in ps if 'example.com' in p.url)
pg.title

''

Every CDP domain is an attribute: cdp.page.navigate(url=...) sends Page.navigate. There is no generated code behind this – CDPDomain and CDPMethod are built on demand by name, and each CDPMethod looks up its command in the protocol schema to give itself a real keyword-only signature and a docstring listing the parameters. So tab completion, ?, and LLM introspection all work across the whole protocol for the cost of two small classes.

The entry point is on the class itself: CDP.__getattr__ hands any non-underscore attribute to CDPDomain, and CDP.__dir__ advertises the domains by reading _domains – defined here, after the class, and resolved at call time. Those two stay inside the class rather than becoming @patches beside these helpers because exporting module-level functions named __getattr__ or __dir__ would turn them into the module’s PEP 562 attribute hooks.

CDPDomain

def CDPDomain(
    cdp, domain
):

A protocol domain as an attribute namespace of its commands

CDPMethod

def CDPMethod(
    cdp, domain, method
):

A protocol command as a callable with a generated signature and docs

Commands act on a tab through a session: attach starts one for a target and returns its id, which is what every sid argument in the library names. Passing no sid addresses the browser itself.

CDP.attach

async def attach(
    tid:str, # Target id, e.g. from `pages`
)->str: # Session id, for `sid` arguments

Attach to target tid

eval is the workhorse read primitive: evaluate a JS expression in the page, awaiting promises and returning by value. A JS exception surfaces as a Python RuntimeError rather than a result field to check, so a broken expression fails a test at the call site.

CDP.eval

async def eval(
    expr:str, # JS expression; a promise result is awaited
    sid:str=None, # Session to evaluate in
):

Evaluate expr in the page, raising on a JS exception

tid = pg.targetId
sid = await cdp.attach(tid)
await cdp.eval('document.title', sid)

'Example Domain'

Event handling is subscribe-then-act: on registers a fresh queue for some events for the scope of a block, and every matching frame lands in it. The pattern matters because CDP pushes events as they happen – a frame that fired before the queue existed is simply gone, which is why the navigation helpers below always subscribe before triggering anything.

CDP.wait_event

async def wait_event(
    event:str, timeout:int=10
):

The next event frame, within timeout seconds

on

def on(
    *events:str
):

Subscribe to events for the scope of the block, yielding the queue their frames arrive on

The polling waits are all fastcore’s wait_until with a different probe. wait_for polls a JS expression and returns its (truthy) value, and wait_for_selector and wait_defined phrase common conditions as expressions. When no wait_for_* fits, write a probe and call wait_until directly.

CDP.wait_defined

async def wait_defined(
    name:str, sid:str=None, timeout:int=10
):

Wait until the global name exists and is truthy

CDP.wait_for_selector

async def wait_for_selector(
    sel:str, # CSS selector to watch
    present:bool=True, # Wait for a match to appear (True) or for none to remain (False)
    sid:str=None, # Session to wait in
    timeout:int=10, # Seconds to wait before raising
):

Wait for CSS selector sel to match an element (with present=False, to match none)

CDP.wait_for

async def wait_for(
    expr:str, sid:str=None, timeout:int=10
):

Wait for JS expression to be truthy, return its value

t = await cdp.target.createTarget(url='about:blank')
sid = await cdp.attach(t)
page = await cdp.page.enable(sid=sid)

await cdp.page.navigate(sid=sid, url='https://httpbingo.org/forms/post')
await cdp.wait_for_selector('form', sid=sid)

True

await cdp.target.closeTarget(targetId=t)

True

Passing sid to every call gets old, so Page bundles a connection, a target, and its session, and forwards attribute access to the CDP object with sid filled in: helpers become sid-free methods, and domains come back wrapped in PageDomain so their commands are bound too. _copy_meta carries each wrapped callable’s docstring and signature across (minus the sid parameter), keeping introspection as informative on a Page as on a CDP. Page.new also enables focus emulation, so a driven page renders and takes input as if focused even when its tab or window is hidden.

PageDomain

def PageDomain(
    sid, domain
):

A CDP domain bound to one session

Page

def Page(
    cdp:CDP, t:str, sid:str, owned:bool=False, frame_id:str=None
):

A tab and its session: every CDP helper and domain, with sid filled in

CDP.remote_page

async def remote_page(
    port:int=9223, # Remote debugging port of the running Chrome
    debug:bool=None, # Print each event as it arrives?
):

Connect via remote debugging and return a Page for the active tab

CDP.active_page

async def active_page():

A Page driving the focused attachable tab, or None when none has focus; a hidden tab never qualifies, since focus emulation makes hasFocus() lie

remote_page is the quickest start against a debug Chrome: connect and drive whichever tab is focused.

page = await CDP.remote_page(port=lcdp.port)
await page.eval('document.title')

'Example Domain'

CDP.new_page

async def new_page(
    background:bool=False, # Open the tab without focusing it, so the browser window is not raised
):

Create a new tab, return Page

page = await cdp.new_page()
await page.page.navigate(url='https://httpbingo.org/forms/post')
await page.wait_for_selector('form')

True

page.is_open

True

await page.close()

page.is_open

True

attach_page is new_page’s counterpart for a tab that already exists: the same Page proxy, bound to the target you name.

CDP.attach_page

async def attach_page(
    tid:str, # Target id, e.g. from `pages`
)->Page: # Proxy driving that tab

Attach to the existing tab tid

CDP.wait_for_new_page

async def wait_for_new_page(
    existing, # Page target rows or target IDs captured before the opening action
    timeout:float=10, # Seconds to wait before raising
):

Wait for a page absent from existing, attach it, and return its Page

Capture cdp.pages before an action that opens a tab or popup, then pass that snapshot to wait_for_new_page. The wait reads target state rather than depending on a target-created event, so a page that appeared before the wait began is still found.

page = await cdp.attach_page(pg.targetId)
await page.eval('document.title')

'Example Domain'

The private wait machinery shares one deadline across phases and filters every event by the page’s session. Network-idle tracking only starts before an operation, when it can observe each request opening; there is deliberately no post-hoc idle helper that could miss work already in flight.

expect_navigation

def expect_navigation(
    sid:str=None, # Session whose top frame must navigate
    wait:str | None='load', # 'load', 'idle', or None to stop after navigation begins
    timeout:float=10, # Maximum seconds for the action and requested wait
    idle_ms:int=100, # Quiet time after load when `wait='idle'`
):

Subscribe before an action, require a top-frame navigation, then perform the requested wait

expect_navigation is for actions such as clicks: it subscribes first and then requires a top-frame navigation, including history and hash changes. goto already owns the navigation command, so it can use Chrome’s returned loaderId directly and does not need to infer that a navigation began.

CDP.goto

async def goto(
    url:str, # URL to navigate to
    sid:str=None, # Session to navigate
    wait:str | None='load', # 'load' (default), 'idle', or None
    timeout:float=10, # Maximum seconds for navigation and the requested wait
    idle_ms:int=100, # Quiet time after load when `wait='idle'`
):

Navigate to url, perform the requested wait, and raise on a navigation error

Navigation waiting is explicit. goto defaults to wait='load', the useful baseline: Chrome has accepted the navigation and the new document has fired load. Use wait='idle' only when the page’s initial requests must also settle, or wait=None when the caller will wait for a more meaningful condition such as text or an accessibility node. timeout is one deadline for the navigation and requested wait, rather than a fresh allowance for each phase.

expect_navigation is the action form. It subscribes before the wrapped action, requires that the page’s top frame actually navigates (including same-document history/hash changes), and then applies the same wait mode. This is intentionally not a post-hoc readiness call: after an action returns there is no reliable way to reconstruct a request or navigation event that has already happened.

For wait='idle', requestWillBeSent opens an in-flight request and loadingFinished/loadingFailed closes it. The wait ends once none remain and the page’s HTTP and websocket traffic has been quiet for idle_ms. blob: loads are not counted, and WebSocket/EventSource connections stay open by design, so those connections are excluded while their frames still reset the quiet period. Events from other attached tabs are ignored.

page = await Page.new(cdp=cdp)
await page.goto('https://httpbingo.org/forms/post', wait='idle')

from fastcore.test import *

Chrome reports a refused navigation in the command result (errorText) rather than as a protocol error, so goto checks for it and raises at once — the alternative is waiting out a load event that can never come. One notable refusal: transports that attach via chrome.debugger (the extension path) get renderer-initiated semantics, where top-frame data: URLs are banned. For “this HTML, in this page” — test fixtures, generated reports — set_content writes the document directly, no navigation involved.

CDP.set_content

async def set_content(
    html:str, sid:str=None
):

Replace the page’s document with html (via Page.setDocumentContent); no navigation happens

with ExceptionExpected(RuntimeError, 'ERR_NAME_NOT_RESOLVED'): await page.goto('https://nonexistent.invalid/')

await page.set_content('<h1>Receipt</h1><p>Order 42 confirmed</p>')
test_eq(await page.eval('document.querySelector("h1").innerText'), 'Receipt')

screenshot returns an IPython Image, so in a notebook the capture displays inline; full=True captures the whole scrollable page rather than the viewport.

CDP.screenshot

async def screenshot(
    sid:str=None, full:bool=False
):

Screenshot of the viewport, or the whole scrollable page if full

A short pause before capturing gives the compositor time to paint the freshly set content:

await asyncio.sleep(0.2)

img = await page.screenshot()
# img

await page.close()

await cdp.close()

LLMs and accessibility

cdp = await CDP.remote(port=lcdp.port)

page = await cdp.new_page()
await page.goto('https://httpbingo.org/forms/post')
await page.eval('document.title')

''

What the protocol gives back is a flat list of nodes – hundreds for even a small form, most of them unnamed wrappers and layout artifacts. Here is the raw material:

await page.accessibility.enable()
tree = await page.accessibility.getFullAXTree()
len(tree)

tree[0]

{'nodeId': '16',
 'ignored': False,
 'role': {'type': 'internalRole', 'value': 'RootWebArea'},
 'chromeRole': {'type': 'internalRole', 'value': 144},
 'name': {'type': 'computedString',
  'value': '',
  'sources': [{'type': 'relatedElement', 'attribute': 'aria-labelledby'},
   {'type': 'attribute', 'attribute': 'aria-label'},
   {'type': 'attribute', 'attribute': 'aria-label', 'superseded': True},
   {'type': 'relatedElement', 'nativeSource': 'title'}]},
 'properties': [{'name': 'focusable',
   'value': {'type': 'booleanOrUndefined', 'value': True}},
  {'name': 'focused', 'value': {'type': 'booleanOrUndefined', 'value': True}},
  {'name': 'url',
   'value': {'type': 'string', 'value': 'https://httpbingo.org/forms/post'}}],
 'childIds': ['17'],
 'backendDOMNodeId': 16,
 'frameId': '17E820913776A78C240F3DA0D5364667'}

await page.close()

AXNode is the tree form of one node, and its repr is the point: markdown bullet lines, one per node, showing role, name, the properties that are actually set, and the backend id that click and friends take. This section is called “LLMs and accessibility” because that rendering is the page summary an LLM driving the browser reads – compact, semantic, and carrying the interaction handles inline.

AXTree

def AXTree(
    raw:dict
):

Chrome accessibility tree node with compact repr

AXNode

def AXNode(
    raw:dict
):

Chrome accessibility tree node with compact repr

build_ax_tree links the flat list into a tree and then _simplify prunes it: an unnamed none/generic/paragraph node teaches nothing, so its children splice up into its place. That one rule removes most of the raw dump’s bulk without losing a control or a label, and _set_parents then wires the parent links that up and path walk.

build_ax_tree

def build_ax_tree(
    nodes:list
):

Build AXNode tree from flat CDP accessibility node list

ax_tree is the one-call read – enable, fetch, build, simplify – and the usual way in:

CDP.ax_tree

async def ax_tree(
    sid:str=None, # Session to read
    frame_id:str=None, # Frame to read; the session's main frame if None
):

Get the accessibility tree for a session or one of its frames

CDP.wait_for_child_frame

async def wait_for_child_frame(
    url:str, # Text contained in the frame URL
    sid:str=None, # Session to wait in
    timeout:float=10, # Seconds to wait before raising
):

Wait for a frame whose URL contains url and return its metadata

CDP.frames

async def frames(
    sid:str=None
):

Return the page’s current frames in tree order

CDP.frame_page

async def frame_page(
    url:str, # Text contained in the frame's URL
    sid:str=None, # Session whose frames to search
    timeout:float=10, # Seconds to wait before raising
)->Page: # Proxy bound to the frame: this session with the frame id filled in, or the frame's own session

A Page for the child frame whose URL contains url, wherever Chrome renders it

frames returns the page’s current frame tree and wait_for_child_frame polls it until a frame URL contains the requested text; the returned frame’s id goes to ax_tree or wait_for_ax, since a frame’s content is absent from its parent’s accessibility tree. Frames that Chrome renders in another process are absent from this tree altogether: they are iframe targets, with a session of their own. frame_page covers both placements.

page = await cdp.new_page()
await page.goto('https://httpbingo.org/forms/post')

root = await page.ax_tree()
root

RootWebArea “” focusable=True focused=True url=https://httpbingo.org/forms/post [#16]
- form “” [#2]
  - LabelText “” [#21]
    - StaticText “Customer name:” [#64]
      - InlineTextBox “Customer name:”
    - textbox “Customer name:” focusable=True editable=plaintext settable=True [#3]
  - LabelText “” [#24]
    - StaticText “Telephone:” [#65]
      - InlineTextBox “Telephone:”
    - textbox “Telephone:” focusable=True editable=plaintext settable=True [#4]
  - LabelText “” [#27]
    - StaticText “E-mail address:” [#66]
      - InlineTextBox “E-mail address:”
    - textbox “E-mail address:” focusable=True editable=plaintext settable=True [#5]
  - group “Pizza Size” [#29]
    - Legend “” [#30]
      - StaticText “Pizza Size” [#67]
        
        InlineTextBox “Pizza Size”
    - radio ” Small” focusable=True [#7]
    - radio ” Medium” focusable=True [#8]
    - radio ” Large” focusable=True [#9]
  - group “Pizza Toppings” [#37]
    - Legend “” [#38]
      - StaticText “Pizza Toppings” [#71]
        
        InlineTextBox “Pizza Toppings”
    - checkbox ” Bacon” focusable=True [#10]
    - checkbox ” Extra Cheese” focusable=True [#11]
    - checkbox ” Onion” focusable=True [#12]
    - checkbox ” Mushroom” focusable=True [#13]
  - LabelText “” [#48]
    - StaticText “Preferred delivery time:” [#76]
      - InlineTextBox “Preferred delivery time:”
    - InputTime “Preferred delivery time:” focusable=True settable=True [#14]
      - spinbutton “Hours Hours” focusable=True settable=True valuemin=1 valuemax=12 [#52]
        
        StaticText “–” [#77]
        
        InlineTextBox “–”
      - StaticText “:” [#78]
        
        InlineTextBox “:”
      - spinbutton “Minutes Minutes” focusable=True settable=True valuemax=59 [#54]
        
        StaticText “–” [#79]
        
        InlineTextBox “–”
      - StaticText ” ” [#80]
        
        InlineTextBox ” ”
      - spinbutton “AM/PM AM/PM” focusable=True settable=True valuemin=1 valuemax=2 [#56]
        
        StaticText “–” [#81]
        
        InlineTextBox “–”
      - button “Show time picker” focusable=True hasPopup=menu [#57]
  - LabelText “” [#59]
    - StaticText “Delivery instructions:” [#82]
      - InlineTextBox “Delivery instructions:”
    - textbox “Delivery instructions:” focusable=True editable=plaintext settable=True multiline=True [#6]
  - button “Submit order” focusable=True [#63]
    - StaticText “Submit order” [#83]
      - InlineTextBox “Submit order”

AXNode.find_all

def find_all(
    role:str=None, # Accessibility role to match exactly, e.g. 'button'
    name:str=None, # Substring of the accessible name to match
):

Find all descendants matching role and/or name substring

AXNode.find_id

def find_id(
    role:str=None, # Accessibility role to match exactly, e.g. 'button'
    name:str=None, # Substring of the accessible name to match
)->int: # The backend node id, for `click`, `fill_text` and friends; None if no match

Find first descendant matching role and/or name substring, and return its backend id

AXNode.find

def find(
    role:str=None, # Accessibility role to match exactly, e.g. 'button'
    name:str=None, # Substring of the accessible name to match
):

Find first descendant matching role and/or name substring

nmid = root.find_id('textbox', 'Customer name')
phid = root.find_id('textbox', 'Telephone')
nmid

find/find_id target a control you already know is there. Orientation on an unfamiliar page is the inverse problem — where is the content, and what surrounds it? grep regex-searches every node name and returns one line per hit: the #id drops straight into click/fill_text, and the ancestor path says where on the page the hit sits (InlineTextBox layout artifacts are skipped, since each just duplicates its parent’s text). From a leaf hit, up climbs to the enclosing widget, and view renders a subtree to a bounded depth. So the reading workflow is: grep to locate, up/view to read just that neighborhood — never the whole tree.

AXNode.grep

def grep(
    pattern:str='', # Regex over node names
    role:str=None, # Only nodes with this accessibility role, when given
    ignore_case:bool=True, # Case-insensitive match?
    max_results:int=20, # Stop after this many hits
)->AXMatches:

Regex-search descendant names, for orientation: hits carry ids and ancestor paths

AXMatches

def AXMatches(
    *args, **kwargs
):

grep hits, one line per node: id, role, name, ancestor path

AXNode.view

def view(
    depth:int=None
)->AXView:

Markdown subtree rooted here, to depth levels (None = unbounded)

AXView

def AXView(
    *args, **kwargs
):

A rendered subtree, displayed as markdown

AXNode.path

def path():

Ancestor chain as a ’ > ’ joined summary, root first

AXNode.up

def up(
    n:int=1
):

The nth ancestor (None past the root)

root.grep('pizza')

#29 group "Pizza Size" — RootWebArea > form
#67 StaticText "Pizza Size" — RootWebArea > form > group "Pizza Size" > Legend
#37 group "Pizza Toppings" — RootWebArea > form
#71 StaticText "Pizza Toppings" — RootWebArea > form > group "Pizza Toppings" > Legend

role= narrows a text match that lands on several node kinds, and view with a depth reads a hit’s neighborhood without dumping its whole subtree — elided levels end in …:

sz = root.grep('pizza size', role='group')[0]
assert sz.role == 'group'
sz.view(1)

group “Pizza Size” [#29]
- Legend “” [#30] …
- radio ” Small” focusable=True [#7]
- radio ” Medium” focusable=True [#8]
- radio ” Large” focusable=True [#9]

A hit is often a leaf inside the widget that matters: up climbs to it, and path names where in the page a node sits.

cheese = root.grep('extra cheese')[0]
assert (cheese.role, cheese.up().role) == ('checkbox', 'group')
cheese.path()

'RootWebArea > form > group "Pizza Toppings"'

The ax tree and the DOM speak different id spaces: ax nodes carry backend ids, which are what click, tap, and fill_text take, while the DOM/CSS domains want the front-end nodeId. sel_node resolves a CSS selector to a nodeId, and sel_backend_id crosses back the other way – selector to backend id – so the interaction verbs below can take a CSS address too. matched_styles answers the design question computed styles can’t – why an element looks the way it does – as every matching rule in cascade order (winners last) with its origin. It takes a selector or an ax backend id, so a grep or find_id hit can be interrogated directly.

CDP.matched_styles

async def matched_styles(
    target:str | int, sid:str=None
)->MatchedStyles:

Matching CSS rules for a selector or an ax backend node id, with each rule’s origin

MatchedStyles

def MatchedStyles(
    *args, **kwargs
):

Matched rules in cascade order (winners last), one line per rule

CDP.sel_backend_id

async def sel_backend_id(
    sel:str, sid:str=None
)->int:

Backend node id of the first element matching CSS selector sel, for click and friends

CDP.sel_node

async def sel_node(
    sel:str, sid:str=None
)->int:

DOM nodeId of the first element matching CSS selector sel

ms = await page.matched_styles('form')
assert ms and all(r.origin for r in ms)
ms

[user-agent] form 
[user-agent] address, blockquote, center, div, figur…

await page.matched_styles(root.find_id('button', 'Submit order'))

[user-agent] button 
[user-agent] input, textarea, select, button 
[user-agent] input[type="button" i], input[type="sub… 
[user-agent] input[type="button" i], input[type="sub…

Sometimes no protocol command does the job and the answer is JS on one node. js_node crosses the id gap – resolve a backend node id to a live object, then call a function with it as this – and js_node_run wraps plain statements in that function. Several helpers below are one js_node_run each.

CDP.js_node_run

async def js_node_run(
    code:str, # JS statements, with the node as `this`
    backendNodeId:int, # Node, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
):

Run code with a DOM node as this

CDP.js_node

async def js_node(
    fn:str, # JS function declaration, called with the node as `this`
    backendNodeId:int, # Node, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
):

Call function fn on a DOM node, returning the protocol result

Pointer events address viewport coordinates, so interaction starts with geometry: _node_center reads the center of a node’s content box, and _scroll_center scrolls the node to the viewport center first, which is the pair every pointer verb needs – an off-screen target would otherwise receive events at coordinates outside the window. scroll_to is also useful on its own, e.g. to bring a node on-screen before a viewport screenshot.

scroll_to returns only after two animation frames. The scroll moves the layout at once, but the compositor’s hit-test data, which routes pointer events between the page and any frames it isolates, follows on the next composited frame. Pressing before that frame lands the event on whatever the stale data placed at those coordinates: a Stripe card frame a scroll away from the button, say, so the button sees only the release.

CDP.scroll_to

async def scroll_to(
    backendNodeId:int, # Node, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
):

Scroll a node to the center of the viewport, and wait for the frame that composites the scroll

Activation has three deliberately separate paths. click engages hover first – scroll, then a real mouse move – and re-reads the center before pressing, since hover-gated UI can change the box. tap sends Chrome’s trusted tap gesture with no mouse move at all, for controls where hover handling is unwanted or unreliable. dom_click calls the element’s JavaScript activation, so it works on anything clickable but does not produce trusted user input. hover is also a verb in its own right, for UI that only appears under the pointer (demonstrated with the sel_* helpers below). Each verb is bounded by _bounded, so a swallowed event fails in seconds with the verb’s name rather than hanging a test.

CDP.dom_click

async def dom_click(
    backendNodeId:int, # Node, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
    timeout:float=5, # Maximum seconds to wait
):

Activate a node with its DOM click, bounded by timeout

CDP.tap

async def tap(
    backendNodeId:int, # Node, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
    timeout:float=5, # Maximum seconds for the whole tap
):

Activate a node with a trusted tap gesture, without mouse hover

CDP.click

async def click(
    backendNodeId:int, # Node, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
    timeout:float=5, # Maximum seconds for the whole click
):

Click a node with real pointer events, bounded by timeout

CDP.hover

async def hover(
    backendNodeId:int, # Node to hover, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
):

Scroll a node into view and move the mouse to its center, firing its hover events and CSS :hover

await page.DOM.focus(backendNodeId=nmid)
await page.input.insertText(text='Jeremy Howard')

await page.DOM.focus(backendNodeId=phid)
await page.input.insertText(text='555-1234')

{}

await page.click(root.find_id('radio', 'Large'))
await page.eval(r'''document.querySelector('input[type=checkbox]').addEventListener('click', e => sessionStorage.fastcdpTap = e.isTrusted, {once:true})''')
await page.tap(root.find_id('checkbox', 'Extra Cheese'))
test_eq(await page.eval('sessionStorage.fastcdpTap'), 'true')

CDP.fill_text

async def fill_text(
    backendNodeId:int, # The text control, e.g. from `AXNode.find_id`
    text:str, # Text to type into it
    sid:str=None, # Session the node lives in
):

Replace the contents of a text control

await page.fill_text(root.find_id('textbox', 'Delivery'), 'Ring the doorbell twice')

await page.js_node_run('this.value = "18:30"', root.find_id('InputTime', 'delivery time'))

{'type': 'undefined'}

fill_text inserts text without firing key events. Keyboard shortcuts and key-driven UI need the events themselves. press sends one real keydown/keyup pair with modifiers carried on the events, and type presses each character of a string. mod=True holds the platform primary modifier, resolved on the machine running Python: Command on macOS, Control elsewhere.

App-level keydown listeners see every chord. The browser’s own editing behavior does not: a synthetic Cmd+A selects nothing. With Meta held, press therefore also issues the matching editing command for a, c, x, v, z, and y.

CDP.type

async def type(
    text:str, # Characters to press, one key event pair each
    sid:str=None, # Session to send to
):

Press each character of text in turn; fill_text is faster when no per-key handling matters

CDP.press

async def press(
    key:str, # A single character, or a key name from `_keys` such as 'Enter'
    sid:str=None, # Session to send to
    ctrl:bool=False, # Hold Control
    shift:bool=False, # Hold Shift
    alt:bool=False, # Hold Alt/Option
    meta:bool=False, # Hold Meta/Command
    mod:bool=False, # Hold the platform primary modifier: Command on macOS, Control elsewhere
):

Send one key press, with modifiers, as real keydown/keyup events

did = root.find_id('textbox', 'Delivery')
await page.DOM.focus(backendNodeId=did)
await page.press('a', mod=True)
await page.type('Leave at door')
await page.press('Backspace')
val = await page.eval('document.querySelector("textarea").value')
test_eq(val, 'Leave at doo')
val

'Leave at doo'

A form submission combines two independent choices: how to activate the control, and what completion means. click_and_wait is the convenient common case — mouse input followed by a required navigation. Compose tap or dom_click with expect_navigation explicitly when either is the reliable activation path for a particular control.

CDP.click_and_wait

async def click_and_wait(
    backendNodeId:int, # The element to click, e.g. from `AXNode.find_id`
    sid:str=None, # Session the node lives in
    wait:str | None='load', # 'load', 'idle', or None to stop after navigation begins
    timeout:float=10, # Maximum seconds for the action and requested wait
    idle_ms:int=100, # Quiet time after load when `wait='idle'`
):

Click with real pointer events and wait for its navigation

await page.eval("sessionStorage.removeItem('fastcdpPointer'); document.querySelector('form').addEventListener('mousedown', () => sessionStorage.fastcdpPointer = '1', {once:true})")
await page.click_and_wait(root.find_id('button', 'Submit order'))
test_eq(await page.eval("sessionStorage.fastcdpPointer"), '1')

click_and_wait is the common mouse-click-plus-navigation operation; the assertion above confirms the real mousedown reached the form. For another activation path, compose the primitives explicitly: async with page.expect_navigation(): await page.tap(node_id). Both forms require a top-frame navigation, so they fail clearly when activation did nothing rather than mistaking the old document’s already-complete state for success.

When activation swaps content in place (tab panels, htmx, SPAs), there is no navigation to expect and the tree in hand goes stale. Use ordinary click, tap, or dom_click, then wait for the application result. wait_for_ax polls until a node matching role/name (as in find) exists and returns the fresh tree — the wait and the re-read are one call.

CDP.wait_for_ax

async def wait_for_ax(
    role:str=None, # Accessibility role to match exactly, as in `find`
    name:str=None, # Substring of the accessible name to match, as in `find`
    pred:callable=None, # Extra test a matching node must pass, e.g. `lambda n: not n.props.get('disabled')`
    sid:str=None, # Session to wait in
    frame_id:str=None, # Frame to read; the session's main frame if None
    timeout:int=10, # Seconds to wait before raising
):

Poll ax_tree until a node matches role/name (as in find) and pred; returns the fresh tree

await page.eval(r'setTimeout(() => { const h = document.createElement("h2"); h.textContent = "Order received"; document.body.append(h) }, 300)')
fresh = await page.wait_for_ax('heading', 'Order received')
fresh.find('heading', 'Order received').name

'Order received'

A node can be present before it is usable. Google Cloud Console’s support-email picker, for one, reports disabled in the accessibility tree while it loads, with no DOM attribute to select on. pred adds a test the matching node must pass, so the wait ends when the node is in the state the next action needs, not merely when it exists.

await page.eval(r'const b = document.createElement("button"); b.textContent = "Pay"; b.disabled = true; document.body.append(b); setTimeout(() => b.disabled = false, 300)')
fresh = await page.wait_for_ax('button', 'Pay', pred=lambda n: not n.props.get('disabled'))
fresh.find('button', 'Pay').props

Debugging

Helpers for debugging apps: buffered console and network history, auto-handled dialogs, and a few conveniences. CDP only delivers events once the relevant domain is enabled, so each start_* helper enables it and buffers from that moment on.

CDP.console

async def console(
    pattern:str=None, sid:str=None
):

Console/exception messages buffered since start_console, filtered by regex pattern

CDP.start_console

async def start_console(
    sid:str=None
):

Enable and start buffering console messages and uncaught exceptions

start_console begins capture, and console returns everything seen so far; error: entries are uncaught exceptions, with their stack. Messages logged before start_console was called are never seen, so call it right after creating a page.

page = await cdp.new_page()
await page.start_console()
await page.eval(r'console.log("hello", 42); console.warn("watch out")')
await page.eval(r'setTimeout(() => { throw new Error("boom") }, 0)')
await asyncio.sleep(0.1)
await page.console()

['log: hello 42',
 'warning: watch out',
 'error: Error: boom\n    at <anonymous>:1:26']

The pattern regex filters entries:

test_eq(await page.console(r'watch'), ['warning: watch out'])
assert any('boom' in s for s in await page.console(r'error:'))

CDP.response_body

async def response_body(
    requestId:str, sid:str=None
):

Body of a response seen by start_network, decoded if base64

CDP.requests

async def requests(
    pattern:str=None, sid:str=None
):

(status,url,requestId) of responses buffered since start_network, url filtered by regex pattern

CDP.start_network

async def start_network(
    sid:str=None
):

Enable and start buffering network responses

requests answers “what did the page load, and with what status?”, and response_body fetches a body by the returned request id (Chrome only keeps bodies while the page is alive).

await page.start_network()
await page.goto('https://httpbingo.org/forms/post')
await page.requests(r'httpbingo')

[(200, 'https://httpbingo.org/forms/post', 'F7CB668E1CCF95D17926A114457506DC')]

st,url,rid = first(await page.requests(r'forms/post'))
test_eq(st, 200)
assert '<form' in (await page.response_body(rid))

Websocket testing

Htmx apps stream their UI over one websocket, so when a swap never appears on screen the question is always which side failed: did the server not send the frame, or did the browser not apply it? start_ws buffers frames in both directions, and ws_frames returns them as WSFrame rows that parse the payload the way htmx will read it: each row lists the frame’s top-level elements with their id and hx-swap-oob, since those are the swap units. The list repr shows the gap between frames, because delivery timing is usually half the diagnosis. The buffer is a snapshot: a frame still in flight is simply not there yet, and a log read too early looks complete. So a test that depends on a frame waits for it with wait_for_frame, naming the payload and, where both directions could match, the direction with sent=. A small echo server, started here and closed at the end, stands in for the app:

WSFrame

def WSFrame(
    sent:bool, # Did the page send it (else receive)?
    ts:float, # The event's timestamp, seconds
    payload:str, # The frame's text payload
):

One captured websocket frame

CDP.wait_for_frame

async def wait_for_frame(
    pattern:str, # Regex the payload must match
    sent:bool=None, # True for frames the page sent, False for received, None for both
    sid:str=None, # Session whose frames to read
    timeout:int=10, # Seconds to wait before raising
)->WSFrames:

Poll ws_frames until a frame matches; returns the matching frames

CDP.ws_frames

async def ws_frames(
    pattern:str=None, # Regex the payload must match
    sent:bool=None, # True for frames the page sent, False for received, None for both
    sid:str=None, # Session whose frames to read
)->WSFrames:

Frames buffered since start_ws; a snapshot, so to read a frame that may still be in flight use wait_for_frame

CDP.start_ws

async def start_ws(
    sid:str=None
):

Enable and start buffering websocket frames, both directions

WSFrames

def WSFrames(
    *args, **kwargs
):

Captured frames, one per line with the gap since the frame before

from http import HTTPStatus

async def _echo(ws):
    async for m in ws: await ws.send(m)
def _page(conn, req):
    if 'Upgrade' not in req.headers: return conn.respond(HTTPStatus.OK, 'ws demo')
esrv = await websockets.serve(_echo, '127.0.0.1', 0, process_request=_page)
eport = esrv.sockets[0].getsockname()[1]

await page.goto(f'http://127.0.0.1:{eport}/')  # ws to loopback needs a loopback page origin
await page.start_ws()
await page.eval(f'''window._w = new WebSocket("ws://127.0.0.1:{eport}");
    _w.onopen = () => _w.send('<div id="msgs" hx-swap-oob="beforeend"><p>hi</p></div><span id="dot"></span>');''')
await page.wait_for_frame(r'hi', sent=False)  # the echo has come back
frames = await page.ws_frames()
frames

→ div#msgs[beforeend] span#dot
+0.000s ← div#msgs[beforeend] span#dot

test_eq(len(frames), 2)
assert frames[0].sent and not frames[1].sent
test_eq(frames[0].elements, [('div', 'msgs', 'beforeend'), ('span', 'dot', None)])
test_eq(len(await page.ws_frames(r'hi')), 2)
test_eq(len(await page.ws_frames(r'hi', sent=True)), 1)
test_eq(len(await page.ws_frames(r'nomatch')), 0)
esrv.close()

Test rungs

A browser test against a live app is a ladder of named steps. When one fails, the questions are which step, and what the page was doing at that moment. evidence answers the second: one report drawn from the debugging buffers, with a section for each capture that was started. Rung answers the first: any exception inside the context re-raises as an AssertionError naming the rung, with the page’s evidence attached. Rungs binds the page once and logs each rung’s duration. Display it for the ladder’s timing profile.

Rungs

def Rungs(
    page:NoneType=None, # `Page` (or `CDP`) passed to every rung
):

Rung factory sharing one page and a timing log; display it for per-rung times

Rung

def Rung(
    name:str, # Name of this step, quoted in the failure
    page:NoneType=None, # `Page` (or `CDP`) whose debugging buffers join the failure; None attaches nothing
    times:list=None, # Log gaining `(name, seconds)` on exit; `Rungs` supplies a shared one
):

Async context: a failure inside re-raises named after the rung, with the page’s evidence attached

CDP.evidence

async def evidence(
    pattern:str=None, sid:str=None
):

Debugging buffers as one report: console tail, error responses (urls matching pattern), and ws frames, for each capture that was started

await page.eval('console.warn("low disk")')
print(await page.evidence())

console: ['log: hello 42', 'warning: watch out', 'error: Error: boom\n    at <anonymous>:1:26', 'warning: low disk']
error responses: []
frames:
→ div#msgs[beforeend] span#dot
+0.000s ← div#msgs[beforeend] span#dot

A passing rung is silent. A failing one names itself and carries the report. Both add their duration to the shared log:

rungs = Rungs(page)
async with rungs('page renders'): test_eq(await page.eval('document.body.innerText'), 'ws demo')
with ExceptionExpected(AssertionError, 'seed loads'):
    async with rungs('seed loads'): test_eq(await page.eval('document.querySelectorAll("#nope").length'), 1)
rungs

  0.000 page renders
  0.001 seed loads

Helpers

CDP.handle_dialogs

async def handle_dialogs(
    accept:bool=True, # Answer each dialog with OK (True) or Cancel (False)
    text:str=None, # Text to enter into a `prompt` dialog
    sid:str=None, # Session whose dialogs to answer
):

Auto-respond to JS dialogs from now on, recording (type,message) in dialogs

Without a handler, an unexpected alert or confirm blocks the page – and any eval that triggered it – forever. After handle_dialogs, dialogs are answered as they open: accept=False dismisses them, and text fills prompts.

await page.handle_dialogs()
ok = await page.eval(r'confirm("Proceed?")')
ok, page.dialogs

(True, [('confirm', 'Proceed?')])

test_eq(ok, True)
test_eq(page.dialogs, [('confirm', 'Proceed?')])

CDP.wait_for_text

async def wait_for_text(
    text:str, # Text to look for
    present:bool=True, # Wait for it to appear (True) or to go away (False)
    sel:str=None, # CSS selector of the element to read; the page body if None
    sid:str=None, # Session to wait in
    timeout:int=10, # Seconds to wait before raising
):

Wait for text to appear in (or, with present=False, disappear from) the page body or one element

CDP.select_option

async def select_option(
    backendNodeId:int, # The `<select>` node, e.g. from `AXNode.find_id`
    value:str, # Option value to select
    sid:str=None, # Session the node lives in
):

Set a <select> element’s value and fire its change event

click on a <select> doesn’t open native dropdowns under CDP, so select_option sets the value directly (firing change so the app reacts). wait_for_text complements wait_for_selector when the interesting change is text, e.g. htmx swaps. And screenshot above takes full=True for the whole scrollable page.

await page.set_content('<select><option>small<option>large</select><div style="height:3000px"></div>')
await page.select_option((await page.ax_tree()).find_id('combobox'), 'large')
test_eq(await page.eval(r'document.querySelector("select").value'), 'large')

hs = [int.from_bytes(i.data[20:24], 'big') for i in (await page.screenshot(), await page.screenshot(full=True))]
assert hs[1] > hs[0]
hs

[469, 3035]

await page.eval(r'setTimeout(() => document.body.append(" loaded!"), 300)')
await page.wait_for_text('loaded!')
await page.eval(r'setTimeout(() => { document.body.innerHTML = "gone" }, 200)')
await page.wait_for_text('loaded!', present=False)
await page.eval(r'setTimeout(() => window.app = {ready: true}, 200)')
await page.wait_defined('app')

CDP.sel_click

async def sel_click(
    sel:str, sid:str=None
):

click the first element matching CSS selector sel

CDP.sel_exists

async def sel_exists(
    sel:str, sid:str=None
)->bool:

Whether any element matches CSS selector sel

CDP.sel_text

async def sel_text(
    sel:str, sid:str=None
):

textContent of the first element matching CSS selector sel, or None

On a page you already know, a CSS selector is the address, the way an ax grep hit is on one you don’t. The sel_* helpers take that address directly. sel_click is click on the first match. sel_text reads its textContent. sel_exists asks presence as a boolean, because a DOM element is not a value eval can return, so a bare querySelector in a wait_for looks right and is wrong. wait_for_selector takes present=False to wait for something to go away, matching wait_for_text, and wait_for_text takes sel= to scope to one region, which is what an htmx oob swap changes.

await page.set_content('<button onclick="out.textContent=\'done\'; spin.remove()">go</button><div id=out>waiting</div><span id=spin>...</span>')
test_eq(await page.sel_text('#out'), 'waiting')
test_eq(await page.sel_exists('#spin'), True)
await page.sel_click('button')
await page.wait_for_text('done', sel='#out')

True

The click removed the spinner and changed the text. wait_for_selector(present=False) is how a test waits for something to go away, and reading both back shows the end state.

await page.wait_for_selector('#spin', present=False)
(await page.sel_text('#out'), await page.sel_exists('#spin'))

('done', False)

Hover matters when UI only appears under the pointer: hidden row actions, tooltips, collapse chevrons. hover scrolls a node into view and moves the real mouse to its center, engaging CSS :hover and mouse events, and sel_hover takes the selector form. sel_attr and sel_count read an attribute and count matches. sel_map applies a JS function to every match, and sel_attrs is its attribute form: one call reads a column of the page, such as every row’s id.

CDP.sel_attrs

async def sel_attrs(
    sel:str, name:str, sid:str=None
)->list:

Attribute name of every element matching CSS selector sel, in document order

CDP.sel_map

async def sel_map(
    sel:str, fn:str, sid:str=None
)->list:

JS function fn applied to every element matching CSS selector sel, in document order

CDP.sel_count

async def sel_count(
    sel:str, sid:str=None
)->int:

Number of elements matching CSS selector sel

CDP.sel_attr

async def sel_attr(
    sel:str, name:str, sid:str=None
):

Attribute name of the first element matching CSS selector sel, or None

CDP.sel_hover

async def sel_hover(
    sel:str, sid:str=None
):

hover the first element matching CSS selector sel

await page.set_content('<style>#menu{display:none} #row:hover #menu{display:inline}</style><section style="height:3000px"></section><div id=row>item<span id=menu>edit</span></div><p class=x><p class=x>')
await page.sel_hover('#row')
test_eq(await page.eval('getComputedStyle(document.querySelector("#menu")).display'), 'inline')
await page.eval('window.md = 0; document.querySelector("#row").addEventListener("mousedown", () => md++)')
await page.sel_click('#row')
test_eq(await page.eval('md'), 1)
test_eq(await page.sel_attrs('.x', 'class'), ['x', 'x'])
test_eq(await page.sel_map('p,div', 'e => e.tagName'), ['DIV', 'P', 'P'])
(await page.sel_attr('#menu', 'id'), await page.sel_count('.x'))

('menu', 2)

Where Chrome renders a child frame depends on the host page as much as on the frame. Fixture HTML set on a fresh tab keeps even a cross-site frame in the tab’s process, so it shows up in frames, and frame_page returns a Page on the same session with the frame id bound: ax_tree and wait_for_ax then read that frame, and fill_text, click and the other node actions take its backend ids as usual.

await page.set_content('<h1>Host</h1><iframe src="https://example.com/"></iframe>')
fp = await page.frame_page('example.com')
test_eq(fp.sid, page.sid)
(await fp.ax_tree()).find('heading').name

Under a real site the same frame is isolated into its own process, so it is an iframe target rather than a frame of the page. frame_page then returns a Page on the frame’s own session, where eval and every other helper run inside the frame. The first call asks Chrome to auto-attach the page’s child frames, as they appear and recursively, so the search covers exactly this page’s frames, nested ones included, and never another tab’s. Stripe’s card elements are the everyday case: the form the user types into is a frame served from js.stripe.com, and the 3D Secure challenge is a frame inside that one.

await page.goto('https://example.org/')
await page.eval('document.body.appendChild(Object.assign(document.createElement("iframe"), {src: "https://example.com/"})).tagName')
fp = await page.frame_page('example.com')
assert fp.sid != page.sid
test_eq(await fp.eval('location.host'), 'example.com')
(await fp.ax_tree()).find('heading').name

await page.close()

await cdp.close()

To finish, exercise the browser we launched at the start end to end, then quit it:

page2 = await lcdp.new_page()
await page2.goto('https://example.com')
test_eq(await page2.eval('document.title'), 'Example Domain')

await lcdp.quit()
assert not lcdp.is_open and lcdp.proc.returncode is not None

Finally, sandbox registration: safepyrun runs LLM-written code under an allowlist, and allow is how a library declares which callables such code may use. cdp_yolo allowlists every fastcdp entry point wholesale, for sessions where driving the browser is the whole point.

cdp_yolo

def cdp_yolo():

Allow all CDP classes in safepyrun