Module Reference¶
Autodoc-generated reference for the public cycletls API. See the
API Usage guide for usage examples.
Clients¶
- class cycletls.CycleTLS(use_batching=False, batch_size=32, flush_interval=0.0001)[source]¶
CycleTLS client for making HTTP requests with advanced TLS fingerprinting.
- __init__(use_batching=False, batch_size=32, flush_interval=0.0001)[source]¶
Initialize CycleTLS client.
- Parameters:
use_batching (
bool, default:False) – If True, route individual requests through the batch FFI path for higher throughput.batch_size (
int, default:32) – Maximum number of requests to accumulate per batch.flush_interval (
float, default:0.0001) – Maximum time (seconds) to wait before flushing a partial batch.
- async abatch(requests)[source]¶
Send multiple requests concurrently as an async batch.
Each request is submitted asynchronously and all run in parallel.
- async aclose()[source]¶
Async close for httpx-compatible async cleanup.
This is the async counterpart of
close(). It allows usage with async cleanup patterns likeawait client.aclose().Directly sets
_closedrather than callingclose()so that subclasses that overrideclose()as a coroutine (e.g.AsyncCycleTLS) work without needing to be awaited separately.- Return type:
- async adelete(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async DELETE request.
- Return type:
- async aget(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async GET request.
- Return type:
- async ahead(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async HEAD request.
- Return type:
- async aoptions(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async OPTIONS request.
- Return type:
- async apatch(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async PATCH request.
- Return type:
- async apost(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async POST request.
- Return type:
- async aput(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an async PUT request.
- Return type:
- async arequest(method, url, params=None, data=None, json_data=None, json=None, files=None, fingerprint=None, auth=None, poll_interval=0.0, timeout=30.0, use_callback=False, **kwargs)[source]¶
Send an async HTTP request with enhanced data handling.
This is the async counterpart of
request(). It shares the same request preparation logic and adds async-specific polling parameters.- Parameters:
method (
str) – HTTP method (GET, POST, etc.)url (
str) – Target URLparams (
Union[Mapping[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]],Sequence[Tuple[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]]],None], default:None) – Query parameters to append to URLdata (
Union[Dict[str,Any],str,bytes,None], default:None) – Form data or raw bodyjson_data (
Optional[Dict[str,Any]], default:None) – JSON data (auto-serialized)json (
Optional[Dict[str,Any]], default:None) – Alias for json_data (matching requests library convention)files (
Optional[Dict[str,Any]], default:None) – File uploads (multipart form data)fingerprint (
Union[str,TLSFingerprint,None], default:None) – TLS fingerprint profile name or TLSFingerprint instance.auth (
Optional[Tuple[str,str]], default:None) – (username, password) tuple for HTTP Basic authenticationpoll_interval (
float, default:0.0) – Time between completion checks (default: 0s = adaptive)timeout (
float, default:30.0) – Maximum wait time for request completion (default: 30s)use_callback (
bool, default:False) – Use pipe-based notification instead of polling. Reduces FFI calls but creates an OS pipe per request, so it is slower under heavy concurrency. DefaultFalse(polling).**kwargs (
Any) – Additional CycleTLS options (headers, cookies, proxy, etc.)
- Returns:
Response object with status, headers, body, cookies, etc.
- Return type:
- batch(requests)[source]¶
Send multiple requests in a single batch.
This is more efficient than individual requests for high-throughput scenarios because it amortizes FFI call overhead across all requests. Requests are executed in parallel on the Go side.
- Parameters:
requests (
list[dict]) – List of request dictionaries. Each dict should contain: - url (required): Target URL - method (required): HTTP method (GET, POST, etc.) - params (optional): Query parameters - data (optional): Form data or raw body - json_data (optional): JSON data (auto-serialized) - headers (optional): Custom headers - And any other CycleTLS options (ja3, proxy, timeout, etc.)- Return type:
- Returns:
List of Response objects in the same order as input requests.
- Raises:
CycleTLSError – If the client is closed or batch request fails.
Example
>>> with CycleTLS() as client: ... responses = client.batch([ ... {"url": "https://api.example.com/users", "method": "GET"}, ... {"url": "https://api.example.com/posts", "method": "GET"}, ... {"url": "https://api.example.com/data", "method": "POST", "json_data": {"key": "value"}}, ... ]) ... for resp in responses: ... print(resp.status_code)
- delete(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a DELETE request.
- Return type:
- get(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a GET request.
- Return type:
- head(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a HEAD request.
- Return type:
- options(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an OPTIONS request.
- Return type:
- patch(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a PATCH request.
- Return type:
- post(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a POST request.
- Return type:
- put(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a PUT request.
- Return type:
- request(method, url, params=None, data=None, json_data=None, json=None, files=None, fingerprint=None, auth=None, **kwargs)[source]¶
Send an HTTP request with enhanced data handling.
- Parameters:
method (
str) – HTTP method (GET, POST, etc.)url (
str) – Target URLparams (
Union[Mapping[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]],Sequence[Tuple[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]]],None], default:None) – Query parameters to append to URLdata (
Union[Dict[str,Any],str,bytes,None], default:None) – Form data or raw bodyjson_data (
Optional[Dict[str,Any]], default:None) – JSON data (auto-serialized)json (
Optional[Dict[str,Any]], default:None) – Alias for json_data (matching requests library convention)files (
Optional[Dict[str,Any]], default:None) – File uploadsfingerprint (
Union[str,TLSFingerprint,None], default:None) – TLS fingerprint profile name or TLSFingerprint instance. If a string, looks up the profile in FingerprintRegistry. Applies the profile’s ja3, user_agent, header_order, etc.auth (
Optional[Tuple[str,str]], default:None) – (username, password) tuple for HTTP Basic authentication**kwargs (
Any) –Additional CycleTLS options. Notable options include:
headers: dict of extra HTTP headers
cookies: list of cookie dicts
ja3: JA3 TLS fingerprint string
user_agent: User-Agent header value
proxy: proxy URL (e.g. “http://user:pass@host:port”)
local_address: local IP address to bind the outgoing TCP connection to (useful on multi-homed hosts)
timeout: request timeout in seconds
insecure_skip_verify: skip TLS certificate verification
enable_connection_reuse: reuse persistent connections (default True)
- Returns:
Response object with status, headers, body, cookies, etc.
- Return type:
Example
>>> # Use a built-in fingerprint profile >>> response = client.get("https://example.com", fingerprint="chrome_120") >>> >>> # Or use a custom TLSFingerprint instance >>> from cycletls.fingerprints import TLSFingerprint >>> custom = TLSFingerprint(name="custom", ja3="...", user_agent="Custom/1.0") >>> response = client.get("https://example.com", fingerprint=custom)
- class cycletls.AsyncCycleTLS(use_batching=False, batch_size=32, flush_interval=0.0001)[source]¶
Bases:
CycleTLSBackward-compatible async client.
Use
CycleTLSdirectly for new code — it supports both sync and async.- async delete(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a DELETE request.
- Return type:
- async get(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a GET request.
- Return type:
- async head(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a HEAD request.
- Return type:
- async options(url, params=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends an OPTIONS request.
- Return type:
- async patch(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a PATCH request.
- Return type:
- async post(url, params=None, data=None, json_data=None, json=None, fingerprint=None, auth=None, **kwargs)[source]¶
Sends a POST request.
- Return type:
- class cycletls.Session(base_url=None, auth=None, use_batching=False, batch_size=32, flush_interval=0.0001)[source]¶
Bases:
CycleTLSA session object that maintains persistent cookies and headers.
Similar to requests.Session, this class extends CycleTLS to provide automatic cookie and header persistence across requests.
- headers¶
Persistent headers sent with all requests
- Type:
Example
>>> with Session() as session: ... # Set persistent headers ... session.headers['Authorization'] = 'Bearer token123' ... ... # Login - cookies are automatically saved ... session.post('/login', json_data={'user': 'admin', 'pass': 'secret'}) ... ... # Subsequent requests include cookies and headers automatically ... response = session.get('/profile') ... assert 'session_id' in response.cookies
- __init__(base_url=None, auth=None, use_batching=False, batch_size=32, flush_interval=0.0001)[source]¶
Initialize a Session.
- Parameters:
base_url (
Optional[str], default:None) – Base URL prepended to relative URLs (e.g. ‘https://api.example.com’)auth (
Optional[Tuple[str,str]], default:None) – Default (username, password) tuple for HTTP Basic authenticationuse_batching (
bool, default:False) – If True, route individual requests through the batch FFI path for higher throughput.batch_size (
int, default:32) – Maximum number of requests to accumulate per batch.flush_interval (
float, default:0.0001) – Maximum time (seconds) to wait before flushing a partial batch.
- async arequest(method, url, params=None, data=None, json_data=None, json=None, files=None, fingerprint=None, auth=None, poll_interval=0.0, timeout=30.0, use_callback=False, **kwargs)[source]¶
Send an async HTTP request with session persistence.
Async counterpart of
request(). Session cookies, headers, auth, and base_url are all applied the same way.- Parameters:
method (
str) – HTTP methodurl (
str) – Target URL (relative URLs are joined with base_url)params (
Union[Mapping[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]],Sequence[Tuple[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]]],None], default:None) – Query parametersdata (
Union[Dict[str,Any],str,bytes,None], default:None) – Request body datajson_data (
Optional[Dict[str,Any]], default:None) – JSON request bodyjson (
Optional[Dict[str,Any]], default:None) – Alias for json_datafiles (
Optional[Dict[str,Any]], default:None) – File uploadsfingerprint (
Union[str,TLSFingerprint,None], default:None) – TLS fingerprint profileauth (
Optional[Tuple[str,str]], default:None) – (username, password) tuple (overrides session.auth)poll_interval (
float, default:0.0) – Time between polls (default: 0 = adaptive)timeout (
float, default:30.0) – Maximum wait time in seconds (default: 30s)use_callback (
bool, default:False) – Use pipe-based notification instead of polling.**kwargs (
Any) – Additional request options
- Returns:
Response object
- Return type:
- close()[source]¶
Close the session and cleanup resources.
This clears the session cookies and headers, then closes the underlying CycleTLS connection.
- request(method, url, params=None, data=None, json_data=None, json=None, files=None, fingerprint=None, auth=None, **kwargs)[source]¶
Send an HTTP request with session persistence.
Session cookies, headers, auth, and base_url are all applied automatically. Request-specific values take precedence over session values.
- Parameters:
method (
str) – HTTP methodurl (
str) – Target URL (relative URLs are joined with base_url)params (
Union[Mapping[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]],Sequence[Tuple[str,Union[str,int,float,bool,None,Sequence[Union[str,int,float,bool,None]]]]],None], default:None) – Query parametersdata (
Union[Dict[str,Any],str,bytes,None], default:None) – Request body datajson_data (
Optional[Dict[str,Any]], default:None) – JSON request bodyjson (
Optional[Dict[str,Any]], default:None) – Alias for json_datafiles (
Optional[Dict[str,Any]], default:None) – File uploadsfingerprint (
Union[str,TLSFingerprint,None], default:None) – TLS fingerprint profileauth (
Optional[Tuple[str,str]], default:None) – (username, password) tuple (overrides session.auth)**kwargs (
Any) – Additional request options
- Returns:
Response object with status, headers, body, etc.
- Return type:
Module-level functions¶
Sync¶
These functions manage a global session automatically.
- cycletls.request(method, url, **kwargs)[source]¶
Make an HTTP request using the global session with defaults.
- Parameters:
- Return type:
- Returns:
Response object
Example
>>> import cycletls >>> response = cycletls.request('GET', 'https://example.com')
- cycletls.get(url, **kwargs)[source]¶
Make a GET request using the global session with defaults.
- Parameters:
url (
str) – Target URL**kwargs – Additional request options (headers, params, etc.)
- Return type:
- Returns:
Response object
Example
>>> import cycletls >>> response = cycletls.get('https://example.com') >>> print(response.status_code)
- cycletls.post(url, data=None, json_data=None, **kwargs)[source]¶
Make a POST request using the global session with defaults.
- Parameters:
- Return type:
- Returns:
Response object
Example
>>> import cycletls >>> response = cycletls.post('https://api.example.com', json_data={'key': 'value'})
- cycletls.put(url, data=None, json_data=None, **kwargs)[source]¶
Make a PUT request using the global session with defaults.
- cycletls.patch(url, data=None, json_data=None, **kwargs)[source]¶
Make a PATCH request using the global session with defaults.
- cycletls.delete(url, **kwargs)[source]¶
Make a DELETE request using the global session with defaults.
Async¶
- async cycletls.aget(url, **kwargs)[source]¶
Make an async GET request (use ‘aget’ or just ‘get’ with await).
- Parameters:
url (
str) – Target URL**kwargs – Additional request options
- Return type:
- Returns:
Response object
Example
>>> import cycletls >>> response = await cycletls.aget('https://example.com') >>> # Or use the original function with await: >>> response = await cycletls.get(url) # Will auto-detect async context
- async cycletls.apost(url, data=None, json_data=None, **kwargs)[source]¶
Make an async POST request.
- Parameters:
- Return type:
- Returns:
Response object
Example
>>> import cycletls >>> response = await cycletls.apost('https://api.example.com', json_data={'key': 'value'})
- async cycletls.aput(url, data=None, json_data=None, **kwargs)[source]¶
Make an async PUT request.
- Return type:
- async cycletls.apatch(url, data=None, json_data=None, **kwargs)[source]¶
Make an async PATCH request.
- Return type:
- async cycletls.async_request(method, url, **kwargs)[source]¶
Make an async HTTP request.
- Parameters:
- Return type:
- Returns:
Response object
Example
>>> import cycletls >>> response = await cycletls.async_request('GET', 'https://example.com')
- async cycletls.async_get(url, params=None, **kwargs)[source]¶
Send an async GET request (convenience function).
- Return type:
- async cycletls.async_post(url, params=None, data=None, json_data=None, **kwargs)[source]¶
Send an async POST request (convenience function).
- Return type:
Configuration¶
- cycletls.set_default(**kwargs)[source]¶
Set default configuration values for module-level functions.
- Parameters:
ja3 – Default JA3 fingerprint string
ja4r – Default JA4 raw format fingerprint
http2_fingerprint – Default HTTP/2 fingerprint
quic_fingerprint – Default QUIC fingerprint
disable_grease – Default GREASE disable setting
user_agent – Default User-Agent string
proxy – Default proxy URL
timeout – Default timeout in seconds
enable_connection_reuse – Default connection reuse setting
insecure_skip_verify – Default TLS verification setting
server_name – Default SNI override
force_http1 – Default HTTP/1.1 forcing
force_http3 – Default HTTP/3 forcing
protocol – Default protocol selection
disable_redirect – Default redirect handling
header_order – Default header ordering
order_headers_as_provided – Default header order preservation
- Return type:
Example
>>> import cycletls >>> cycletls.set_default( ... proxy='socks5://127.0.0.1:9050', ... timeout=10, ... ja3='771,4865-4866...' ... ) >>> response = cycletls.get('https://example.com') # Uses defaults
- cycletls.get_default(key)[source]¶
Get a default configuration value.
- Parameters:
key (
str) – Configuration key (withoutdefault_prefix)- Return type:
- Returns:
The configuration value, or None if not set
Example
>>> import cycletls >>> cycletls.set_default(timeout=10) >>> cycletls.get_default('timeout') 10
- cycletls.reset_defaults()[source]¶
Reset all default configuration values.
- Return type:
Example
>>> import cycletls >>> cycletls.set_default(proxy='socks5://127.0.0.1:9050') >>> cycletls.reset_defaults() # Clear all defaults
- cycletls.close_global_session()[source]¶
Manually close the global session.
Useful for explicit cleanup in notebooks or when you want to reset the global session with different settings.
Example
>>> import cycletls >>> response = cycletls.get('https://example.com') >>> cycletls.close_global_session() # Explicit cleanup >>> response = cycletls.get('https://example.com') # New session
Response and data structures¶
- class cycletls.schema.Response(request_id, status_code, raw_headers=<factory>, body='', body_bytes=None, raw_cookies=None, final_url=None)[source]¶
Enhanced response model with binary data and cookies.
- property headers: CaseInsensitiveDict¶
Return headers as a case-insensitive dictionary.
- class cycletls.schema.Request(url, method, body='', body_bytes=None, headers=<factory>, ja3=None, ja4r=None, http2_fingerprint=None, quic_fingerprint=None, disable_grease=False, server_name=None, insecure_skip_verify=False, tls13_auto_retry=True, force_http1=False, force_http3=False, protocol=None, user_agent='Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:87.0) Gecko/20100101 Firefox/87.0', proxy='', local_address=None, cookies=None, timeout=6, disable_redirect=False, enable_connection_reuse=True, header_order=None, order_headers_as_provided=None)[source]¶
Enhanced request model with all CycleTLS features.
- class cycletls.schema.Cookie(name, value, path=None, domain=None, expires=None, max_age=None, secure=False, http_only=False, same_site=None)[source]¶
Cookie model with full attributes.
- class cycletls.CaseInsensitiveDict(data=None, **kwargs)[source]¶
A case-insensitive dict-like object.
Implements all methods and operations of MutableMapping as well as dict’s .copy(). All keys are expected to be strings. The structure remembers the case of the last key to be set, and iter(instance), keys(), items(), iterkeys(), and iteritems() will contain case-sensitive keys. However, querying and contains testing is case insensitive:
cid = CaseInsensitiveDict() cid['Accept'] = 'application/json' cid['aCCEPT'] == 'application/json' # True list(cid) == ['Accept'] # True
For example,
headers['content-encoding']will return the value of a'Content-Encoding'response header, regardless of how the header name was originally stored.If the constructor, .update, or equality comparison operations are given keys that have equal
.lower()values, the behavior is undefined.
- class cycletls.CookieJar(cookies=None)[source]¶
A dict-like container for managing cookies.
This class provides a convenient interface for working with Cookie objects, allowing both dict-like access and Cookie object manipulation.
Example:
jar = CookieJar([Cookie(name='session', value='abc123')]) jar['session'] # Returns 'abc123' jar.set('user', 'john', domain='.example.com') jar.get_dict() # Returns {'session': 'abc123', 'user': 'john'}
- get_dict()[source]¶
Return a simple dict of cookie names to values.
- Return type:
- Returns:
A dictionary mapping cookie names to their values.
Fingerprints¶
- class cycletls.TLSFingerprint(name, ja3, ja4r=None, http2_fingerprint=None, quic_fingerprint=None, user_agent=None, header_order=None, headers=None, disable_grease=False, force_http1=False, force_http3=False)[source]¶
A reusable TLS fingerprint configuration.
- class cycletls.FingerprintRegistry[source]¶
Registry for managing fingerprint profiles.
- classmethod by_family(family, platform=None)[source]¶
Return all profiles whose name starts with any prefix for family.
If platform is given, only profiles whose name ends with
_<platform.value>are returned. Profiles without a platform suffix are excluded when platform is specified.- Return type:
- classmethod latest(family=None, platform=None)[source]¶
Return the profile with the highest version number.
family and platform are optional filters; omit both to search the entire registry.
- Return type:
- class cycletls.BrowserFamily(*values)[source]¶
Browser family for fingerprint lookup.
- BRAVE = 'brave'¶
- CHROME = 'chrome'¶
- CHROMIUM = 'chromium'¶
- EDGE = 'edge'¶
- FIREFOX = 'firefox'¶
- OPERA = 'opera'¶
- SAFARI = 'safari'¶
- SAMSUNG = 'samsung'¶
- class cycletls.Platform(*values)[source]¶
Platform / operating system for fingerprint lookup.
- ANDROID = 'android'¶
- IOS = 'ios'¶
- LINUX = 'linux'¶
- MACOS = 'mac'¶
- WINDOWS = 'win'¶
Loading profiles¶
- cycletls.load_fingerprint_from_file(path)[source]¶
Load a single fingerprint from a JSON file and register it.
- Parameters:
- Return type:
- Returns:
The loaded TLSFingerprint
- Raises:
FileNotFoundError – If file doesn’t exist
json.JSONDecodeError – If file is not valid JSON
KeyError – If required fields are missing
- cycletls.load_fingerprints_from_dir(directory)[source]¶
Load all JSON fingerprint files from a directory.
Scans the given directory for .json files and attempts to load each as a TLSFingerprint. Invalid files are logged and skipped.
- Parameters:
directory (
str|Path) – Path to directory containing fingerprint JSON files- Return type:
- Returns:
Number of fingerprints successfully loaded
Example
>>> count = load_fingerprints_from_dir("/path/to/fingerprints") >>> print(f"Loaded {count} fingerprint profiles")
- cycletls.load_fingerprints_from_env()[source]¶
Load fingerprints from directory specified in CYCLETLS_FINGERPRINTS_DIR.
If the environment variable is set, loads all .json files from that directory. If the variable is not set, does nothing.
- Return type:
- Returns:
Number of fingerprints loaded, or 0 if env var not set
Example
>>> import os >>> os.environ["CYCLETLS_FINGERPRINTS_DIR"] = "/my/fingerprints" >>> count = load_fingerprints_from_env()
WebSocket¶
- class cycletls.WebSocketConnection(url, headers=None, ja3=None, user_agent=None, proxy=None, timeout=30)[source]¶
A WebSocket connection with TLS fingerprinting support.
This class provides both synchronous and asynchronous APIs for WebSocket communication. It supports context managers for automatic cleanup.
- Parameters:
url (
str) – WebSocket URL (ws:// or wss://)headers (
Optional[Dict[str,str]], default:None) – Optional headers to include in the handshakeja3 (
Optional[str], default:None) – Optional JA3 fingerprint for TLSuser_agent (
Optional[str], default:None) – Optional User-Agent headertimeout (
int, default:30) – Connection timeout in seconds
Example
>>> # Synchronous usage >>> ws = WebSocketConnection("wss://echo.websocket.org") >>> ws.connect() >>> ws.send("Hello") >>> msg = ws.receive() >>> ws.close() >>> >>> # Context manager >>> with WebSocketConnection("wss://echo.websocket.org") as ws: ... ws.send("Hello") ... msg = ws.receive() >>> >>> # Async context manager >>> async with WebSocketConnection("wss://stream.example.com") as ws: ... await ws.send_async("Hello") ... msg = await ws.receive_async()
- connect()[source]¶
Establish the WebSocket connection.
- Raises:
WebSocketError – If connection fails
- Return type:
- receive(timeout=None)[source]¶
Receive a message from the WebSocket.
- Parameters:
timeout (
Optional[float], default:None) – Optional timeout in seconds (uses connection timeout if not specified)- Return type:
- Returns:
WebSocketMessage with the received data
- Raises:
WebSocketError – If not connected or receive fails
- send(data, binary=False)[source]¶
Send a message over the WebSocket.
- Parameters:
- Raises:
WebSocketError – If not connected or send fails
- Return type:
Server-Sent Events¶
- class cycletls.SSEConnection(url, headers=None, ja3=None, user_agent=None, proxy=None, timeout=30, last_event_id=None)[source]¶
A Server-Sent Events connection with TLS fingerprinting support.
This class provides both synchronous and asynchronous APIs for SSE streams. It supports automatic reconnection and Last-Event-ID tracking.
- Parameters:
url (
str) – SSE endpoint URLheaders (
Optional[Dict[str,str]], default:None) – Optional headers to include in the requestja3 (
Optional[str], default:None) – Optional JA3 fingerprint for TLSuser_agent (
Optional[str], default:None) – Optional User-Agent headertimeout (
int, default:30) – Connection timeout in secondslast_event_id (
Optional[str], default:None) – Optional Last-Event-ID for reconnection
Example
>>> # Synchronous iteration >>> with SSEConnection("https://api.example.com/events") as sse: ... for event in sse: ... if event.event == "update": ... print(event.data) >>> >>> # Async iteration >>> async with SSEConnection("https://api.example.com/events") as sse: ... async for event in sse: ... print(f"[{event.event}] {event.data}")
- __init__(url, headers=None, ja3=None, user_agent=None, proxy=None, timeout=30, last_event_id=None)[source]¶
Exceptions¶
All exceptions inherit from CycleTLSError.
- exception cycletls.CycleTLSError[source]¶
Base exception for all CycleTLS errors.
All CycleTLS-specific exceptions inherit from this class, making it easy to catch any CycleTLS-related error.
- exception cycletls.RequestException[source]¶
Base class for all request-related exceptions.
This is the base exception for errors that occur during the request/response cycle.
- exception cycletls.HTTPError(message, response=None)[source]¶
Bases:
RequestExceptionAn HTTP error occurred (4xx or 5xx status code).
This exception is raised when raise_for_status() is called on a Response with a 4xx or 5xx status code.
- response¶
The Response object (if available)
- exception cycletls.ConnectionError[source]¶
Bases:
RequestExceptionA connection error occurred.
Raised when a request fails due to network connectivity issues, such as DNS lookup failures or connection refused errors.
- exception cycletls.Timeout[source]¶
Bases:
RequestExceptionThe request timed out.
Raised when a request exceeds the specified timeout duration. This includes both connection timeouts and read timeouts.
- exception cycletls.TooManyRedirects[source]¶
Bases:
RequestExceptionToo many redirects occurred.
Raised when the maximum number of redirects is exceeded.
- exception cycletls.InvalidURL[source]¶
Bases:
RequestExceptionThe URL provided was invalid.
Raised when attempting to make a request with a malformed or invalid URL.
- exception cycletls.TLSError[source]¶
Bases:
RequestExceptionA TLS/SSL error occurred.
Raised when there are issues with TLS/SSL certificates, protocol version mismatches, or other TLS-related errors.
- exception cycletls.ProxyError[source]¶
Bases:
RequestExceptionA proxy error occurred.
Raised when there are issues connecting to or communicating through a proxy server.
- exception cycletls.InvalidHeader[source]¶
Bases:
RequestExceptionInvalid header name or value.
Raised when attempting to use an invalid HTTP header name or value.
- exception cycletls.WebSocketError[source]¶
Bases:
CycleTLSErrorRaised when a WebSocket operation fails.
- exception cycletls.SSEError[source]¶
Bases:
CycleTLSErrorRaised when an SSE operation fails.