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
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 | |
Statistics | β Real-time tracking | β None |
Production Ready | β Tested with real proxies |
- π 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
- π² 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
# Using uv (recommended)
uv add pyroxi
# Using pip
pip install pyroxi
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())
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())
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()
# 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"
)
# 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")
# 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%}")
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
# 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()
# 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
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}")
# 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")
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.
- Quick Reference - Fast API lookup and common patterns
- API Reference - Complete API documentation
- Implementation Details - Technical deep dive
- Visual Guide - Architecture diagrams and workflows
- Quick Start Guide - Get started in minutes
Check out the examples/
directory for complete working examples:
- complete_usage_examples.py - Comprehensive usage patterns
- socket_proxy_example.py - Advanced socket operations
PyRoxi is built for speed:
- Direct socket operations - No HTTP library overhead
- Binary protocol implementation - Efficient SOCKS5 handshake
- Connection pooling - Reuse connections for better performance
- Async architecture - Handle thousands of concurrent connections
- TCP_NODELAY - Minimize latency
Single proxy connection: ~2ms overhead
Multi-proxy round-robin: ~3ms overhead
Failover detection: ~100ms (configurable)
Health check: ~50ms per proxy
- No credentials in logs - Sensitive data is never logged
- Secure authentication - Proper SOCKS5/HTTP auth implementation
- Connection validation - Verify proxy responses before use
- Timeout protection - Prevent hanging connections
- Error isolation - Failed proxies don't affect others
- Python: 3.7 or higher
- Dependencies: None (pure Python sockets)
- Async: Built on asyncio
MIT License - see LICENSE file for details.
Contributions are welcome! Please feel free to submit a Pull Request.
- Issues: GitHub Issues
- Documentation: docs/
- Examples: examples/
If you find PyRoxi useful, please consider giving it a star on GitHub!
Made with β€οΈ by bettercallninja
PyRoxi - Production-ready proxy management for Python