Python 中的 HTTPX 代理：配置与异步使用指南 | Geonode

当大多数新的 Python 项目发展到超出 requests 的适用范围时，httpx 便是它们首选的 HTTP 客户端，这主要是因为它为同步和异步代码提供了一套统一的 API。本指南将演示如何将 httpx 代理集成到 Client 和 AsyncClient 中，如何通过轮换的住宅端点进行身份验证，如何固定粘性会话，如何添加 SOCKS5，如何调整超时和重试设置，以及如何诊断您实际会遇到的三种错误。完成后，您将获得一个可重用的客户端，该客户端通过代理进行路由，并能验证其出口 IP。

示例针对 httpx 0.27 或更高版本。如果您使用的是较旧的版本，请注意：在 httpx 0.26 中，proxies= 关键字已重命名为 proxy= 和 mounts=；下文中的所有内容均使用当前名称。

配置 httpx 代理的两种方式：`proxy` 和 `mounts`

httpx 提供了一种简单的配置方式和一种精确的配置方式。proxy= 接受一个 URL，并将所有请求都路由到该 URL。 mounts= 将 URL 模式映射到传输方式，每种传输方式均可配置独立的代理、重试次数和连接限制。

选项	作用范围	适用场景
`proxy="http://..."`	整个客户端，所有协议	所有请求均通过一个代理，这通常是轮询网关的常见场景
`mounts={"https://": HTTPTransport(proxy=...)}`	按 URL 模式	根据协议或主机使用不同的代理，或对内部服务绕过代理
`httpx.get(url, proxy=...)`	单个顶级调用	无需长期保持连接的客户端进行的快速检查和脚本执行

简写形式如下：

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

with httpx.Client(proxy=PROXY, timeout=20.0) as client:
    r = client.get("https://api.ipify.org?format=json")
    print(r.status_code, r.json())

9000 端口是 HTTP(S) 轮询网关，因此每个新连接都会获得一个不同的住宅 IP。凭据以 Basic 认证的形式包含在 URL 中，httpx 会将其转换为 Proxy-Authorization 头部，无需任何额外代码。

按协议方案路由的注意事项

对于 mounts，您可以为 http:// 和 https:// 目标指定不同的传输方式。但需要注意的是，代理 URL 本身的协议方案。对于 https:// 目标，httpx 会向代理发送 CONNECT，然后在该隧道内与源服务器协商 TLS。通往代理的跳转采用明文 HTTP，因此 https:// 挂载中的代理 URL 通常应以 http:// 开头，而非 https://。若在此处写入 https://，则会指示 httpx 将代理跳转本身封装在 TLS 中——而大多数网关并不支持此操作，从而导致看似与代理无关的握手失败。

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

mounts = {
    "http://": httpx.HTTPTransport(proxy=PROXY),
    "https://": httpx.HTTPTransport(proxy=PROXY),  # http:// scheme is correct here
    # internal services stay direct: mount a transport without a proxy
    "all://*.internal.example": httpx.HTTPTransport(),
}

with httpx.Client(mounts=mounts) as client:
    print(client.get("https://api.ipify.org?format=json").json())

更具体的模式优先于通用模式，因此 all://*.internal.example 挂载会绕过该域名的代理，而其他所有流量则按常规路由。

URL 中的身份验证

Geonode 的凭据是明文用户名和密码，因此只需使用用户信息表单 http://user:pass@host:port 即可。有两点实用注意事项。首先，如果密码包含 @、: 或 / 之类的字符，请在构建 URL 之前使用 urllib.parse.quote 对其进行百分比编码。其次，通过从环境中读取密钥，避免将密钥放入源代码控制中：

import os
from urllib.parse import quote

import httpx

user = os.environ["GEONODE_USER"]
password = quote(os.environ["GEONODE_PASS"], safe="")
proxy = f"http://{user}:{password}@premium-residential.geonode.com:9000"

client = httpx.Client(proxy=proxy)

后续的所有内容都是通过扩展用户名（而非 URL）来添加定位和会话选项，因此相同的模式仍然有效。

