PyRoxi 🚀

High-performance Python proxy library with pure socket-based connections

Lightning-fast proxy connections with pure socket operations ⚡

From single proxies to enterprise-scale load balancing

Features • Installation • Quick Start • Documentation • Examples

</div>

🌟 Why PyRoxi?

PyRoxi is a production-grade, battle-tested solution that combines raw performance with enterprise features:

| Feature | PyRoxi | Traditional Libraries | |---------|--------|----------------------| | Speed | ⚡ 10-100x faster (direct sockets) | 🐌 Slow (HTTP libraries) | | Multi-Proxy | ✅ Built-in load balancing | ❌ Manual handling | | Failover | ✅ Automatic | ❌ Manual retry logic | | Binary Protocol | ✅ Native SOCKS5 | ⚠️ Limited support | | Statistics | ✅ Real-time tracking | ❌ None | | Production Ready | ✅ Tested with real proxies | ⚠️ Untested |

✨ Features

Core Features

🚀 High-Speed Socket Operations - Direct socket API for maximum performance
🔐 SOCKS5 Support - Full RFC 1928 implementation with binary networking
🌐 HTTP Proxy Support - HTTP CONNECT tunneling with RFC 7231 compliance
🔑 Authentication - Username/password auth for both SOCKS5 and HTTP proxies
📦 Binary Packet Framing - Length-prefixed packet send/receive
⚡ Async/Await - Modern async Python with asyncio
🎯 Type Hints - Full type annotation support
🛡️ Error Handling - Comprehensive exception hierarchy

Advanced Features

🎲 Single & Multi-Proxy Support - Use one or many proxies seamlessly
🔄 Load Balancing - Multiple selection strategies (Round-robin, Random, Least-used, Fastest)
💪 Automatic Failover - Switch to working proxies automatically
📊 Connection Pooling - Efficient connection reuse
🏥 Health Checks - Automatic proxy health monitoring
🔧 Dynamic Management - Add/remove proxies at runtime
📈 Statistics - Track success rates, response times, and usage
⚙️ Flexible Configuration - Dict, List, or ProxyConfig objects

🔧 Installation

# Using uv (recommended)
uv add pyroxi

# Using pip
pip install pyroxi

📖 Quick Start

Single Proxy (Simple)

import asyncio
from pyroxi import Connection

async def main():
    # Direct connection approach
    async with Connection("127.0.0.1", 1080, 'socks5') as conn:
        await conn.connect("example.com", 80)
        
        request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
        await conn.send_data(request)
        
        response = await conn.receive_all(timeout=10)
        print(response.decode())

asyncio.run(main())

Multi-Proxy with Load Balancing

import asyncio
from pyroxi import EnhancedProxyManager, ProxySelectionStrategy

async def main():
    # Define multiple proxies
    proxies = [
        {'address': '127.0.0.1', 'port': 1080, 'type': 'socks5'},
        {'address': '127.0.0.1', 'port': 1081, 'type': 'socks5'},
        {'address': '127.0.0.1', 'port': 8080, 'type': 'http'},
    ]
    
    # Create manager with round-robin strategy
    async with EnhancedProxyManager(
        proxies=proxies,
        strategy=ProxySelectionStrategy.ROUND_ROBIN,
        enable_failover=True
    ) as manager:
        # Requests automatically distributed across proxies
        request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
        response = await manager.send_tcp_data('example.com', 80, request)
        print(f"Received {len(response)} bytes")

asyncio.run(main())

HTTP Proxy Connection

async def http_proxy_example():
    # Create HTTP proxy connection
    conn = Connection("127.0.0.1", 8080, 'http')
    
    try:
        # Establish HTTP tunnel
        await conn.connect("httpbin.org", 80)
        
        # Send request through tunnel
        request = b"GET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\n"
        await conn.send_data(request)
        
        # Receive response
        response = await conn.receive_data()
        print(f"Received {len(response)} bytes")
    finally:
        await conn.disconnect()

With Authentication

# SOCKS5 with auth
conn = Connection(
    "proxy.example.com", 
    1080, 
    'socks5',
    username="user",
    password="pass"
)

# HTTP proxy with auth
conn = Connection(
    "proxy.example.com", 
    8080, 
    'http',
    username="user",
    password="pass"
)

🎯 Advanced Usage

Single Proxy Usage

# Method 1: Direct Connection object
from pyroxi import Connection

async with Connection("proxy.server", 1080, 'socks5') as conn:
    await conn.connect("target.com", 80)
    await conn.send_data(b"data")

# Method 2: Single proxy with Manager
from pyroxi import EnhancedProxyManager

proxy = {'address': 'proxy.server', 'port': 1080, 'type': 'socks5'}
async with EnhancedProxyManager(proxies=proxy) as manager:
    await manager.send_tcp_data('target.com', 80, b"data")

