Proxy (SOCKS4, SOCKS5, HTTP CONNECT) client for Python
npx @tessl/cli install tessl/pypi-python-socks@2.7.0Python SOCKS provides a core proxy client functionality for Python. It supports SOCKS4(a), SOCKS5(h), and HTTP CONNECT proxy protocols with both synchronous and asynchronous APIs. The library offers native integration with multiple async frameworks including asyncio, trio, curio, and anyio, enabling developers to route network connections through proxy servers across different concurrency models.
pip install python-sockspip install python-socks[asyncio], pip install python-socks[trio], pip install python-socks[curio], pip install python-socks[anyio]from python_socks import ProxyType, ProxyError, ProxyTimeoutError, ProxyConnectionError, parse_proxy_urlSynchronous usage:
from python_socks.sync import Proxy, ProxyChainAsynchronous usage (by framework):
from python_socks.async_.asyncio import Proxy as AsyncioProxy
from python_socks.async_.trio import Proxy as TrioProxy
from python_socks.async_.curio import Proxy as CurioProxy
from python_socks.async_.anyio import Proxy as AnyioProxy, ProxyChain as AnyioProxyChainV2 enhanced APIs:
from python_socks.sync.v2 import Proxy as SyncProxyV2, ProxyChain as SyncProxyChainV2
from python_socks.async_.asyncio.v2 import Proxy as AsyncioProxyV2, ProxyChain as AsyncioProxyChainV2import ssl
from python_socks.sync import Proxy
# Create proxy from URL
proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# Connect through proxy - returns standard Python socket
sock = proxy.connect(dest_host='check-host.net', dest_port=443)
# Use the socket normally
sock = ssl.create_default_context().wrap_socket(
sock=sock,
server_hostname='check-host.net'
)
request = (
b'GET /ip HTTP/1.1\r\n'
b'Host: check-host.net\r\n'
b'Connection: close\r\n\r\n'
)
sock.sendall(request)
response = sock.recv(4096)
print(response)import ssl
import asyncio
from python_socks.async_.asyncio import Proxy
async def main():
# Create proxy from URL
proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# Connect through proxy - returns non-blocking socket
sock = await proxy.connect(dest_host='check-host.net', dest_port=443)
# Use with asyncio
reader, writer = await asyncio.open_connection(
host=None,
port=None,
sock=sock,
ssl=ssl.create_default_context(),
server_hostname='check-host.net',
)
request = (
b'GET /ip HTTP/1.1\r\n'
b'Host: check-host.net\r\n'
b'Connection: close\r\n\r\n'
)
writer.write(request)
await writer.drain()
response = await reader.read(4096)
print(response)
writer.close()
await writer.wait_closed()
asyncio.run(main())The library is organized around proxy implementations for different concurrency models:
This modular design allows the library to serve as a foundational dependency for higher-level HTTP client libraries while providing direct access to proxy functionality across different concurrency paradigms.
Common types, exceptions, and utility functions used across all proxy implementations. Includes proxy type enumeration, error handling classes, and URL parsing functionality.
from typing import Tuple, Optional
from enum import Enum
class ProxyType(Enum):
SOCKS4 = 1
SOCKS5 = 2
HTTP = 3
class ProxyError(Exception): ...
class ProxyTimeoutError(TimeoutError): ...
class ProxyConnectionError(OSError): ...
def parse_proxy_url(url: str) -> Tuple[ProxyType, str, int, Optional[str], Optional[str]]: ...Blocking proxy implementations that return standard Python sockets. Supports SOCKS4, SOCKS5, and HTTP CONNECT protocols with authentication and proxy chaining capabilities.
class Proxy:
def __init__(self, proxy_type: ProxyType, host: str, port: int,
username: Optional[str] = None, password: Optional[str] = None,
rdns: Optional[bool] = None): ...
def connect(self, dest_host: str, dest_port: int,
timeout: Optional[float] = None, **kwargs) -> socket.socket: ...
@classmethod
def from_url(cls, url: str, **kwargs) -> 'Proxy': ...Non-blocking proxy implementations for asyncio, trio, curio, and anyio frameworks. Each framework has its own optimized implementation while maintaining a consistent API interface.
class AsyncioProxy:
def __init__(self, proxy_type: ProxyType, host: str, port: int,
username: Optional[str] = None, password: Optional[str] = None,
rdns: Optional[bool] = None, loop: Optional[asyncio.AbstractEventLoop] = None): ...
async def connect(self, dest_host: str, dest_port: int,
timeout: Optional[float] = None, **kwargs) -> socket.socket: ...
@classmethod
def from_url(cls, url: str, **kwargs) -> 'AsyncioProxy': ...Improved proxy implementations with enhanced error handling, SSL support, and additional configuration options. Available for both synchronous and asynchronous usage patterns.
class SyncProxyV2:
def __init__(self, proxy_type: ProxyType, host: str, port: int,
username: Optional[str] = None, password: Optional[str] = None,
rdns: Optional[bool] = None): ...
def connect(self, dest_host: str, dest_port: int,
timeout: Optional[float] = None, **kwargs) -> socket.socket: ...