Skip to main content
Zendriver’s intercept functionality allows you to pause, inspect, modify, or mock network requests and responses using the Chrome DevTools Protocol Fetch domain.

​
Overview

The BaseFetchInterception class provides a context manager for intercepting network requests at different stages:
  • Intercept requests before they’re sent
  • Intercept responses before they’re received
  • Modify request headers, method, URL, or body
  • Modify response status, headers, or body
  • Mock responses completely
  • Block requests

​
Creating an interception

from zendriver.core.intercept import BaseFetchInterception
from zendriver.cdp.fetch import RequestStage
from zendriver.cdp.network import ResourceType

# Intercept requests
interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*api/users.*",
    request_stage=RequestStage.REQUEST,
    resource_type=ResourceType.XHR
)

async with interception as i:
    # Make requests that match the pattern
    await tab.get("https://example.com/users")
    
    # Access intercepted request
    request = await i.request
    print(f"Intercepted: {request.url}")
    
    # Continue the request
    await i.continue_request()

​
Parameters

​
tab
Connection
required
The tab or connection to monitor for requests.
​
url_pattern
str
required
The URL pattern to match. Uses CDP pattern syntax (supports wildcards).
​
request_stage
RequestStage
required
When to intercept: RequestStage.REQUEST (before send) or RequestStage.RESPONSE (after receive).
​
resource_type
ResourceType
required
Type of resource to intercept: ResourceType.XHR, ResourceType.FETCH, ResourceType.DOCUMENT, etc.

​
Properties

​
request

Get the intercepted request.
​
request
cdp.network.Request
The intercepted request object (async property).
async with interception as i:
    await tab.get("https://example.com")
    
    request = await i.request
    print(f"URL: {request.url}")
    print(f"Method: {request.method}")
    print(f"Headers: {request.headers}")

​
response_body

Get the response body (only available when intercepting responses).
​
response_body
tuple[str, bool]
The response body and whether it’s base64 encoded (async property).
# Intercept response
interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*api/data.*",
    request_stage=RequestStage.RESPONSE,
    resource_type=ResourceType.XHR
)

async with interception as i:
    await tab.get("https://example.com")
    
    body, is_base64 = await i.response_body
    if is_base64:
        import base64
        body = base64.b64decode(body).decode()
    print(f"Response: {body}")

​
Methods

​
continue_request()

Continue the intercepted request, optionally modifying it.
​
url
str | None
default:"None"
Override the request URL.
​
method
str | None
default:"None"
Override the request method (GET, POST, etc.).
​
post_data
str | None
default:"None"
Override the POST data.
​
headers
List[HeaderEntry] | None
default:"None"
Override request headers.
​
intercept_response
bool | None
default:"None"
Whether to also intercept the response for this request.
from zendriver.cdp.fetch import HeaderEntry

async with interception as i:
    await tab.get("https://example.com")
    
    # Continue with modifications
    await i.continue_request(
        headers=[
            HeaderEntry(name="Authorization", value="Bearer token123"),
            HeaderEntry(name="Custom-Header", value="value")
        ]
    )

​
fulfill_request()

Fulfill the request with a mock response.
​
response_code
int
required
HTTP status code (e.g., 200, 404).
​
response_headers
List[HeaderEntry] | None
default:"None"
Response headers.
​
body
str | None
default:"None"
Response body (base64 encoded if binary).
​
response_phrase
str | None
default:"None"
HTTP status phrase (e.g., “OK”, “Not Found”).
import json
import base64

async with interception as i:
    await tab.get("https://example.com/api/users")
    
    # Mock response
    mock_data = json.dumps([{"id": 1, "name": "Test User"}])
    await i.fulfill_request(
        response_code=200,
        response_headers=[
            HeaderEntry(name="Content-Type", value="application/json")
        ],
        body=base64.b64encode(mock_data.encode()).decode()
    )

​
continue_response()

Continue the intercepted response, optionally modifying it.
​
response_code
int | None
default:"None"
Override the status code.
​
response_phrase
str | None
default:"None"
Override the status phrase.
​
response_headers
List[HeaderEntry] | None
default:"None"
Override response headers.
# Intercept response stage
interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*",
    request_stage=RequestStage.RESPONSE,
    resource_type=ResourceType.XHR
)

async with interception as i:
    await tab.get("https://example.com")
    
    # Modify response headers
    await i.continue_response(
        response_headers=[
            HeaderEntry(name="X-Modified", value="true")
        ]
    )

