CDP Usage

Sending CDP Commands

Use the send() 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

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.

Core API

CDP Integration

Utilities

Sending CDP Commands

Command Structure

Return Values

Common CDP Tasks

Navigation

Screenshots

JavaScript Execution

DOM Manipulation

Network Monitoring

Input Simulation

Emulation

Event Handling

Common Events

Request Interception

Performance Monitoring

Storage Access

Advanced Examples

Capture network response bodies

Custom device emulation

Best Practices

See Also

​Sending CDP Commands

​Command Structure

​Return Values

​Common CDP Tasks

​Navigation

​Screenshots

​JavaScript Execution

​DOM Manipulation

​Network Monitoring

​Input Simulation

​Emulation

​Event Handling

​Common Events

​Request Interception

​Performance Monitoring

​Storage Access

​Advanced Examples

​Capture network response bodies

​Custom device emulation

​Best Practices

​See Also

Sending CDP Commands

Command Structure

Return Values

Common CDP Tasks

Navigation

Screenshots

JavaScript Execution

DOM Manipulation

Network Monitoring

Input Simulation

Emulation

Event Handling

Common Events

Request Interception

Performance Monitoring

Storage Access

Advanced Examples

Capture network response bodies

Custom device emulation

Best Practices

See Also