`Client` 与 `AsyncClient` 的同步

异步客户端使用相同的 proxy= 关键字。唯一的区别出现在切换到 mounts 时：AsyncClient 需要 httpx.AsyncHTTPTransport，而不是 httpx.HTTPTransport。将同步传输传递给异步客户端会在首次请求时引发异常。

import asyncio
import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

async def fetch(client: httpx.AsyncClient, url: str) -> str:
    r = await client.get(url)
    r.raise_for_status()
    return r.json()["ip"]

async def main() -> None:
    transport = httpx.AsyncHTTPTransport(proxy=PROXY, retries=2)
    async with httpx.AsyncClient(
        mounts={"all://": transport}, timeout=httpx.Timeout(10.0, connect=5.0)
    ) as client:
        urls = ["https://api.ipify.org?format=json"] * 5
        ips = await asyncio.gather(*(fetch(client, u) for u in urls))
        print(ips)

asyncio.run(main())

由于端口 9000 会按连接轮转，因此五个并发请求通常会显示几个不同的 IP 地址。这是轮转池的预期行为；若需要它们保持一致，请使用粘性会话（见下文）。

SOCKS5 至 `httpx[socks]`

在 httpx 代理配置中，SOCKS 支持是可选的。请使用 pip install "httpx[socks]" 进行安装（该包会添加 socksio 包），然后切换方案。 Geonode 中的 SOCKS5 轮换端口为 11000：

import httpx

PROXY = "socks5://USERNAME:PASSWORD@premium-residential.geonode.com:11000"

with httpx.Client(proxy=PROXY) as client:
    print(client.get("https://api.ipify.org?format=json").json())

若未安装该附加组件，httpx 一旦检测到 socks5:// 代理 URL，便会触发一个提及 socksio 的 ImportError 错误。对于普通爬取任务，HTTP 端口更易于调试，因为 407 的响应内容清晰易读。

环境变量与 `trust_env`

在默认的 trust_env=True 环境下，httpx 会读取 HTTP_PROXY、HTTPS_PROXY、ALL_PROXY 和 NO_PROXY，并自动应用它们。这在工作站上很方便，但在持续集成（CI）环境中却容易造成混淆——因为环境中的企业代理会悄无声息地重写您的路由。如果您在代码中设置了代理，请考虑禁用环境变量查找，以确保在任何地方的行为都保持一致：

import httpx

client = httpx.Client(
    proxy="http://USERNAME:PASSWORD@premium-residential.geonode.com:9000",
    trust_env=False,
)

显式的 proxy= 已经优先于环境设置；trust_env=False 还会阻止 httpx 读取 SSL_CERT_FILE，因此请有意识地进行设置。

超时与传输层重试

代理会增加一个跳数，因此连接时间会延长，而默认的 5 秒超时对于家庭用户出口而言有时会显得较紧。httpx.Timeout 允许您设置总预算并单独设定连接限制，而 HTTPTransport(retries=...) 则会在请求发送之前就对失败的连接尝试进行重试。

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

transport = httpx.HTTPTransport(proxy=PROXY, retries=2)
timeout = httpx.Timeout(15.0, connect=8.0)

with httpx.Client(mounts={"all://": transport}, timeout=timeout) as client:
    r = client.get("https://api.ipify.org?format=json")
    print(r.json())

传输重试涵盖 ConnectError 以及建立连接过程中出现的类似故障。它们不会根据状态码进行重试，也不会在已发送字节后进行重试，因此来自网关的 502 或目标站点的 429 仍会触发您的代码。请在您自己编写的一个小型循环中处理这些情况，您还可以在该循环中决定是否更改会话 ID。

使用 Geonode 用户名语法实现粘性会话

对于登录、分页、下载等多步骤流程，每次连接都轮换 IP 地址是不正确的。对于此类场景，您的 httpx 代理必须在不同请求中保持相同的出口 IP 地址。 Geonode 将此信息编码到用户名中，并通过 HTTP(S) 粘性端口 10000 提供服务：

