CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-python-socks

Proxy (SOCKS4, SOCKS5, HTTP CONNECT) client for Python

—

Pending
Overview
Eval results
Files

async-api.mddocs/

Asynchronous API

Non-blocking proxy implementations for asyncio, trio, curio, and anyio frameworks. Each framework has its own optimized implementation while maintaining a consistent API interface. The async implementations return sockets configured for non-blocking operation, suitable for integration with async I/O frameworks.

Framework-Specific Imports

# Asyncio
from python_socks.async_.asyncio import Proxy as AsyncioProxy

# Trio  
from python_socks.async_.trio import Proxy as TrioProxy

# Curio
from python_socks.async_.curio import Proxy as CurioProxy

# AnyIO (with proxy chaining)
from python_socks.async_.anyio import Proxy as AnyioProxy, ProxyChain as AnyioProxyChain

Capabilities

Asyncio Proxy

Asyncio-specific proxy implementation optimized for asyncio event loops.

import asyncio
import socket
from typing import Optional

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: ...
    
    @property
    def proxy_host(self) -> str:
        """Get proxy host address."""
        ...
    
    @property
    def proxy_port(self) -> int:
        """Get proxy port number."""
        ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'AsyncioProxy':
        """Create proxy instance (deprecated, use __init__ directly)."""
        ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'AsyncioProxy':
        """Create proxy instance from URL."""
        ...

Asyncio-Specific Parameters:

  • loop: Optional event loop (defaults to current loop)

Usage:

import asyncio
import ssl
from python_socks.async_.asyncio import Proxy

async def main():
    # Create proxy
    proxy = Proxy.from_url('socks5://user:pass@127.0.0.1:1080')
    
    # Connect through proxy
    sock = await proxy.connect('httpbin.org', 443, timeout=30)
    
    # Use with asyncio high-level API
    reader, writer = await asyncio.open_connection(
        host=None,
        port=None,
        sock=sock,
        ssl=ssl.create_default_context(),
        server_hostname='httpbin.org'
    )
    
    # Make request
    writer.write(b'GET /ip HTTP/1.1\r\nHost: httpbin.org\r\n\r\n')
    await writer.drain()
    
    response = await reader.read(4096)
    print(response.decode())
    
    writer.close()
    await writer.wait_closed()

asyncio.run(main())

Trio Proxy

Trio-specific proxy implementation for structured concurrency.

import trio
from typing import Optional

class TrioProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None
    ): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None,
        **kwargs
    ) -> trio.socket.SocketType: ...
    
    @property
    def proxy_host(self) -> str:
        """Get proxy host address."""
        ...
    
    @property
    def proxy_port(self) -> int:
        """Get proxy port number."""
        ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'TrioProxy':
        """Create proxy instance (deprecated, use __init__ directly)."""
        ...
    
    @classmethod  
    def from_url(cls, url: str, **kwargs) -> 'TrioProxy':
        """Create proxy instance from URL."""
        ...

Usage:

import trio
import ssl
from python_socks.async_.trio import Proxy

async def main():
    # Create proxy
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Connect through proxy  
    sock = await proxy.connect('example.com', 443)
    
    # Use with trio's SSL wrapper
    ssl_context = ssl.create_default_context()
    stream = trio.SSLStream(
        trio.SocketStream(sock),
        ssl_context,
        server_hostname='example.com'
    )
    
    # Make request
    await stream.send_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
    response = await stream.receive_some(4096)
    print(response.decode())
    
    await stream.aclose()

trio.run(main)

Curio Proxy

Curio-specific proxy implementation for async/await concurrency.

import curio.io
from typing import Optional

class CurioProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None
    ): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None,
        **kwargs
    ) -> curio.io.Socket: ...
    
    @property
    def proxy_host(self) -> str:
        """Get proxy host address."""
        ...
    
    @property
    def proxy_port(self) -> int:
        """Get proxy port number."""
        ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'CurioProxy':
        """Create proxy instance (deprecated, use __init__ directly)."""
        ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'CurioProxy':
        """Create proxy instance from URL."""
        ...

