Modern Python workflow orchestration framework with native async support. Route Prefect flow HTTP requests through KnoxProxy for proxied web scraping, API monitoring, and distributed data collection tasks.
Install Prefect and httpx in your Python environment.
pip install prefect httpxStore proxy credentials securely using a Prefect Secret block.
prefect block register -m prefect.blocks.systemCreate a Prefect task that routes HTTP requests through KnoxProxy using httpx.
Compose tasks into a Prefect flow with retry logic and concurrency.
Pass a country parameter to dynamically change the proxy username.
proxy_url = f"http://USER-country-{country}:PASS@gw.knoxproxy.com:7000"Run the flow locally or deploy to Prefect Cloud / self-hosted server.
python proxied_flow.pyimport httpx
from prefect import flow, task, get_run_logger
from prefect.blocks.system import Secret
PROXY_BASE = "gw.knoxproxy.com:7000"
@task(retries=3, retry_delay_seconds=5)
async def fetch_url(url: str, country: str = "") -> dict:
"""Fetch a URL through KnoxProxy with optional geo-targeting."""
logger = get_run_logger()
user = Secret.load("knoxproxy-user").get()
password = Secret.load("knoxproxy-pass").get()
username = f"{user}-country-{country}" if country else user
proxy_url = f"http://{username}:{password}@{PROXY_BASE}"
async with httpx.AsyncClient(
proxy=proxy_url,
timeout=30.0,
) as client:
response = await client.get(url)
response.raise_for_status()
logger.info(f"Fetched {url} via {country or 'auto'}: {response.status_code}")
return response.json()
@flow(name="proxied-scraper", log_prints=True)
async def proxied_scraper(urls: list[str], country: str = ""):
"""Scrape multiple URLs through KnoxProxy."""
results = []
for url in urls:
result = await fetch_url(url, country=country)
results.append(result)
print(f"Completed {len(results)} requests")
return results
if __name__ == "__main__":
import asyncio
asyncio.run(proxied_scraper(
urls=["https://httpbin.org/ip", "https://httpbin.org/headers"],
country="us",
))Each task execution gets a fresh IP by default. For sticky sessions across tasks in a flow run, use USER-session-{flow_run_id} where the flow run ID is available from the Prefect runtime context.
| Problem | Fix |
|---|---|
| httpx.ProxyError: 407 Proxy Authentication Required | Verify secrets with: prefect block inspect secret/knoxproxy-user. Re-save if incorrect. |
| httpx.ConnectTimeout: timed out | Increase the timeout parameter in httpx.AsyncClient. The task retries (retries=3) will handle transient failures. |
| ValueError: Secret block not found | Create them: prefect block create secret --name knoxproxy-user --value YOUR_USER |
USER-country-de-city-berlin-session-profile07Order matters -- geo flags before the session flag. The session name is free text; use the profile ID so the mapping is self-documenting. Password stays as issued; no flags belong there. HTTP on :7000, SOCKS5 on :7001, same credentials.
Yes. Replace httpx.AsyncClient with requests.get(url, proxies={"https": proxy_url}). However, httpx supports async natively which is better for Prefect 3.x.
Use Prefect task concurrency with .map() or asyncio.gather(). Each concurrent task gets its own proxy IP.
Yes. Deploy the flow to Prefect Cloud and ensure your worker has outbound access to gw.knoxproxy.com:7000.
Set the proxy username to USER-session-MYSESSION. All requests with the same session ID reuse the same IP.
Free trial on rotating residential -- 5 minutes setup, no credit card.