USER[-type-residential|datacenter|mix][-country-xx][-state-name|-city-name][-asn-N][-session-XXXXXXXX][-lifetime-SECONDS]

会话 ID 由您生成的 8 个字母数字字符组成；有效期为 180 至 86400 秒（24 小时）。州和城市不能组合使用，且 asn 需要指定国家。

import secrets
import httpx

def sticky_client(country: str = "us", lifetime: int = 600) -> httpx.Client:
    session_id = secrets.token_hex(4)  # 8 hex characters
    username = f"USERNAME-country-{country}-session-{session_id}-lifetime-{lifetime}"
    proxy = f"http://{username}:PASSWORD@premium-residential.geonode.com:10000"
    return httpx.Client(proxy=proxy, timeout=httpx.Timeout(15.0, connect=8.0))

with sticky_client() as client:
    first = client.get("https://api.ipify.org?format=json").json()["ip"]
    second = client.get("https://api.ipify.org?format=json").json()["ip"]
    print(first == second, first)

新的 session_id 表示新的 IP 地址，因此“手动”轮换只需创建一个新的客户端即可。请根据任务需求将粘性会话的生存时间设置得尽可能短。

验证出口 IP

上述每个示例都会访问 https://api.ipify.org?format=json，无论哪个地址到达该服务器，它都会返回 {"ip": "..."}。只需将其与一次直接调用（不经代理）进行对比，你就能得到一个仅需两行的快速测试，从而证明代理确实位于请求路径中。在调试其他问题之前请先执行此操作：许多“httpx 代理无法工作”的报告，其实是因为环境变量或挂载模式接管了请求，导致请求从未经过代理。

处理 407 `ProxyError` 和 `ConnectError`

几乎所有 httpx 代理问题都由三种故障引起。下表列出了每种故障的成因及解决方法。

症状	异常	常见原因	解决方法
响应状态码 407	`httpx.ProxyError`（在为 HTTPS 目标建立隧道时触发）或针对 HTTP 目标的普通 407 响应	凭据错误，或网关不接受的用户名选项	检查用户名语法；对密码进行百分比编码
隧道被拒绝，代理返回 502/503	`httpx.ProxyError`	网关无法打开隧道，或协议端口错误	确认 HTTP(S) 端口为 9000/10000， SOCKS5 请使用 11000/12000
无法连接到代理主机	`httpx.ConnectError`	DNS、防火墙或主机名错误	手动解析主机；延长 `connect` 超时
请求挂起后失败	`httpx.ConnectTimeout` / `httpx.ReadTimeout`	住宅网出口速度慢或目标被封锁	增加超时时间，使用新的会话 ID 重试

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

def get_via_proxy(url: str) -> httpx.Response | None:
    try:
        with httpx.Client(proxy=PROXY, timeout=httpx.Timeout(15.0, connect=8.0)) as client:
            r = client.get(url)
            if r.status_code == 407:
                raise RuntimeError("proxy rejected credentials (407)")
            return r
    except httpx.ProxyError as e:
        print("proxy refused tunnel:", e)
    except httpx.ConnectError as e:
        print("cannot connect to proxy host:", e)
    except httpx.TimeoutException as e:
        print("timed out:", type(e).__name__)
    return None

resp = get_via_proxy("https://api.ipify.org?format=json")
print(resp.json() if resp else "failed")

所有 httpx 异常均继承自 httpx.HTTPError，因此使用单一的通用代码 except 即可满足日志记录需求；但将 ProxyError 与 ConnectError 区分开来，可帮助您判断是应检查凭据还是网络问题。

常见错误

在代理 URL 中使用 `https://`

如上所述，代理 URL 方案描述了通往代理的跳转路径。即使目标是 https://，HTTP(S) 端口也应使用 http://。请将 https:// 保留给向代理显式声明支持 TLS 的网关。

将 `HTTPTransport` 混入 `AsyncClient` 中

位于 AsyncClient 上的 mounts 必须使用 AsyncHTTPTransport。该同步传输在构造时会被接受，但在请求时会因一个毫无帮助的错误而失败，因此，在从未发送过请求的测试中，这一点很容易被忽略。