Usage:

import curio
import ssl
from python_socks.async_.curio import Proxy

async def main():
    # Create proxy
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Connect through proxy
    sock = await proxy.connect('example.com', 80)
    
    # Use with curio
    stream = sock.as_stream()
    
    await stream.write(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
    response = await stream.read(4096)
    print(response.decode())
    
    await stream.close()

curio.run(main)

AnyIO Proxy

AnyIO-compatible proxy implementation that works across multiple async frameworks.

import ssl
from typing import Optional

class AnyioProxy:
    def __init__(
        self,
        proxy_type: ProxyType,
        host: str,
        port: int,
        username: Optional[str] = None,
        password: Optional[str] = None,
        rdns: Optional[bool] = None,
        proxy_ssl: Optional[ssl.SSLContext] = None
    ): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        dest_ssl: Optional[ssl.SSLContext] = None,
        timeout: Optional[float] = None,
        **kwargs
    ) -> 'AnyioSocketStream': ...
    
    @property
    def proxy_host(self) -> str:
        """Get proxy host address."""
        ...
    
    @property
    def proxy_port(self) -> int:
        """Get proxy port number."""
        ...
    
    @classmethod
    def create(cls, *args, **kwargs) -> 'AnyioProxy':
        """Create proxy instance (deprecated, use __init__ directly)."""
        ...
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'AnyioProxy':
        """Create proxy instance from URL."""
        ...

Usage:

import anyio
import ssl
from python_socks.async_.anyio import Proxy

async def main():
    # Create proxy
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Connect through proxy
    sock = await proxy.connect('httpbin.org', 443)
    
    # Use with anyio
    ssl_context = ssl.create_default_context()
    stream = await anyio.connect_tcp(
        remote_host='httpbin.org',
        remote_port=443,
        sock=sock,
        tls=True,
        tls_hostname='httpbin.org'
    )
    
    await stream.send(b'GET /ip HTTP/1.1\r\nHost: httpbin.org\r\n\r\n')
    response = await stream.receive(4096)
    print(response.decode())
    
    await stream.aclose()

# Works with asyncio
anyio.run(main, backend='asyncio')

# Also works with trio  
anyio.run(main, backend='trio')

AnyIO Proxy Chain

Asynchronous proxy chaining with AnyIO compatibility.

class ProxyChain:
    def __init__(self, proxies: Iterable[AnyioProxy]): ...
    
    async def connect(
        self,
        dest_host: str,
        dest_port: int,
        timeout: Optional[float] = None
    ) -> socket.socket: ...

Note: This is the only non-deprecated proxy chain implementation in the async API.

Usage:

import anyio
from python_socks.async_.anyio import Proxy, ProxyChain

async def main():
    # Create proxy chain
    proxy1 = Proxy.from_url('socks5://first-proxy:1080')
    proxy2 = Proxy.from_url('http://second-proxy:8080')
    
    chain = ProxyChain([proxy1, proxy2])
    
    # Connect through chain
    sock = await chain.connect('example.com', 80)
    
    # Use the socket
    stream = anyio.SocketStream(sock)
    await stream.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
    response = await stream.receive(4096)
    print(response.decode())
    
    await stream.aclose()

anyio.run(main, backend='asyncio')

Advanced Usage Examples

Concurrent Connections

import asyncio
from python_socks.async_.asyncio import Proxy

async def fetch_url(proxy, url_host, url_path):
    """Fetch URL through proxy."""
    try:
        sock = await proxy.connect(url_host, 80, timeout=10)
        
        reader, writer = await asyncio.open_connection(
            host=None, port=None, sock=sock
        )
        
        request = f'GET {url_path} HTTP/1.1\r\nHost: {url_host}\r\n\r\n'
        writer.write(request.encode())
        await writer.drain()
        
        response = await reader.read(4096)
        writer.close()
        await writer.wait_closed()
        
        return response.decode()
        
    except Exception as e:
        return f"Error: {e}"

