FastCDP API

Source and API details 
len(_cdp_domains), [d['domain'] for d in _cdp_domains[:5]]
(55, ['Accessibility', 'Animation', 'Audits', 'Autofill', 'BackgroundService'])

source

cdp_conninfo

def cdp_conninfo(
    p:NoneType=None, d:NoneType=None
):

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. 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. 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/6e04b056-236c-45c3-b0d5-597d0e45723e'

source

CDP

def CDP(
    wsconn:NoneType=None, debug:bool=False
):

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.

# await cdp.close()

source

CDP.remote

async def remote(
    port:int=9223, debug:NoneType=None
):

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.

source

chrome_bin

def chrome_bin():

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

source

CDP.launch

async def launch(
    user_data_dir:NoneType=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

source

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=False)
lcdp.port
64002

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)

source

CDP.pages

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

The browser’s open page targets

source

Targets

def Targets(
    *args, **kwargs
):

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

await cdp('Target.createTarget', url='https://example.com')
'F4C2B457E5F37B006CD4C2F8D787DBC6'
ps = await cdp.pages
pg = first(p for p in ps if 'example.com' in p.url)
pg.title
''

source

CDPDomain

def CDPDomain(
    cdp, domain
):

Initialize self. See help(type(self)) for accurate signature.

source

CDPMethod

def CDPMethod(
    cdp, domain, method
):

Initialize self. See help(type(self)) for accurate signature.

source

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

source

CDP.attach

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

Attach to target tid

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

source

CDP.wait_event

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

Call self as a function.

on

def on(
    event
):

Call self as a function.

source

CDP.wait_for_selector

async def wait_for_selector(
    sel, sid:NoneType=None, timeout:int=10
):

Wait for CSS selector to match an element

source

CDP.wait_for

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

Wait for JS expression to be truthy, return its value

source

CDP.wait_load

async def wait_load(
    sid:NoneType=None, timeout:int=10
):

Call self as a function.

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)
True
await cdp.target.closeTarget(targetId=t)
True

source

PageDomain

def PageDomain(
    sid, domain
):

Initialize self. See help(type(self)) for accurate signature.

source

Page

def Page(
    cdp, t, sid, owned:bool=False
):

Initialize self. See help(type(self)) for accurate signature.

source

CDP.remote_page

async def remote_page(
    port:int=9223, debug:NoneType=None
):

Connect via remote debugging and return Page for the active tab

source

CDP.active_page

async def active_page():

Call self as a function.

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'

source

CDP.new_page

async def new_page():

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.

source

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

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

source

CDP.goto

async def goto(
    url, sid:NoneType=None, **kwargs
):

Navigate to url and wait for load+idle

wait_ready

def wait_ready(
    sid:NoneType=None, timeout:int=10, idle_ms:int=500
):

Context manager: subscribes before action, waits for load+idle after

source

CDP.wait_for_ready

async def wait_for_ready(
    sid:NoneType=None, timeout:int=10, idle_ms:int=500
):

Wait until network is idle for idle_ms

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

source

CDP.screenshot

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

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

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')
''
await page.accessibility.enable()
tree = await page.accessibility.getFullAXTree()
len(tree)
90
tree[0]
{'nodeId': '2',
 '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': ['18'],
 'backendDOMNodeId': 2,
 'frameId': 'B327EB775230754D1A31282EDBCC91BD'}
await page.close()

source

AXTree

def AXTree(
    raw
):

Chrome accessibility tree node with compact repr

source

AXNode

def AXNode(
    raw
):

Chrome accessibility tree node with compact repr

source

build_ax_tree

def build_ax_tree(
    nodes
):

Build AXNode tree from flat CDP accessibility node list

source

CDP.ax_tree

async def ax_tree(
    sid:NoneType=None
):