预期从轮转端口获取一个 IP 地址

端口 9000 会随每个连接轮转。如果连续两个请求必须共享同一 IP，请使用端口 10000，并在用户名中包含 -session- 标识。仅重复使用 Client 并不能锁定 IP，因为连接池中的连接可能会被替换。

允许环境变量覆盖代码

Docker 镜像或 CI 运行器中出现的零星 HTTPS_PROXY 会将您以为是直连的流量进行路由，而 NO_PROXY 可能会将您预期应通过代理访问的主机排除在外。当代理路由必须是确定性的时，请设置为 trust_env=False。

下一步

一旦上述客户端返回一个住宅IP，且跨请求保持粘性会话，剩下的变量就是其背后的代理池。Geonode的住宅代理提供了本指南中使用的端点： 250 万个在线 IP，每月在 195 多个国家/地区轮换 3000 万至 5000 万个独立 IP，可通过用户名语法实现国家、州、城市和 ASN 定向，粘性会话最长可达 24 小时。批量采购价格低至 0.27 美元/GB，入门套餐为 10 GB，月费 7.92 美元，3 天试用费为 5 美元。如果您打算基于此构建爬虫，网页抓取用例详细介绍了如何将代理与重试逻辑结合使用，而轮换数据中心代理对于不封锁数据中心IP范围的目标而言，是更经济的选项。

常见问题

如何为所有请求设置 httpx 代理？

将代理 URL 传递给客户端构造函数：httpx.Client(proxy="http://user:pass@host:port")。通过该客户端发出的所有请求（无论目标是 http:// 还是 https://），都会通过该代理进行路由。该关键字同样适用于 httpx.AsyncClient 以及 httpx.get() 等顶级助手。URL 中的凭据会自动作为 Proxy-Authorization 发送。

httpx 中的代理和挂载有什么区别？

proxy 是一个快捷方式，它将所有流量都路由到同一个代理。mounts 将 URL 模式（如 "http://" 或 "https://"）映射到传输层，每个传输层都可以拥有自己的代理、重试和限制。当您需要按方案或按主机使用不同的代理，或者希望通过挂载不带代理的传输来绕过某些目标的代理时，请使用 mounts。

为什么当代理 URL 以 https://? 开头时，我的 https:// 挂载会失败？

代理 URL 中的方案描述的是 httpx 与代理之间的通信方式，而非目标。对于 https:// 目标，httpx 会通过与代理的明文 HTTP 连接建立 CONNECT 隧道，因此代理 URL 通常应使用 http://。使用 https:// 会要求 httpx 与代理本身协商 TLS，而大多数代理端点并不支持此功能。

httpx 是否支持 SOCKS5 代理？

支持，但需先安装带 pip install httpx[socks] 的可选依赖项，该依赖项会引入 socksio。随后，请像配置 HTTP 代理一样，精确地传入 proxy="socks5://user:pass@host:port"。使用 Geonode 时，SOCKS5 的轮询端口为 11000，SOCKS5 的粘性端口为 12000；HTTP(S) 端口不接受 SOCKS 握手。

当代理连接失败时，httpx 能否自动重试请求？

仅在传输层。httpx.HTTPTransport(retries=2) 会重试失败的连接尝试（包括连接到代理时的错误），但绝不会重试已收到 407 或 502 等响应的请求。针对 HTTP 状态码的重试应由您在自己的循环中处理，您可以在其中切换会话 ID 或有意识地进行退避。

httpx 代理配置在使用住宅 IP 时是否存在限制？

代理仅会更改您的出口 IP；它不会改变网站允许的行为。请遵守 robots.txt、服务条款和速率限制，保持适度的并发量，并避免收集您无权处理的个人数据。住宅 IP 会话的时长上限为 24 小时，超过此时间后 IP 将被释放，您必须开启新的会话。

Python 中的 HTTPX 代理：配置、身份验证和异步请求

配置 httpx 代理的两种方式：proxy 和 mounts