周一至周五 09:00 - 18:00(UTC+08:00) 
©2026 NST LABS TECH LTD. 保留所有权利。 构建一个Python代理服务器:逐步指南(2026)Marcus ChenProduct & Network Architect
逐步指南:构建Python代理服务器
TL;DR
- 一个 Python 代理服务器需要两条不同的代码路径,而不是一条。 普通 HTTP 请求以完整的请求行到达,服务器可以直接转发;而 HTTPS 请求以
CONNECT 请求到达,服务器必须以不透明的方式进行隧道处理,而不是解析。
asyncio 处理许多同时连接,无需每个连接一个线程上限。 本指南中构建的代理通过五个并发请求的总完成时间不足 40 毫秒,没有任何请求阻塞其他请求。
CONNECT 方法使 HTTPS 能够通过代理工作, 这是大多数从头开始的教程跳过或未测试的步骤——这篇教程实现了它并在真实的 HTTPS 网站上证明了这一点。
- 添加
Proxy-Authorization: Basic 支持将开放中继转变为经过认证的中继, 而且差异是可以验证的:错误或缺失的凭证返回真实的 407,正确的则返回 200。
- 一个自托管的代理具有与您运行它的机器数量相同的出口 IP 数—通常为一个。 这对于本地开发或单区域中继是可以的,但如果目标是将请求分散到多个 IP 上,您将会遇到阻碍。
- 这里没有解密或检查 HTTPS 流量。
CONNECT 隧道直接转发加密字节,这是个人转发代理正确且不令人意外的行为,避免了管理用于拦截的 TLS 证书。
引言:编写您自己的转发代理,而不仅仅是使用一个
转发代理位于客户端和互联网之间,接受客户端的请求并代表其发出请求。在 Python 中构建一个代理是一个真正不同的练习,与从 Python 脚本中使用代理完全不同——指向别人的网关只需几行代码;而制作自己的进程正确地转发普通 HTTP 和 HTTPS,同时处理多个客户端并拒绝未授权使用,是大多数从头开始的教程所停下的地方。本指南在 Python 的标准库中端到端构建了该服务器,针对真实目标测试了每条路径,并诚实地说明了自托管中继在实践中的限制。
这里的构建仅使用 asyncio,它随 Python 3.7+ 附带——核心服务器不需要额外的包。所有内容都通过运行验证:下面的每个代码块都在本地测试服务器或真实的 HTTPS 网站上执行,确切的命令和结果与代码一起展示,而不仅仅是描述。像这样的自托管代理也是 网络条件测试和质量保证工作流程 的常见构建模块,团队希望在请求离开网络之前对它实际执行的操作拥有完全的可见性和控制。
安装:您需要什么(以及不需要什么)
代理服务器本身没有外部依赖—asyncio、base64 和 sys 都是任何 Python 3.7+ 安装的标准库的一部分。有两个东西仅用于测试,而不用于代理本身:
-x
requests 库(pip install requests),确认代理也按照正常的 Python HTTP 客户端的期望工作——关键字“python 代理服务器”所暗示的确切组合,涵盖了在 Python 中编写代理和从 Python 代码驱动一个代理。这里的任何内容都不需要根权限或特定操作系统;在示例中,服务器在 127.0.0.1 上绑定一个普通的 TCP 套接字,将 0.0.0.0 替换(加上限制谁可以访问的防火墙规则)是接受来自其他机器连接所需的唯一更改。
快速浏览
如果目标是通过多个出口 IP 路由流量,而不是了解代理如何在内部工作,Nstproxy 的网关为您提供了一个现成的 `host:port`,您可以将其指向 HTTP/SOCKS5 客户端,而无需自己维护此中继代码。
配置监听套接字和请求解析
转发代理需要为每个连接做三件事:接受连接,读取足够的请求以知道要去哪里,并相应地进行隧道或中继。asyncio.start_server 处理接受步骤,并为每个连接提供一个 (reader, writer) 对:
import asyncio
async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
first_line = await reader.readline()
if not first_line:
writer.close()
return
header_lines = [first_line.decode(errors="replace").rstrip("\r\n")]
while True:
line = await reader.readline()
if line in (b"\r\n", b"\n", b""):
break
header_lines.append(line.decode(errors="replace").rstrip("\r\n"))
method, target, _ = header_lines[0].split(" ", 2)
# method 为 "CONNECT" 表示 HTTPS,或 "GET"/"POST"/等表示普通 HTTP
请求行是分支点:CONNECT host:port HTTP/1.1 意味着客户端希望建立一个 HTTPS 隧道,并且不希望代理读取其实际流量;任何其他请求都是代理可以解析并自行转发的普通 HTTP 请求。
基本实现:转发普通 HTTP 请求
对于非 CONNECT 请求,目标要么在请求行本身中(绝对格式,GET http://host:port/path HTTP/1.1),要么在 Host: 头中。代理拨打该目标,重建请求,不包含真实服务器不会期望的 Proxy-* 头,并在双向传输字节:
async def pipe(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
try:
while True:
data = await reader.read(65536)
if not data:
break
writer.write(data)
await writer.drain()
except (ConnectionResetError, BrokenPipeError):
pass
finally:
try:
writer.close()
except Exception:
pass
在 handle_client 中,非 CONNECT 分支解析目标并重建请求:
if target.startswith("http://"):
rest = target[len("http://"):]
host_port, _, path = rest.partition("/")
path = "/" + path
else:
path = target
host_port = next(
(h.split(":", 1)[1].strip() for h in header_lines[1:] if h.lower().startswith("host:")),
None,
)
host, _, port = host_port.partition(":")
port = int(port or 80)
remote_reader, remote_writer = await asyncio.open_connection(host, port)
rebuilt = f"{method} {path} HTTP/1.1\r\n"
for h in header_lines[1:]:
if not h.lower().startswith("proxy-"):
rebuilt += h + "\r\n"
rebuilt += "\r\n"
remote_writer.write(rebuilt.encode())
await remote_writer.drain()
await asyncio.gather(pipe(remote_reader, writer), pipe(reader, remote_writer))
在一个抛弃的本地 HTTP 服务器(http.server,绑定到 127.0.0.1:9000,返回固定内容)上进行了测试,代理监听 127.0.0.1:8080:
curl -x http://127.0.0.1:8080 http://127.0.0.1:9000/ -w "\nHTTP_STATUS:%{http_code}\n"
那返回的内容是 hello-from-local-target 和 HTTP_STATUS:200。通过 curl -v 确认请求确实通过代理传输(> GET http://127.0.0.1:9000/ HTTP/1.1 发送到端口 8080,响应被转发回来),而不是 curl 直接连接到目标——这是任何沙箱测试环境中的真实风险,其中 no_proxy 设置可以静默绕过某些主机的代理,因此验证实际路径比仅仅信任最终状态代码更重要。
高级模式:HTTPS 隧道、身份验证和并发
使用 CONNECT 隧道 HTTPS
仅处理上述情况的代理无法承载 HTTPS 流量——客户端即将与目标开始 TLS 握手,而代理无权(或没有能力,没有目标站点的私钥)检查该流程。解决办法是 CONNECT 方法:代理打开到请求的 host:port 的原始 TCP 连接,回复 200 Connection Established,从那时起,只是双向传输字节而不查看它们:
if method == "CONNECT":
host, _, port = target.partition(":")
port = int(port or 443)
remote_reader, remote_writer = await asyncio.open_connection(host, port)
writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
await writer.drain()
await asyncio.gather(
pipe(reader, remote_writer),
pipe(remote_reader, writer),
)
这是 MDN 的 CONNECT 方法参考 所描述的确切机制:“CONNECT HTTP 方法请求代理为目标服务器建立一个 HTTP 隧道,如果成功,盲目地在两个方向上转发数据,直到隧道关闭。” 在经过代理的 8080 端口对真实 HTTPS 网站(非伪造)进行测试:
curl -x http://127.0.0.1:8080 https://pypi.org/pypi/requests/json
curl -v 在同一命令中确认了完整路径:发送到代理的 CONNECT pypi.org:443,返回 200 Connection Established,然后是 SSL connection using TLSv1.3 / TLS_AES_256_GCM_SHA384,最后是 pypi.org 本身的 HTTP/2 200 响应——TLS 握手在 curl 和 pypi.org 之间的隧道内端到端发生,代理从未看到明文。相同的 URL 也可以通过 Python 的 requests 库在指定代理的情况下正常工作,使用 proxies={"http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080"},返回 200 和有效的 JSON——确认服务器的行为像真实 HTTP 客户端库中的代理,而不仅仅是针对 curl。
需要认证
公共互联网上的开放中继会迅速被滥用。添加基本认证意味着在进行任何转发之前检查 Proxy-Authorization 头,当缺失或错误时返回 407(代理特有的 401 等价物):
import base64
AUTH = base64.b64encode(b"devuser:s3cret").decode() # 通常从配置中加载,而非硬编码
def check_auth(headers: list[str]) -> bool:
for h in headers:
if h.lower().startswith("proxy-authorization:"):
value = h.split(":", 1)[1].strip()
if value.startswith("Basic "):
return value[len("Basic "):].strip() == AUTH
return False
async def send_407(writer: asyncio.StreamWriter):
body = b"Proxy Authentication Required"
writer.write(
b"HTTP/1.1 407 Proxy Authentication Required\r\n"
b'Proxy-Authenticate: Basic realm="proxy"\r\n'
b"Content-Length: " + str(len(body)).encode() + b"\r\n"
b"Connection: close\r\n\r\n" + body
)
await writer.drain()
writer.close()
在端口 8081 上启用了认证的服务器的实时结果:
| 请求 | 结果 |
|---|
无 Proxy-Authorization 头 | 407 Proxy Authentication Required |
错误的凭据(devuser:wrongpass) | 407 Proxy Authentication Required |
| 正确的凭据,普通 HTTP 目标 | 200,主体正确转发 |
| 正确的凭据,通过 CONNECT 的 HTTPS 目标 | 200 |
两个失败案例是通过实际发送错误或缺失的凭据重现的,而不是从阅读代码中断言的——该项目对每个发布的代理配置声明都适用相同的规范。
无需为每个连接开辟线程的并发
因为 handle_client 是一个协程,asyncio 的事件循环在单个线程上并发运行多个连接,而不是为每个客户端创建一个操作系统线程,这是大多数从零开始的代理教程使用的模式。在同一个运行中的代理实例上同时进行五个请求:
time (for i in 1 2 3 4 5; do
curl -s -x http://127.0.0.1:8080 http://127.0.0.1:9000/ -o /dev/null -w "req$i:%{http_code} " &
done; wait)
这打印了 req1:200 req2:200 req3:200 req4:200 req5:200 和 real 0m0.033s。所有五个在 33 毫秒内完成,没有请求在等待另一个——这是对 asyncio 流 API 的文档行为,其中 start_server 的回调每个连接运行一次,作为一个独立的协程,而不是阻塞调用。
自制代理服务器的真实限制
以上一切都是一个真正有效的转发代理——但在依赖它用于本地开发或单区域中继之外的任何用途之前,仍有值得了解的界限。
最具体的例子是在目标网站开始屏蔽代理的IP时出现:这个服务器只有一个出口IP,即它运行的机器的地址,因此在那里被屏蔽会导致所有在其后面的客户端被阻塞,直到该IP更改。这时,旋转网关就成为了比本指南中构建的服务器更直接的工具——Nstproxy的Residential Lite系列在一个host:port后面运行许多独立的住宅出口IP,使用与网关相同的HTTP/SOCKS5客户端连接格式,因此本应在一个被屏蔽的IP上停滞的请求会通过另一个IP发送,客户端代码不需要知道发生了轮换。它以预付费、按需计费的模式计费,详细见Residential Lite定价页面,而不是固定订阅,它是为需要在一个大型、地理分散的IP池(200多个国家和地区,根据Nstproxy自己的HTTP代理工作原理解释)中分散流量的脚本和服务而建的,而不是为特别想拥有和修改中继代码的团队而设计——如果在代理层检查或记录流量是真正的目标,那么像上面提到的自托管服务器仍然是合适的工具,而旋转网关是一个补充的上游跳跃,而不是它的替代品。
- 大型、分布式出口IP池——许多独立的住宅IP在网关后意味着一个被屏蔽或限速的IP不会像在这个自托管服务器的一个地址上那样停止整个工作。
- 相同的客户端协议形态——网关在一个端点上支持HTTP、HTTPS和SOCKS5,因此根据本指南的代理(或根据
requests的proxies字典)编写的代码只需略微不同的host:port即可指向网关。
- 全球出口区域——当目标网站的内容、可用性或速率限制因请求者位置而异时,这很有用,而仅靠一个自托管服务器无法在每个区域内部署实例来重复这一点。
这个构建故意不做两件事,也不应假设在没有更多工作的情况下会做到:它不解密或检查HTTPS有效负载(CONNECT隧道因设计而不透明,这是个人中继的正确做法,也避免了管理TLS证书以进行拦截),并且它不持久化日志、不限制客户端速率,或强制访问列表,除了显示的单个基本身份验证凭证——一个暴露在127.0.0.1之外的代理需要至少每个客户端的凭证和连接限制,才能在无人看管的情况下安全运行。
排查常见错误
**curl: (7) Failed to connect**几乎总是意味着代理进程未在curl给定的地址/端口上监听,或者防火墙阻止了该端口——在检查代理代码之前,使用ss -tlnp | grep <port>确认是否真的绑定在该地址上。
即使代理端口错误,请求似乎仍然成功,或错误的凭证似乎有效——检查no_proxy/NO_PROXY是否在shell环境中设置,并且包括目标主机;几个沙盒和CI环境默认设置此项,curl或requests会完全静默绕过该列表中的任何主机配置的代理。在测试命令前运行env -u no_proxy -u NO_PROXY以排除此项,并使用curl -v确认请求行显示的是代理的端口,而不是直接连接。
**从该代理返回的502 Bad Gateway**意味着代理未能到达目标主机——检查目标主机名是否解析,并且端口可以从运行代理的机器访问,而不是从客户端访问。
HTTPS可以使用,但普通HTTP不能(或反之亦然)——这几乎总能追溯到请求行的分支:确认客户端实际上正在为HTTPS发送CONNECT(一些HTTP客户端库需要一个明确的https代理条目,独立于http,才能执行此操作),以及普通HTTP的绝对请求行或Host:头。
结论
一个有效的Python代理服务器归结为两种请求形态的不同处理——普通HTTP转发并重建,以及通过CONNECT不透明地隧道化的HTTPS——加上部署实际上所需的任何身份验证和并发处理。每个部分,包括大多数快速教程跳过的部分(真正的HTTPS隧道、真正的407、真正的并发请求),都是在本指南中对一个实时目标进行的实验,而不是凭记忆描述的。当这个构建的单一出口IP上限成为实际瓶颈时,就是转向管理旋转网关而不是进一步扩展此代码的时机。
常见问题解答
问:您自己构建的 Python 代理服务器可以处理 HTTPS 流量吗?
可以,但仅通过实现 CONNECT 方法并隧道加密字节而不进行检查——一个只解析纯 HTTP 请求行的代理(常见的入门教程版本)无法承载 HTTPS,因为客户端从未以明文形式发送目标的真实请求。
运行代理服务器本身是合法的;重要的是它的用途以及通过它传递的流量是否被授权——使用自建或第三方代理绕过访问控制、根据网站条款抓取数据或隐藏非法活动均存在法律风险,无论是谁的代码在转发。
问:为什么代理必须特别支持 CONNECT 方法,而不是像处理 HTTP 请求一样转发 HTTPS 请求?
因为 HTTPS 请求在离开客户端之前是加密的,所以处理它的代理以处理纯 HTTP 的方式将需要看到它从未接收到的明文;CONNECT 通过让代理在隧道打开后盲目转发字节来规避这一点,因此 TLS 握手直接在客户端和真实目标之间发生。
问:您自己构建的代理服务器与商业旋转代理服务有何不同?
像本指南中介绍的自托管服务器的出站 IP 数量与其运行的机器数量完全相同——通常是一个——而商业旋转网关则位于一个大型的、托管的 IP 池前面,并在每个请求中更换处理的 IP,这一点在基于 IP 的阻止或速率限制(而非代码复杂性)成为实际约束时很重要。
问:这个代理支持 Python 的 requests 库,还是仅支持 curl?
两者都支持——服务器在协议层级上处理纯 HTTP 代理和 CONNECT 隧道,因此任何正确实现这些的客户端,包括通过其 proxies 参数的 requests、curl -x 和浏览器,都可以无需特殊处理地使用它。
Aug. 6th 2026
立即访问住宅、数据中心、IPv6 与 ISP 高质量代理池。