How MediaCrawler's Proxy IP Pool Works with Multiple Providers

MediaCrawler abstracts proxy management behind a provider-agnostic pool system that validates, rotates, and automatically refreshes IP addresses from static strings or third-party services like Kuai Daili and Wandou HTTP.

MediaCrawler is an open-source crawling framework that handles proxy rotation through a modular proxy IP pool system designed to work seamlessly across different providers. The architecture decouples proxy acquisition from consumption, allowing developers to switch between static proxies and dynamic commercial providers without modifying the core crawling logic. Understanding how this system manages provider-specific implementations is essential for maintaining high-availability scraping operations.

Core Architecture Components

The proxy system centers on four interconnected components defined in separate modules under the proxy/ directory.

ProxyIpPool (in proxy/proxy_ip_pool.py) acts as the central manager. It maintains an internal list of proxy objects, handles validation, and supplies fresh IPs on demand. The pool operates independently of the source, meaning the same ProxyIpPool class manages proxies whether they come from a static string or a paid API.

ProxyProvider (in proxy/base_proxy.py) defines the abstract interface that every provider must implement. The base class enforces a single method signature: async def get_proxy(self, num: int) -> List[IpInfoModel]. This contract ensures that the pool can request any number of proxies from any provider without knowing the implementation details.

IpInfoModel and ProviderNameEnum (in proxy/types.py) provide the data structures. IpInfoModel is a dataclass containing fields for ip, port, protocol, username, password, and expired_time_ts. The enum defines supported providers including KUAI_DAILI_PROVIDER, WANDOU_HTTP_PROVIDER, and STATIC_PROVIDER.

Concrete providers in proxy/providers/ implement the acquisition logic. For example, kuaidl_proxy.py fetches from the Kuai Daili API, while wandou_http_proxy.py handles Wandou HTTP integration.

Provider Registration and Configuration

At module load time, the system creates a registry dictionary named IpProxyProvider that maps string identifiers to instantiated provider objects:

IpProxyProvider = {
    ProviderNameEnum.KUAI_DAILI_PROVIDER.value: new_kuai_daili_proxy(),
    ProviderNameEnum.WANDOU_HTTP_PROVIDER.value: new_wandou_http_proxy(),
    ProviderNameEnum.STATIC_PROVIDER.value: StaticProxyProvider(),
}

The runtime selection depends on the config.IP_PROXY_PROVIDER_NAME setting. When create_ip_pool() (around lines 198-218 in proxy/proxy_ip_pool.py) executes, it retrieves the appropriate provider from this dictionary. If the configuration name does not exist in IpProxyProvider, the code raises a descriptive ValueError (lines 205-210).

For static proxies, the system uses StaticProxyProvider, which parses a single URL from config.STATIC_PROXY_URL and returns it as a one-element list. This provider is defined in the same file as ProxyIpPool.

Initializing the Pool and Loading Proxies

The factory function create_ip_pool() handles initialization logic specific to provider types:

ip_provider = IpProxyProvider.get(config.IP_PROXY_PROVIDER_NAME)
pool = ProxyIpPool(
    ip_pool_count=ip_pool_count,
    enable_validate_ip=False if is_static else enable_validate_ip,
    ip_provider=ip_provider,
)
await pool.load_proxies()

The function automatically disables validation for static proxies since they never expire, while enabling it for dynamic providers that require health checks.

When load_proxies() executes, it delegates to the provider's get_proxy() method:

self.proxy_list = await self.ip_provider.get_proxy(self.ip_pool_count)

Dynamic providers asynchronously call their respective external APIs and translate the JSON responses into IpInfoModel objects. The static provider, conversely, parses the configured URL string into an IpInfoModel (lines 75-85) without making network requests.

Optional Proxy Validation

