ProxyBroker2 lets you add your own proxy sources without modifying the codebase. The most common use case — a Docker user dropping YAML files into a folder — works out of the box.
The recommended UX for the published Docker image. Drop YAML/JSON config files into a folder, bind-mount it to /configs, and ProxyBroker picks them up automatically.
# ~/proxy-sources/my_secret_list.yaml
name: My Internal Source
type: simple
url: https://my-private-server.example.com/proxies.txt
format: text # auto, text, json, csv, html
protocols:
- HTTP
- HTTPS
max_connections: 4
timeout: 20docker run --rm \
-v ~/proxy-sources:/configs \
bluet/proxybroker2 \
find --types HTTP --limit 10That is it. Multiple *.yaml / *.yml / *.json files in the directory are all loaded. Files starting with _ are ignored.
The CLI looks for a provider directory in this order — the first match wins:
- Each
--provider-dir PATH(repeatable) on the command line. $PROXYBROKER_PROVIDER_DIRenvironment variable (single path)./configsif it exists in the container (the Docker convention shown above).
If none of those are set, only the bundled provider list is used (existing behaviour).
By default, configs from your directory are added to the bundled provider list. To use only your own configs and skip every bundled source, you currently need the Python API:
from proxybroker import Broker
# Only providers from /configs, no bundled defaults.
broker = Broker(providers=[], provider_dirs=['/configs'])There is no CLI equivalent today — --provider URL always adds a single URL, it cannot express the empty list. If you need a no-defaults Docker workflow, run a tiny Python wrapper instead of python -m proxybroker. Tracking issue: open one if you want a --no-default-providers CLI flag.
If you are scripting against the library directly, you can also pass Provider instances or use the helper classes.
from proxybroker import SimpleProvider, Broker
class MyProvider(SimpleProvider):
domain = "mysite.com"
def __init__(self):
super().__init__(
url="http://mysite.com/proxies.txt",
format='text',
proto=('HTTP', 'HTTPS')
)
broker = Broker(providers=[MyProvider()])load_python_providers_from_directory() is also available, but it executes arbitrary Python code from every .py file in the directory. Only point it at a directory whose contents you fully control. The CLI never calls it; you must opt in from Python:
from proxybroker import Broker, load_python_providers_from_directory
trusted_providers = load_python_providers_from_directory('/path/I/trust')
broker = Broker(providers=trusted_providers)Perfect for sites that provide proxies in common formats:
from proxybroker import SimpleProvider
class TextListProvider(SimpleProvider):
"""Provider for plain text proxy lists."""
domain = "example.com"
def __init__(self):
super().__init__(
url="http://example.com/proxies.txt",
format='text', # auto-detect, text, json, csv, html
proto=('HTTP', 'HTTPS')
)
class JSONAPIProvider(SimpleProvider):
"""Provider for JSON APIs."""
domain = "api.example.com"
def __init__(self):
super().__init__(
url="http://api.example.com/proxies",
format='json',
proto=('HTTP', 'HTTPS', 'SOCKS5')
)Supported Formats:
text: Plain text list (IP:PORT per line)json: JSON response with proxy datacsv: CSV format proxy listshtml: HTML pages (uses default pattern matching)auto: Automatically detect format
For sites with pagination:
from proxybroker import PaginatedProvider
class PaginatedSite(PaginatedProvider):
domain = "proxypages.com"
def __init__(self):
super().__init__(
base_url="http://proxypages.com/list/page-{}.html", # {} = page number
start_page=1,
max_pages=10,
proto=('HTTP', 'HTTPS')
)URL Patterns:
http://site.com/page-{}.html- Page number in URL pathhttp://site.com/proxies?page={}- Page as query parameter
For APIs requiring authentication:
from proxybroker import APIProvider
class AuthenticatedAPI(APIProvider):
domain = "api.proxies.com"
def __init__(self, api_key):
super().__init__(
api_url="https://api.proxies.com/v1/list",
api_key=api_key,
response_format='json',
proxy_path='data.proxies', # Path to proxy list in JSON
proto=('HTTP', 'HTTPS')
)For complex sites requiring custom logic:
from proxybroker import Provider
import re
class ComplexProvider(Provider):
domain = "complex-site.com"
async def _pipe(self):
"""Custom scraping pipeline."""
# Step 1: Get main page
main_page = await self.get("http://complex-site.com/")
# Step 2: Extract proxy page URLs
urls = re.findall(r'href="(/proxy/\d+)"', main_page)
# Step 3: Fetch all proxy pages
full_urls = [f"http://complex-site.com{u}" for u in urls]
await self._find_on_pages(full_urls)
def find_proxies(self, page):
"""Custom proxy extraction."""
# Extract proxies from JavaScript
pattern = r'addProxy\("(\d+\.\d+\.\d+\.\d+)",\s*(\d+)\)'
return re.findall(pattern, page)name: My Simple Proxy List
type: simple
url: http://example.com/proxies.txt
format: text # auto|text|json|csv|html
pattern: null # Optional custom regex
protocols:
- HTTP
- HTTPS
max_connections: 4
timeout: 20name: My Paginated Site
type: paginated
url: http://example.com/proxies/page-{}.html
page_param: page # For query parameter style
start_page: 1
max_pages: 10
page_step: 1
protocols:
- HTTP
- HTTPS
- SOCKS4
- SOCKS5
max_connections: 4
timeout: 20{
"name": "My Proxy API",
"type": "api",
"url": "http://api.example.com/v1/proxies",
"api_key": "your-api-key",
"headers": {
"User-Agent": "ProxyBroker/2.0"
},
"response_format": "json",
"proxy_path": "data.proxies",
"protocols": ["HTTP", "HTTPS"],
"max_connections": 2,
"timeout": 30
}from proxybroker import Broker
# Load all providers from a directory
broker = Broker(provider_dirs=['./custom_providers/'])
# Combine with default providers
broker = Broker(provider_dirs=['./custom_providers/']) # Uses defaults + custom
# Only custom providers (no defaults)
broker = Broker(
providers=[], # Empty list = no defaults
provider_dirs=['./custom_providers/']
)from proxybroker import Broker, SimpleProvider
# Define provider in code
code_provider = SimpleProvider(
url="http://example.com/proxies.txt",
format='text'
)
# Load from both code and config files
broker = Broker(
providers=[code_provider],
provider_dirs=['./config_providers/']
)from proxybroker import create_provider_config_template
# Create a template configuration file
create_provider_config_template(
'my_provider.yaml',
provider_type='simple' # simple|paginated|api
)For sites with non-standard proxy formats:
class CustomPatternProvider(SimpleProvider):
def __init__(self):
# Pattern for format: "proxy://192.168.1.1@8080"
pattern = r'proxy://(\d+\.\d+\.\d+\.\d+)@(\d+)'
super().__init__(
url="http://site.com/proxies.html",
pattern=pattern
)Respect rate limits when scraping:
class RateLimitedProvider(Provider):
def __init__(self):
super().__init__(
url="http://site.com/proxies",
max_conn=1, # One connection at a time
timeout=30
)
async def get(self, url, **kwargs):
await asyncio.sleep(2) # 2 second delay
return await super().get(url, **kwargs)For complex authentication flows:
class OAuthProvider(Provider):
async def _pipe(self):
# Get OAuth token
token_response = await self.get(
"http://site.com/oauth/token",
data={"client_id": "...", "client_secret": "..."},
method="POST"
)
token = json.loads(token_response)['access_token']
# Use token for proxy list
await self._find_on_page(
"http://site.com/api/proxies",
headers={"Authorization": f"Bearer {token}"}
)class RobustProvider(Provider):
def find_proxies(self, page):
try:
# Try JSON parsing
data = json.loads(page)
return [(p['ip'], p['port']) for p in data['proxies']]
except:
# Fallback to regex
return self._find_proxies(page)- Be Respectful: Use appropriate
max_connand timeouts - Handle Errors: Always have fallback parsing logic
- Test First: Test your provider with a small limit
- Use Caching: Implement caching for expensive operations
- Document: Add docstrings explaining the site's format
- Check the URL is accessible
- Verify the format detection
- Test your regex pattern
- Enable debug logging:
import logging
logging.basicConfig(level=logging.DEBUG)If you get blocked:
- Reduce
max_connto 1 or 2 - Increase
timeout - Add delays between requests
- Use rotating user agents
For nested JSON structures:
# Response: {"success": true, "data": {"proxies": [...]}}
provider = APIProvider(
api_url="...",
proxy_path="data.proxies" # Navigate nested structure
)If you create a provider for a popular proxy source, consider contributing it:
- Test thoroughly
- Follow the existing code style
- Add to the PROVIDERS list in
providers.py - Submit a pull request
Here's a complete example for a fictional proxy site:
from proxybroker import Provider
import re
import json
class MyProxySite(Provider):
"""Provider for myproxysite.com - a fictional proxy listing site."""
domain = "myproxysite.com"
def __init__(self):
super().__init__(
url="https://myproxysite.com/api/proxies",
proto=('HTTP', 'HTTPS', 'SOCKS4', 'SOCKS5'),
max_conn=3,
timeout=30
)
async def _pipe(self):
"""Fetch proxy lists from multiple endpoints."""
# Get list of available proxy lists
index = await self.get(self.url + "/index")
try:
data = json.loads(index)
endpoints = data.get('endpoints', [])
except:
endpoints = ['/free', '/premium']
# Fetch each endpoint
urls = [self.url + endpoint for endpoint in endpoints]
await self._find_on_pages(urls)
def find_proxies(self, page):
"""Extract proxies from API response."""
proxies = []
try:
data = json.loads(page)
# Handle API response format
for item in data.get('proxies', []):
ip = item.get('ip')
port = str(item.get('port'))
protocols = item.get('protocols', ['HTTP'])
if ip and port:
# You can filter by protocol if needed
proxies.append((ip, port))
except json.JSONDecodeError:
# Fallback to HTML parsing
# Look for: <span class="proxy">192.168.1.1:8080</span>
pattern = r'<span class="proxy">(\d+\.\d+\.\d+\.\d+):(\d+)</span>'
proxies = re.findall(pattern, page)
return proxies
# Usage
if __name__ == "__main__":
import asyncio
import asyncio
from proxybroker import Broker
async def consume(q):
while True:
proxy = await q.get()
if proxy is None:
break
print(f"{proxy.host}:{proxy.port} [{proxy.types}]")
async def main():
proxies = asyncio.Queue()
broker = Broker(proxies, providers=[MyProxySite()])
await asyncio.gather(
broker.find(types=['HTTP', 'HTTPS'], limit=10),
consume(proxies),
)
asyncio.run(main())This provider:
- Fetches an index of available endpoints
- Handles both JSON and HTML responses
- Includes error handling
- Respects rate limits
- Can be easily modified for real sites