Enterprise workflow orchestration platform for data pipelines and ETL. Route Airflow HTTP operators and Python tasks through KnoxProxy for proxied API calls, web scraping DAGs, and geo-distributed data collection.
In the Airflow UI, go to Admin > Connections. Add a new connection of type HTTP with your proxy details.
Connection Id: knoxproxy
Connection Type: HTTP
Host: gw.knoxproxy.com
Port: 7000
Login: USER
Password: PASSEnsure requests is available in your Airflow environment.
pip install requestsCreate a new Python file in your dags/ folder for the proxied workflow.
Write a Python callable that uses requests with the KnoxProxy proxy configuration.
Modify the proxy username for country-specific routing.
proxies = {"https": "http://USER-country-jp:PASS@gw.knoxproxy.com:7000"}Trigger the DAG manually from the Airflow UI or CLI and verify the task logs show a proxied IP.
from datetime import datetime, timedelta
import requests
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.hooks.base import BaseHook
PROXY_CONN_ID = "knoxproxy"
def fetch_with_proxy(url: str, country: str = "", **kwargs):
"""Fetch a URL through KnoxProxy."""
conn = BaseHook.get_connection(PROXY_CONN_ID)
username = conn.login
if country:
username = f"{username}-country-{country}"
proxy_url = f"http://{username}:{conn.password}@{conn.host}:{conn.port}"
proxies = {"http": proxy_url, "https": proxy_url}
response = requests.get(url, proxies=proxies, timeout=30)
response.raise_for_status()
# Push result to XCom for downstream tasks
kwargs["ti"].xcom_push(key="response", value=response.text[:5000])
return response.status_code
with DAG(
dag_id="proxied_scraper",
start_date=datetime(2026, 1, 1),
schedule_interval=timedelta(hours=6),
catchup=False,
tags=["scraping", "proxy"],
) as dag:
scrape_us = PythonOperator(
task_id="scrape_us_endpoint",
python_callable=fetch_with_proxy,
op_kwargs={"url": "https://httpbin.org/ip", "country": "us"},
)
scrape_de = PythonOperator(
task_id="scrape_de_endpoint",
python_callable=fetch_with_proxy,
op_kwargs={"url": "https://httpbin.org/ip", "country": "de"},
)
scrape_us >> scrape_deEach PythonOperator task execution gets a fresh IP. For sticky sessions across tasks in a single DAG run, use USER-session-{run_id} where run_id is available from the Airflow context.
| Problem | Fix |
|---|---|
| requests.exceptions.ProxyError: Cannot connect to proxy | Verify the knoxproxy connection in Admin > Connections. Test connectivity from the Airflow worker: curl -x http://USER:PASS@gw.knoxproxy.com:7000 https://httpbin.org/ip |
| Task fails with 407 Proxy Authentication Required | Edit the knoxproxy connection and re-enter credentials. Ensure no URL-encoding issues with special characters. |
| Task timeout exceeded | Increase execution_timeout on the operator. Add retry logic with retries=3 and retry_delay=timedelta(seconds=10). |
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.
HttpOperator does not natively support proxy settings, since it lacks a parameter for routing HTTP calls through a proxy gateway. Use PythonOperator with the requests library instead, passing the proxies dict built from your Airflow Connection credentials, or write a custom HttpHook subclass that reads the same connection and applies KnoxProxy on every call.
Create the connection through the AIRFLOW_CONN_KNOXPROXY environment variable instead of the Admin UI, since managed platforms like MWAA and Cloud Composer often restrict direct UI access to connections. Set the variable to the full connection URI with your KnoxProxy username and password, and confirm outbound traffic to gw.knoxproxy.com:7000 is allowed through the platform network configuration.
Each task execution gets a fresh IP by default, since KnoxProxy assigns a new exit IP to every proxy connection Airflow opens. No additional configuration is needed for this automatic rotation, but you can keep the same IP across retries within one DAG run by using USER-session-{run_id} as the connection username.
Yes, Airflow 2.7+ supports deferrable operators that run async Python code, so you can use httpx.AsyncClient with the same proxy_url built from your Airflow Connection inside an async PythonOperator. This frees the worker slot while waiting on the proxied request, which helps when a DAG runs many concurrent KnoxProxy calls.
Rotating residential proxies -- 10 minutes setup, instant activation, 14-day money-back guarantee.