Get accessibility tree for session

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 [#2]
    • form “” [#5]
      • LabelText “” [#22]
        • StaticText “Customer name:” [#63]
          • InlineTextBox “Customer name:”
        • textbox “Customer name:” focusable=True editable=plaintext settable=True [#6]
      • LabelText “” [#25]
        • StaticText “Telephone:” [#64]
          • InlineTextBox “Telephone:”
        • textbox “Telephone:” focusable=True editable=plaintext settable=True [#7]
      • LabelText “” [#28]
        • StaticText “E-mail address:” [#65]
          • InlineTextBox “E-mail address:”
        • textbox “E-mail address:” focusable=True editable=plaintext settable=True [#8]
      • group “Pizza Size” [#30]
        • Legend “” [#31]
          • StaticText “Pizza Size” [#66]
            • InlineTextBox “Pizza Size”
        • radio ” Small” focusable=True [#10]
        • radio ” Medium” focusable=True [#11]
        • radio ” Large” focusable=True [#12]
      • group “Pizza Toppings” [#38]
        • Legend “” [#39]
          • StaticText “Pizza Toppings” [#70]
            • InlineTextBox “Pizza Toppings”
        • checkbox ” Bacon” focusable=True [#13]
        • checkbox ” Extra Cheese” focusable=True [#14]
        • checkbox ” Onion” focusable=True [#15]
        • checkbox ” Mushroom” focusable=True [#16]
      • LabelText “” [#49]
        • StaticText “Preferred delivery time:” [#75]
          • InlineTextBox “Preferred delivery time:”
        • InputTime “Preferred delivery time:” focusable=True settable=True [#17]
          • spinbutton “Hours Hours” focusable=True settable=True valuemin=1 valuemax=12 [#53]
            • StaticText “–” [#76]
              • InlineTextBox “–”
          • StaticText “:” [#77]
            • InlineTextBox “:”
          • spinbutton “Minutes Minutes” focusable=True settable=True valuemax=59 [#55]
            • StaticText “–” [#78]
              • InlineTextBox “–”
          • StaticText ” ” [#79]
            • InlineTextBox ” ”
          • spinbutton “AM/PM AM/PM” focusable=True settable=True valuemin=1 valuemax=2 [#57]
            • StaticText “–” [#80]
              • InlineTextBox “–”
          • button “Show time picker” focusable=True hasPopup=menu [#3]
      • LabelText “” [#59]
        • StaticText “Delivery instructions:” [#81]
          • InlineTextBox “Delivery instructions:”
        • textbox “Delivery instructions:” focusable=True editable=plaintext settable=True multiline=True [#9]
      • button “Submit order” focusable=True [#62]
        • StaticText “Submit order” [#82]
          • InlineTextBox “Submit order”

source

AXNode.find_all

def find_all(
    role:NoneType=None, name:NoneType=None
):

Find all descendants matching role and/or name substring

source

AXNode.find_id

def find_id(
    role:NoneType=None, name:NoneType=None
):

Find first descendant matching role and/or name substring

source

AXNode.find

def find(
    role:NoneType=None, name:NoneType=None
):

Find first descendant matching role and/or name substring

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

source

CDP.click

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

Click a DOM node

source

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

source

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

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.click(root.find_id('checkbox', 'Extra Cheese'))

source

CDP.fill_text

async def fill_text(
    backendNodeId, text, sid:NoneType=None
):

Call self as a function.

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'}

source

CDP.click_and_wait

async def click_and_wait(
    backendNodeId, sid:NoneType=None, **kwargs
):

Click element and wait for load+idle

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

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.

source

CDP.console

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

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

source

CDP.start_console

async def start_console(
    sid:NoneType=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.

from fastcore.test import *
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:'))

source

CDP.response_body

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

Body of a response seen by start_network, decoded if base64

source

CDP.requests

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

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

source

CDP.start_network

async def start_network(
    sid:NoneType=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', 'A0012C0BE59AE318F96FC9B3C00B7D1A')]
st,url,rid = first(await page.requests(r'forms/post'))
test_eq(st, 200)
assert '<form' in (await page.response_body(rid))

source

CDP.handle_dialogs

async def handle_dialogs(
    accept:bool=True, text:NoneType=None, sid:NoneType=None
):

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?')])

source

CDP.wait_for_text

async def wait_for_text(
    text, present:bool=True, sid:NoneType=None, timeout:int=10
):

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

source

CDP.select_option

async def select_option(
    backendNodeId, value, sid:NoneType=None
):

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.goto('data:text/html,<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
[884, 6070]
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)
True
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

source

cdp_yolo

def cdp_yolo():

Allow all CDP classes in safepyrun