async def main():
    proxy = Proxy.from_url('socks5://proxy.example.com:1080')
    
    # Concurrent requests
    tasks = [
        fetch_url(proxy, 'httpbin.org', '/ip'),
        fetch_url(proxy, 'httpbin.org', '/user-agent'),
        fetch_url(proxy, 'httpbin.org', '/headers')
    ]
    
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        print(f"Request {i + 1}: {result[:100]}...")

asyncio.run(main())

Timeout and Error Handling

import asyncio
from python_socks import ProxyError, ProxyTimeoutError, ProxyConnectionError
from python_socks.async_.asyncio import Proxy

async def robust_connect(proxy_url, dest_host, dest_port, max_retries=3):
    """Async connection with retry logic."""
    
    for attempt in range(max_retries):
        try:
            proxy = Proxy.from_url(proxy_url)
            return await proxy.connect(dest_host, dest_port, timeout=10)
            
        except ProxyConnectionError as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise
            
        except ProxyTimeoutError:
            if attempt < max_retries - 1:
                continue
            raise
            
        except ProxyError as e:
            # Don't retry authentication errors
            raise

async def main():
    try:
        sock = await robust_connect(
            'socks5://proxy.example.com:1080',
            'api.example.com',
            443
        )
        print("Connected successfully!")
        sock.close()
    except Exception as e:
        print(f"Connection failed: {e}")

asyncio.run(main())

Framework Detection Pattern

import sys

# Detect available async framework
if 'trio' in sys.modules:
    from python_socks.async_.trio import Proxy
    framework = 'trio'
elif 'curio' in sys.modules:
    from python_socks.async_.curio import Proxy  
    framework = 'curio'
else:
    from python_socks.async_.asyncio import Proxy
    framework = 'asyncio'

print(f"Using {framework} proxy implementation")

Cross-Framework Compatibility with AnyIO

import anyio
from python_socks.async_.anyio import Proxy

async def universal_proxy_request(proxy_url, dest_host, dest_port, request_data):
    """Request that works with any anyio-supported backend."""
    
    proxy = Proxy.from_url(proxy_url)
    sock = await proxy.connect(dest_host, dest_port)
    
    stream = anyio.SocketStream(sock)
    
    try:
        await stream.send(request_data)
        response = await stream.receive(4096)
        return response
    finally:
        await stream.aclose()

# Works with asyncio
async def asyncio_main():
    result = await universal_proxy_request(
        'socks5://proxy:1080',
        'example.com', 
        80,
        b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n'
    )
    print("Asyncio result:", result.decode()[:100])

# Works with trio
async def trio_main():
    result = await universal_proxy_request(
        'socks5://proxy:1080',
        'example.com',
        80, 
        b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n'
    )
    print("Trio result:", result.decode()[:100])

# Run with different backends
anyio.run(asyncio_main, backend='asyncio')
anyio.run(trio_main, backend='trio')

Framework Integration Notes

Asyncio Integration

  • Returns sockets in non-blocking mode
  • Compatible with asyncio.open_connection()
  • Works with asyncio SSL contexts
  • Integrates with asyncio timeout handling

Trio Integration

  • Optimized for trio's structured concurrency model
  • Compatible with trio.SocketStream
  • Works with trio's SSL wrapper
  • Follows trio's cancellation semantics

Curio Integration

  • Designed for curio's async/await model
  • Compatible with curio stream abstractions
  • Integrates with curio's timeout mechanisms
  • Follows curio's task management patterns

AnyIO Integration

  • Backend-agnostic implementation
  • Works across asyncio, trio, and curio
  • Provides consistent API regardless of backend
  • Enables framework-portable applications

Install with Tessl CLI

npx tessl i tessl/pypi-python-socks

docs

async-api.md

core-api.md

index.md

sync-api.md

v2-api.md

tile.json