Sending CDP Commands
Use thesend() method on Tab or Browser objects to send CDP commands:
import nodriver as uc
from nodriver import cdp
async def main():
browser = await uc.start()
tab = browser.main_tab
# Send CDP command
result = await tab.send(cdp.page.navigate(url='https://example.com'))
browser.stop()
Command Structure
CDP commands are generator objects created by calling methods from CDP domain modules:from nodriver import cdp
# Create command (returns generator)
command = cdp.page.navigate(url='https://example.com')
# Send command (await the result)
result = await tab.send(command)
Return Values
CDP commands return different types based on the command:from nodriver import cdp
# Single value
html = await tab.send(cdp.page.get_frame_tree())
# Multiple values (tuple unpacking)
frame_id, loader_id, error_text = await tab.send(
cdp.page.navigate(url='https://example.com')
)
# No return value (None)
await tab.send(cdp.page.enable())
Common CDP Tasks
Navigation
from nodriver import cdp
# Navigate to URL
frame_id, loader_id, error = await tab.send(
cdp.page.navigate(url='https://example.com')
)
# Go back
await tab.send(cdp.page.navigate_to_history_entry(entry_id=-1))
# Reload page
await tab.send(cdp.page.reload(ignore_cache=True))
# Stop loading
await tab.send(cdp.page.stop_loading())
Screenshots
from nodriver import cdp
import base64
# Capture viewport screenshot
screenshot_base64 = await tab.send(
cdp.page.capture_screenshot(
format='png',
quality=90
)
)
# Save screenshot
with open('screenshot.png', 'wb') as f:
f.write(base64.b64decode(screenshot_base64))
# Full page screenshot
await tab.send(cdp.emulation.set_device_metrics_override(
width=1280,
height=1024,
device_scale_factor=1,
mobile=False
))
screenshot = await tab.send(
cdp.page.capture_screenshot(
format='png',
capture_beyond_viewport=True
)
)
JavaScript Execution
from nodriver import cdp
# Evaluate expression
result = await tab.send(
cdp.runtime.evaluate(
expression='2 + 2',
return_by_value=True
)
)
print(result.value) # 4
# Execute async JavaScript
result = await tab.send(
cdp.runtime.evaluate(
expression='''
new Promise(resolve => {
setTimeout(() => resolve('done'), 1000);
})
''',
await_promise=True,
return_by_value=True
)
)
print(result.value) # 'done'
# Call function on object
remote_object = await tab.send(
cdp.runtime.evaluate(
expression='document',
return_by_value=False
)
)
result = await tab.send(
cdp.runtime.call_function_on(
function_declaration='function() { return this.title; }',
object_id=remote_object.object_id,
return_by_value=True
)
)
print(result.value) # Page title
DOM Manipulation
from nodriver import cdp
# Get document
doc = await tab.send(cdp.dom.get_document(depth=-1, pierce=True))
# Query selector
node_id = await tab.send(
cdp.dom.query_selector(
node_id=doc.node_id,
selector='button.submit'
)
)
# Get outer HTML
html = await tab.send(
cdp.dom.get_outer_html(node_id=node_id)
)
# Set attribute
await tab.send(
cdp.dom.set_attribute_value(
node_id=node_id,
name='class',
value='button primary'
)
)
# Remove node
await tab.send(cdp.dom.remove_node(node_id=node_id))
Network Monitoring
from nodriver import cdp
# Enable network tracking
await tab.send(cdp.network.enable())
# Set user agent
await tab.send(
cdp.network.set_user_agent_override(
user_agent='Custom User Agent/1.0'
)
)
# Block URLs
await tab.send(
cdp.network.set_blocked_ur_ls(
urls=['*://*.ads.com/*', '*://*.tracking.com/*']
)
)
# Get cookies
cookies = await tab.send(cdp.network.get_all_cookies())
for cookie in cookies:
print(f"{cookie.name}: {cookie.value}")
# Set cookie
await tab.send(
cdp.network.set_cookie(
name='session',
value='abc123',
domain='example.com',
path='/',
secure=True,
http_only=True
)
)
Input Simulation
from nodriver import cdp
# Mouse click
await tab.send(
cdp.input_.dispatch_mouse_event(
type_='mousePressed',
x=100,
y=200,
button='left',
click_count=1
)
)
await tab.send(
cdp.input_.dispatch_mouse_event(
type_='mouseReleased',
x=100,
y=200,
button='left',
click_count=1
)
)
# Keyboard input
await tab.send(
cdp.input_.dispatch_key_event(
type_='keyDown',
text='Hello'
)
)
# Insert text
await tab.send(
cdp.input_.insert_text(text='Hello World')
)
Emulation
from nodriver import cdp
# Set geolocation
await tab.send(
cdp.emulation.set_geolocation_override(
latitude=37.7749,
longitude=-122.4194,
accuracy=100
)
)
# Set timezone
await tab.send(
cdp.emulation.set_timezone_override(
timezone_id='America/New_York'
)
)
# Emulate mobile device
await tab.send(
cdp.emulation.set_device_metrics_override(
width=375,
height=812,
device_scale_factor=3,
mobile=True
)
)
# Set touch emulation
await tab.send(
cdp.emulation.set_touch_emulation_enabled(
enabled=True,
max_touch_points=5
)
)
Event Handling
Listen to CDP events using event handlers:from nodriver import cdp
async def main():
browser = await uc.start()
tab = await browser.get('https://example.com')
# Enable network domain
await tab.send(cdp.network.enable())
# Define event handler
def on_request(event: cdp.network.RequestWillBeSent):
print(f"Request: {event.request.method} {event.request.url}")
def on_response(event: cdp.network.ResponseReceived):
print(f"Response: {event.response.status} {event.response.url}")
# Add handlers
tab.add_handler(cdp.network.RequestWillBeSent, on_request)
tab.add_handler(cdp.network.ResponseReceived, on_response)
# Trigger events by navigating
await tab.get('https://example.org')
await tab.wait(3)
browser.stop()
Common Events
from nodriver import cdp
# Page events
cdp.page.LoadEventFired
cdp.page.DomContentEventFired
cdp.page.FrameNavigated
cdp.page.FrameStartedLoading
# Network events
cdp.network.RequestWillBeSent
cdp.network.ResponseReceived
cdp.network.LoadingFinished
cdp.network.LoadingFailed
# Console events
cdp.log.EntryAdded
# Runtime events
cdp.runtime.ConsoleAPICalled
cdp.runtime.ExceptionThrown
Request Interception
Intercept and modify network requests:from nodriver import cdp
async def main():
browser = await uc.start()
tab = await browser.get('https://example.com')
# Enable fetch domain for interception
await tab.send(
cdp.fetch.enable(
patterns=[
cdp.fetch.RequestPattern(
url_pattern='*',
resource_type=cdp.network.ResourceType.DOCUMENT
)
]
)
)
# Handle intercepted requests
def on_request_paused(event: cdp.fetch.RequestPaused):
# Modify headers
headers = event.request.headers.copy()
headers['X-Custom-Header'] = 'MyValue'
# Continue request with modified headers
asyncio.create_task(
tab.send(
cdp.fetch.continue_request(
request_id=event.request_id,
headers=headers
)
)
)
tab.add_handler(cdp.fetch.RequestPaused, on_request_paused)
await tab.get('https://example.org')
await tab.wait(3)
browser.stop()
Performance Monitoring
from nodriver import cdp
# Enable performance tracking
await tab.send(cdp.performance.enable())
# Get metrics
metrics = await tab.send(cdp.performance.get_metrics())
for metric in metrics:
print(f"{metric.name}: {metric.value}")
# Performance timeline
await tab.send(cdp.performance_timeline.enable(
event_types=['mark', 'measure']
))
Storage Access
from nodriver import cdp
# Local storage
await tab.send(cdp.dom_storage.enable())
storage_id = cdp.dom_storage.StorageId(
security_origin='https://example.com',
is_local_storage=True
)
# Get items
items = await tab.send(
cdp.dom_storage.get_dom_storage_items(storage_id=storage_id)
)
# Set item
await tab.send(
cdp.dom_storage.set_dom_storage_item(
storage_id=storage_id,
key='theme',
value='dark'
)
)
# Clear storage
await tab.send(
cdp.dom_storage.clear(storage_id=storage_id)
)
Advanced Examples
Capture network response bodies
from nodriver import cdp
import asyncio
async def main():
browser = await uc.start()
tab = await browser.get('https://example.com')
await tab.send(cdp.network.enable())
responses = {}
async def on_response(event: cdp.network.ResponseReceived):
request_id = event.request_id
url = event.response.url
try:
body = await tab.send(
cdp.network.get_response_body(request_id=request_id)
)
responses[url] = body
except:
pass
tab.add_handler(cdp.network.ResponseReceived, on_response)
await tab.get('https://api.example.com/data')
await tab.wait(3)
for url, body in responses.items():
print(f"URL: {url}")
print(f"Body: {body[:100]}...") # First 100 chars
browser.stop()
Custom device emulation
from nodriver import cdp
async def emulate_iphone(tab):
# Set viewport
await tab.send(
cdp.emulation.set_device_metrics_override(
width=390,
height=844,
device_scale_factor=3,
mobile=True,
)
)
# Set user agent
await tab.send(
cdp.emulation.set_user_agent_override(
user_agent='Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X) AppleWebKit/605.1.15',
platform='iPhone'
)
)
# Enable touch
await tab.send(
cdp.emulation.set_touch_emulation_enabled(
enabled=True,
max_touch_points=5
)
)
Best Practices
Enable domains before use - Many CDP commands require enabling their domain first with
domain.enable().Handle async events - Event handlers may need to use
asyncio.create_task() for async operations.Check return types - CDP commands return different types. Check the type hints or documentation.
Cleanup handlers - Remove event handlers when done to prevent memory leaks.
See Also
- CDP Overview - Introduction to CDP in nodriver
- Chrome DevTools Protocol - Official CDP documentation
- Tab class - Tab methods including
send()