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, you can replace httpx.AsyncClient with requests.get(url, proxies={"https": proxy_url}) inside the same Prefect task without changing the retry or Secret block logic. Note that httpx supports async natively, which fits Prefect 3.x flows better when you need concurrent proxied requests across many tasks in one flow run.
Use Prefect task concurrency, either .map() to fan a task out over a list of URLs or asyncio.gather() inside a flow, to run proxied requests in parallel. Each concurrent task execution opens its own connection through KnoxProxy and gets its own rotated IP, so parallel runs do not share one exit address.
Yes, deploy the same flow to Prefect Cloud exactly as you would run it locally, since the proxy logic lives inside the task code rather than the deployment target. Just make sure the worker executing the flow has outbound network access to gw.knoxproxy.com:7000, and that Secret blocks for your credentials are registered in that workspace.
Set the proxy username to USER-session-MYSESSION, or dynamically to USER-session-{flow_run_id} using the Prefect runtime context, so every task shares one identifier. All requests that carry the same session ID reuse the same exit IP for the full flow run, which keeps cookies and rate limits consistent across tasks.
Rotating residential proxies -- 5 minutes setup, instant activation, 14-day money-back guarantee.