When enable_validate_ip is True, the pool validates proxies before returning them. The _is_valid_proxy() method performs an HTTP GET request to self.valid_ip_url (defaults to https://echo.apifox.cn/).

The validation process constructs a proxy URL from the IpInfoModel fields, creates an httpx async client using make_async_client from tools/httpx_util.py, and checks for a 200 status response. If validation fails, the pool discards the proxy and attempts another one from the list.

Retrieving and Auto-Refreshing Proxies

The get_proxy() method selects a random entry from the pool, removes it from the internal list, and stores it as self.current_proxy. If the pool empties, it automatically triggers _reload_proxies() to fetch a fresh batch from the provider.

For expiration-aware rotation, the get_or_refresh_proxy() method (lines 30-44) checks is_current_proxy_expired() before returning the current proxy. This function compares the current timestamp against the expired_time_ts field in IpInfoModel. When the proxy expires, the pool automatically refreshes from the provider before supplying a new IP.

Extending the System with Custom Providers

Adding a new proxy source requires three steps:

1. Subclass ProxyProvider in proxy/base_proxy.py and implement async def get_proxy(self, num: int) -> List[IpInfoModel].
2. Register the instance in the IpProxyProvider dictionary with a unique string key matching a new ProviderNameEnum entry.
3. Update configuration by setting config.IP_PROXY_PROVIDER_NAME to the new provider's key.

Because all pool logic remains in ProxyIpPool, the rest of the crawler code requires no changes regardless of the proxy source.

Implementation Examples

Initialize a pool using the configured provider:

from proxy.proxy_ip_pool import create_ip_pool

# Assumes config.IP_PROXY_PROVIDER_NAME = "kuai_daili"

pool = await create_ip_pool(ip_pool_count=10, enable_validate_ip=True)

# Get a validated proxy for an outgoing request

proxy = await pool.get_or_refresh_proxy()
proxy_url = f"{proxy.protocol}{proxy.ip}:{proxy.port}"

async with httpx.AsyncClient(proxies=proxy_url) as client:
    resp = await client.get("https://example.com")

Use a static proxy without validation:

from proxy.proxy_ip_pool import StaticProxyProvider

static = StaticProxyProvider()
static_list = await static.get_proxy(num=1)
print(static_list[0].ip, static_list[0].port)  # e.g., 192.0.2.10 8080

Implement a custom provider:

class MyCustomProvider(ProxyProvider):
    async def get_proxy(self, num: int) -> List[IpInfoModel]:
        # Fetch from your service and return IpInfoModel objects

        return [IpInfoModel(ip="10.0.0.1", port=8080, protocol="http://", expired_time_ts=...)]

# Register and use

IpProxyProvider["my_custom"] = MyCustomProvider()
config.IP_PROXY_PROVIDER_NAME = "my_custom"
pool = await create_ip_pool(5, enable_validate_ip=False)

Summary

• MediaCrawler's proxy IP pool uses a provider-agnostic architecture where ProxyIpPool manages consumption while ProxyProvider subclasses handle acquisition.
• The IpProxyProvider registry maps configuration names to provider instances, supporting static strings, Kuai Daili, and Wandou HTTP out of the box.
• Optional validation via httpx checks proxy health against https://echo.apifox.cn/ before use.
• Automatic expiration tracking using expired_time_ts ensures the pool refreshes IPs before they become invalid.
• New providers integrate by implementing a single abstract method and registering in the provider dictionary without modifying core crawler logic.

Frequently Asked Questions

What is the difference between static and dynamic providers in MediaCrawler?

Static providers parse a single proxy URL from config.STATIC_PROXY_URL and return it as a one-element list, never expiring. Dynamic providers like Kuai Daili and Wandou HTTP fetch live proxies from external APIs, include expiration timestamps, and support bulk retrieval. The pool disables validation automatically for static providers since they don't expire, while dynamic providers benefit from pre-flight health checks.

How does MediaCrawler validate proxy IP addresses?

When enable_validate_ip is enabled, the ProxyIpPool._is_valid_proxy() method performs an HTTP GET request through the candidate proxy to https://echo.apifox.cn/. It uses httpx via make_async_client from tools/httpx_util.py and expects a 200 status response. Failed proxies are discarded immediately, and the pool attempts the next available IP.

Can I use multiple proxy providers simultaneously?

The current architecture uses a single provider per pool instance determined by config.IP_PROXY_PROVIDER_NAME. To use multiple providers simultaneously, you would instantiate separate ProxyIpPool objects with different providers and implement custom rotation logic at the application level. The modular design in proxy/proxy_ip_pool.py makes this extension straightforward.

How do I know when a proxy has expired?

The IpInfoModel dataclass includes an expired_time_ts field (Unix timestamp) that indicates when the proxy becomes invalid. The get_or_refresh_proxy() method checks this timestamp against the current time. If the proxy is expired or the pool is empty, it automatically triggers a reload from the provider before returning a fresh IP.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}