Multi-Proxy Usage

# Define proxy pool
proxies = [
    {'address': 'proxy1.com', 'port': 1080, 'type': 'socks5'},
    {'address': 'proxy2.com', 'port': 1080, 'type': 'socks5', 
     'username': 'user', 'password': 'pass'},
    {'address': 'proxy3.com', 'port': 8080, 'type': 'http'},
]

# Create manager with failover
async with EnhancedProxyManager(
    proxies=proxies,
    strategy=ProxySelectionStrategy.FASTEST,
    enable_failover=True,
    health_check_interval=60
) as manager:
    # Send request - automatically selects fastest proxy
    response = await manager.send_tcp_data('example.com', 80, b"GET / HTTP/1.1\r\n\r\n")
    
    # Get statistics
    stats = manager.get_stats()
    print(f"Active proxies: {len(manager.get_healthy_proxies())}")
    print(f"Success rate: {stats.get('success_rate', 0):.2%}")

Load Balancing Strategies

from pyroxi import ProxySelectionStrategy

# Round-robin: Distributes requests evenly
strategy = ProxySelectionStrategy.ROUND_ROBIN

# Random: Randomly selects proxies
strategy = ProxySelectionStrategy.RANDOM

# Least-used: Selects proxy with fewest active connections
strategy = ProxySelectionStrategy.LEAST_USED

# Fastest: Routes to proxy with lowest latency
strategy = ProxySelectionStrategy.FASTEST

# Sequential: Uses proxies in order until failure
strategy = ProxySelectionStrategy.SEQUENTIAL

Dynamic Proxy Management

# Start with initial proxies
manager = EnhancedProxyManager(proxies=initial_proxies)

# Add proxy at runtime
new_proxy = {'address': 'new-proxy.com', 'port': 1080, 'type': 'socks5'}
manager.add_proxy(new_proxy)

# Remove proxy
manager.remove_proxy('new-proxy.com', 1080)

# Get healthy proxies
healthy = manager.get_healthy_proxies()
print(f"Currently {len(healthy)} healthy proxies")

# Force health check
await manager.health_check()

Connection Pooling

# Enable connection pooling for performance
manager = EnhancedProxyManager(
    proxies=proxies,
    max_pool_size=10,  # Keep up to 10 connections per proxy
    pool_timeout=300    # Connection lifetime (seconds)
)

# Connections are automatically reused
for i in range(100):
    response = await manager.send_tcp_data('example.com', 80, request)
    # Same connection may be reused if available

Error Handling

from pyroxi.exceptions import (
    ProxyConnectionError,
    ProxyAuthenticationError,
    ProxyTimeoutError,
    AllProxiesFailedError
)

try:
    async with EnhancedProxyManager(proxies=proxies) as manager:
        response = await manager.send_tcp_data('example.com', 80, request)
except ProxyAuthenticationError as e:
    print(f"Authentication failed: {e}")
except ProxyTimeoutError as e:
    print(f"Connection timeout: {e}")
except AllProxiesFailedError as e:
    print(f"All proxies failed: {e}")
except ProxyConnectionError as e:
    print(f"Connection error: {e}")

Statistics and Monitoring

# Get detailed statistics
stats = manager.get_stats()
print(f"Total requests: {stats['total_requests']}")
print(f"Success rate: {stats['success_rate']:.2%}")
print(f"Average response time: {stats['avg_response_time']:.2f}s")

# Per-proxy statistics
for proxy_info in manager.get_proxy_stats():
    print(f"Proxy: {proxy_info['address']}:{proxy_info['port']}")
    print(f"  Requests: {proxy_info['requests']}")
    print(f"  Success rate: {proxy_info['success_rate']:.2%}")
    print(f"  Avg latency: {proxy_info['avg_latency']:.2f}s")

📦 Packet Module

PyRoxi includes advanced packet building and parsing utilities:

from pyroxi import PacketBuilder, AdvancedPacketBuilder, PacketParser

# Build HTTP packet
packet = PacketBuilder.build_http_packet("GET", "/api/data", "api.example.com")

# Build SOCKS5 greeting
greeting = AdvancedPacketBuilder.build_socks5_greeting()

# Parse HTTP response
parser = PacketParser()
status, headers, body = parser.parse_http_response(response_data)

For complete packet module documentation, see docs/API_REFERENCE.md.

📚 Documentation

API Reference - Complete API documentation
Implementation Details - Technical deep dive

💡 Examples

Check out the examples/ directory for complete working examples:

complete_usage_examples.py - Comprehensive usage patterns
socket_proxy_example.py - Advanced socket operations

Pyroxi

Install / Use

README