​
fail_request()

Fail the request with an error.
​
error_reason
cdp.network.ErrorReason
required
The error reason (e.g., ErrorReason.FAILED, ErrorReason.ABORTED).
from zendriver.cdp.network import ErrorReason

async with interception as i:
    await tab.get("https://example.com/api/data")
    
    # Block the request
    await i.fail_request(error_reason=ErrorReason.BLOCKED_BY_CLIENT)

​
reset()

Reset the interception to reuse it for another request. 
interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*api/.*",
    request_stage=RequestStage.REQUEST,
    resource_type=ResourceType.XHR
)

async with interception as i:
    # First request
    await tab.get("https://example.com/page1")
    await i.continue_request()
    
    # Reset for second request
    await i.reset()
    
    # Second request
    await tab.get("https://example.com/page2")
    await i.continue_request()

​
Usage examples

​
Modify request headers

from zendriver.cdp.fetch import RequestStage, HeaderEntry
from zendriver.cdp.network import ResourceType
from zendriver.core.intercept import BaseFetchInterception

interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*api/.*",
    request_stage=RequestStage.REQUEST,
    resource_type=ResourceType.XHR
)

async with interception as i:
    await tab.get("https://example.com")
    
    # Add authentication header
    await i.continue_request(
        headers=[
            HeaderEntry(name="Authorization", value="Bearer secret-token")
        ]
    )

​
Mock API response

import json
import base64

interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*api/users.*",
    request_stage=RequestStage.REQUEST,
    resource_type=ResourceType.XHR
)

async with interception as i:
    await tab.get("https://example.com/users")
    
    # Return mock data
    mock_users = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
    mock_json = json.dumps(mock_users)
    
    await i.fulfill_request(
        response_code=200,
        response_headers=[
            HeaderEntry(name="Content-Type", value="application/json")
        ],
        body=base64.b64encode(mock_json.encode()).decode()
    )

​
Block requests

from zendriver.cdp.network import ErrorReason

# Block all image requests
interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*\\.(png|jpg|jpeg|gif)$",
    request_stage=RequestStage.REQUEST,
    resource_type=ResourceType.IMAGE
)

async with interception as i:
    await tab.get("https://example.com")
    await i.fail_request(error_reason=ErrorReason.BLOCKED_BY_CLIENT)

​
Modify response

import json
import base64

# Intercept response
interception = BaseFetchInterception(
    tab=tab,
    url_pattern=".*api/data.*",
    request_stage=RequestStage.RESPONSE,
    resource_type=ResourceType.XHR
)

async with interception as i:
    await tab.get("https://example.com")
    
    # Get original response
    body, is_base64_encoded = await i.response_body
    if is_base64_encoded:
        body = base64.b64decode(body).decode()
    
    # Modify data
    data = json.loads(body)
    data["modified"] = True
    
    # Send modified response
    modified_body = json.dumps(data)
    await i.fulfill_request(
        response_code=200,
        response_headers=[
            HeaderEntry(name="Content-Type", value="application/json")
        ],
        body=base64.b64encode(modified_body.encode()).decode()
    )

​
Redirect requests

async with interception as i:
    await tab.get("https://example.com/old-api")
    
    # Redirect to new URL
    await i.continue_request(
        url="https://example.com/new-api"
    )

​
Change request method

async with interception as i:
    await tab.get("https://example.com/form")
    
    # Change GET to POST
    await i.continue_request(
        method="POST",
        post_data=json.dumps({"key": "value"}),
        headers=[
            HeaderEntry(name="Content-Type", value="application/json")
        ]
    )

​
Resource types

Common resource types to intercept:
  • ResourceType.DOCUMENT - HTML pages
  • ResourceType.XHR - XMLHttpRequest
  • ResourceType.FETCH - Fetch API requests
  • ResourceType.SCRIPT - JavaScript files
  • ResourceType.STYLESHEET - CSS files
  • ResourceType.IMAGE - Images
  • ResourceType.MEDIA - Audio/video
  • ResourceType.FONT - Web fonts

​
Request stages

  • RequestStage.REQUEST - Intercept before request is sent
  • RequestStage.RESPONSE - Intercept after response is received

​
Source code

For implementation details, see zendriver/core/intercept.py line 10

Build docs developers (and LLMs) love

Get started for freeTalk to us