简体中文 | English
一个自带采集、自动验证、按地理位置分池的免费代理池服务。 部署一次,你的应用就有了一个稳定的 HTTP 接口去随用随取代理 IP。
A self-running proxy pool with 22 sources, geo-routed dual pools (CN/intl), score-based eviction, and a one-curl HTTP API. Docker-deploy in one command.
如果你的代码里有这种需求:
# 想给请求换一个 IP,不想自己每天去网上扒
requests.get("https://target.com", proxies={"http": "???", "https": "???"})部署 Xproxypool 之后,这一行就变成:
# 拿一个国内代理(默认池)
proxy = httpx.get("http://localhost:12563/get").json()
url = f"http://{proxy['host']}:{proxy['port']}"
requests.get("https://target.com", proxies={"http": url, "https": url})
# 想要海外代理?加一个参数
proxy = httpx.get("http://localhost:12563/get?pool=intl").json()你得到的是:
- 一个常驻服务,24/7 自己抓代理、自己验证、自己淘汰失效的,不用你管
- 两个独立的代理池,国内池(cn)针对国内站点验证、国际池(intl)针对海外站点验证,各 1000 个起步
- 一组 HTTP API,
GET /get一行 curl 就能拿一个可用代理 - 一个评分系统,优先返回历史响应快、稳定的代理(score 0–100)
- 一份地理位置信息,每个代理标了所在国家/省份/运营商
- 完整的可观测性,
/stats实时看池子健康度,日志详细到每次校验
| 能力 | 说明 |
|---|---|
| 多源采集 | 22 个公开代理源(6 个国内站点 + 16 个海外 GitHub 聚合列表),YAML 配置可热改 |
| 双池架构 | cn 池打 eastmoney.com,intl 池打 api.ipify.org,各管各的目标和容量 |
| 协议全支持 | HTTP / HTTPS / SOCKS4 / SOCKS5 同一套接口拿 |
| 地理分流 | 内置 ip2region.xdb 离线库,新代理按 IP 归属自动入对应池 |
| 三态生命周期 | success → recheck → 永久删除,垃圾代理自动出局 |
| 评分淘汰 | 默认 init=50,通过 +10(上限 100),失败 -30,跌破 0 进入 recheck |
| 容量上限 | cn / intl 各默认 1000,Lua 脚本保证高并发下不超员 |
| 零依赖部署 | 一条 docker compose up -d 起整个栈 |
git clone <repo> && cd Xproxypool
docker compose up -d启动后:
- API 监听
http://localhost:12563 - Redis 暴露在
127.0.0.1:6380(主机端口,容器内仍是 6379)
第一次启动后,后台调度器会立即触发一次采集。约 2–5 分钟后两个池就有数据可用。
# 池子健康度
curl http://localhost:12563/stats
# 拿一个国内 http 代理
curl http://localhost:12563/get?pool=cn
# 拿一个海外 socks5,纯文本格式(可直接喂给 curl --proxy)
curl 'http://localhost:12563/get?pool=intl&protocol=socks5&format=text'仓库根目录提供了两个容器外的测试脚本(只用 Python 标准库,零依赖):
# 冒烟:30 秒搞定,告诉你"通了 / 没通"
python test/smoke.py
# 基准:1-2 分钟,告诉你"真实可用率多少 / 哪个源质量好"
python test/bench.py --n 30详见下文 #测试与质量验证。
服务端口:12563(可通过 .env 的 API_PORT 修改)。
| 参数 | 默认 | 说明 |
|---|---|---|
pool |
cn |
cn 或 intl |
protocol |
任意 | http / https / socks4 / socks5,不传则全协议混合 |
min_score |
50 |
只返回分数 ≥ 此值的代理 |
format |
JSON | 设 text 返回纯文本 protocol://host:port |
返回示例:
{
"host": "106.15.137.41",
"port": 50,
"protocol": "http",
"score": 100,
"ping": 412,
"source": "vmheaven",
"anonymous": true,
"region": "中国|0|上海|上海市|电信",
"created_at": "2026-04-27 02:38:11",
"last_check": "2026-04-27 02:48:09"
}无可用代理时返回 404。
curl 'http://localhost:12563/all?pool=cn&protocol=http'返回数组,适合做批量处理或采样。
{
"cn": {"success": 1000, "recheck": 876, "avg_score": 44.8,
"sources": {"vmheaven": 746, "murongpig": 116, ...}},
"intl": {"success": 1000, "recheck": 14, "avg_score": 97.8,
"sources": {...}}
}curl -X DELETE 'http://localhost:12563/delete?pool=cn&proxy=1.2.3.4:8080'所有测试都在容器外运行——这是有意设计的:消费方就在容器外,测试自然也站在消费方视角,跟业务代码用同一套 HTTP 客户端(urllib + ProxyHandler)。
5 步内告诉你整条链路通不通:
$ python test/smoke.py
API: http://localhost:12563
--- 1. /stats ---
PASS cn pool has at least 1 proxy
PASS intl pool has at least 1 proxy
--- 2. cn pool -> eastmoney.com ---
using http://106.15.137.41:50 score=100
PASS HTTP 200
PASS response contains eastmoney marker
--- 3. intl pool -> api.ipify.org ---
using http://206.238.239.5:80 score=100
PASS HTTP 200
PASS response is an IPv4 address
ALL PASS退出码 0/1,可以直接接到 CI 或 cron 监控。
随机抽 N 个代理,真去打目标站点,给你一份消费视角的质量报告:
$ python test/bench.py --n 30 --workers 16
=== pool=cn ===
sampled 30 of 1000 http proxies; target http://www.eastmoney.com
success: 18/30 (60.0%)
ping ms: median=1812 p95=2266 min=411 max=2845
failures:
6 URL Tunnel connection failed: 400 Bad Request
4 TimeoutError
2 URL Tunnel connection failed: 405 Method Not Allowed
by source:
vmheaven 14/16 ( 87.5%)
openproxylist 3/8 ( 37.5%)
89ip 0/4 ( 0.0%) ← 这个源该禁用了
thespeedx 1/2 ( 50.0%)
=== pool=intl ===
success: 29/30 (96.7%)
ping ms: median=441 p95=712 min=287 max=890
by source:
murongpig 22/22 (100.0%)
proxyscrape 4/4 (100.0%)
...bench 报告的"真实成功率"通常会低于 /stats 的 success 数,因为 bench 用 urllib 走 HTTPS CONNECT 隧道(消费方真实路径),容器内 validator 走 HTTP 直连(更宽松)。这个 gap 就是 bench 存在的意义。
| 文件 | 用途 | 是否需要部署 |
|---|---|---|
test/smoke.py |
部署后一键确认服务可用 | 是 |
test/bench.py |
量化代理池真实质量,按 source 拆分 | 是 |
src/proxy_pool/tests/run_tests.py |
容器内自检(55 项,代码层面) | 否(独立) |
最常需要调的几项,改 .env 后 docker compose restart 即可:
| 变量 | 默认 | 改它的理由 |
|---|---|---|
CN_PROXY_CAP / INTL_PROXY_CAP |
1000 |
池子越大覆盖越广,但每轮校验越慢。100–5000 都合理。 |
FETCH_INTERVAL |
20(分钟) |
池满会自动跳过 fetch,所以这个值小于 validate 没用 |
VALIDATE_INTERVAL |
10 |
越短代理越新鲜,但消耗目标站点请求更多 |
SCORE_INIT / SCORE_DECREMENT |
50 / 30 |
必须满足 INIT >= DECREMENT,否则代理一次失败就死(项目早期踩过这个坑) |
VALIDATOR_CONCURRENCY |
50 |
越大校验越快,但宿主机 socket 压力大 |
采集源开关在 sources.yml,把不要的源 enabled: false 即可。bench 报告里 0% 成功率的源就该关掉。
每 20 分钟
│
┌────────────────┴────────────────┐
│ 采集器(22 个源,并发抓取) │
└────────────────┬────────────────┘
│ 去重 + 跳过已知 IP
▼
┌──────────────┐
│ ip2region │ ← 离线 IP 归属库
│ 按国别分流 │
└─┬──────────┬─┘
中国 IP │ │ 海外 IP
▼ ▼
打 eastmoney 打 ipify
│ │
通过则入 cn / intl 池(success)
│ │
每 10 分钟 │ │
重测扣分 ▼ ▼
score ≤ 0 移入 recheck
│ │
每 60 分钟 │ │
给一次复活│ │
通过 → 回 success(若未满)
失败 → 永久删除
Xproxypool/
├── src/ 主项目代码
│ ├── main.py FastAPI 入口
│ ├── proxy_pool/
│ │ ├── api/ HTTP 接口
│ │ ├── core/ 采集调度 / 校验 / Redis 存储
│ │ ├── fetchers/ 22 个数据源实现
│ │ ├── schemas/ Proxy 模型
│ │ ├── tests/run_tests.py 容器内自检
│ │ └── utils/ 配置 / IP 归属 / 日志
│ └── ip2region/ 离线 IP 归属库
│
├── test/ 容器外集成测试
│ ├── smoke.py 冒烟脚本
│ └── bench.py 真实可用率基准
│
├── sources.yml 22 个源的 URL 配置
├── ip2region.xdb 离线 IP 归属数据库
├── pyproject.toml Python 依赖
├── Dockerfile / docker-compose.yml 部署
└── .env 运行时配置
- 爬虫 / 数据采集:批量请求
/get拿不同 IP,绕过单 IP 的 QPS 限制和封禁 - 多账号批量操作:每个账号绑定一个 score 高的 sticky proxy
- 跨境访问测试:
/get?pool=intl拿海外 IP 验证境外可达性 - 网络环境多样性:测试 CDN 在不同省份/运营商下的表现(代理 region 字段精确到运营商)
- CI 集成:
smoke.py接入流水线,代理池失效时自动告警
项目所有扩展点都在 src/ 下,改完 docker compose restart myproxypool 即可生效(src/ 是只读挂载,不需要重新 build 镜像)。
三步:写抓取类 → 注册 → 配置 URL。
第一步:在 src/proxy_pool/fetchers/domestic.py 或 international.py 加一个继承 BaseFetcher 的类。get_html / get_json 已经处理好了直连/代理回退、超时、编码、socket 关闭,你只管解析:
class MyFetcher(BaseFetcher):
source = "myfetcher" # 唯一标识,会落到 Proxy.source 字段
async def fetch(self) -> list[Proxy]:
cfg = get("international", "myfetcher") # 读 sources.yml
if not cfg.get("enabled", True):
return []
# 三种典型的解析模式,按需挑一种:
# A) 站点是 HTML 表格(IP / Port 各一列)
html = await self.get_html(cfg["url"])
return self._parse_table(html, protocol="http")
# B) 站点是纯文本 ip:port 列表(GitHub raw 文件常见)
text = await self.get_html(cfg["url"])
return self._parse_text(text, protocol="socks5")
# C) 站点是 JSON API,自己挑字段
data = await self.get_json(cfg["url"])
return [
Proxy(host=item["ip"], port=item["port"],
protocol=item.get("type", "http"), source=self.source)
for item in data["proxies"]
]第二步:在 src/proxy_pool/fetchers/__init__.py 的 ALL_FETCHERS 列表里加一行,这个类才会被调度器扫到:
from proxy_pool.fetchers.international import MyFetcher # 加 import
ALL_FETCHERS = [
...
MyFetcher, # 加到列表末尾
]第三步:在仓库根的 sources.yml 加一段(URL、分页、是否启用):
international:
myfetcher:
enabled: true
url: "https://example.com/api/proxies?type=all"docker compose restart myproxypool → 下次 fetch_job(20 分钟内)就会跑你的 fetcher。想立刻看效果,看日志找 MyFetcher fetched N proxies。
src/proxy_pool/core/validator.py 顶部:
CN_TARGET = ("http://www.eastmoney.com", _verify_eastmoney)
INTL_TARGET = ("http://api.ipify.org", _verify_ipify)想换目标(比如你只关心代理对自家业务站点的可达性),改 URL + 写一个新的 _verify_xxx 函数即可:
async def _verify_mybusiness(resp: aiohttp.ClientResponse) -> bool:
if resp.status != 200:
return False
body = await resp.text()
return "expected-marker-string" in body
CN_TARGET = ("http://your-business-cn.com/healthz", _verify_mybusiness)注意:目标必须是 HTTP(非 HTTPS),否则 SOCKS / HTTP 代理的 CONNECT 路径不一致会让 SOCKS 代理大量误判失败。如果一定要测 HTTPS,要分别给 HTTP 代理和 SOCKS 代理写不同的请求路径。
src/proxy_pool/api/routes.py,跟现有端点同样的写法。例如加一个"按 region 关键字过滤":
@router.get("/get_by_region")
async def get_by_region(
keyword: str = Query(..., description="region 字段子串,如 '上海' 或 'United States'"),
pool: PoolName = Query("cn"),
):
proxies = await _get_pool(pool).get_success_list()
matched = [p for p in proxies if keyword in p.region]
if not matched:
raise HTTPException(status_code=404, detail=f"No proxy in region containing {keyword!r}")
import random
return random.choice(matched).model_dump()docker compose restart 后 curl 'http://localhost:12563/get_by_region?keyword=上海' 即可。
最常调的两组,改 .env 即可:
# 池子小、想快速迭代
CN_PROXY_CAP=200
INTL_PROXY_CAP=200
VALIDATE_INTERVAL=5 # 5 分钟一轮校验,代理更新鲜
# 池子大、想覆盖广
CN_PROXY_CAP=5000
INTL_PROXY_CAP=5000
VALIDATOR_CONCURRENCY=200 # 并发拉高,否则一轮校验跑不完/stats 已经是结构化 JSON,挂到 Prometheus 最简方式是写个 sidecar 每分钟拉一次,转 metrics。也可以直接 cron:
*/1 * * * * curl -s http://localhost:12563/stats | jq -c \
'{ts: now, cn_success: .cn.success, cn_avg: .cn.avg_score, \
intl_success: .intl.success, intl_avg: .intl.avg_score}' \
>> /var/log/xproxypool-stats.jsonl定期跑 python test/bench.py --n 100,看 by source 那块。某个 source 长期在 0–20% 区间的,直接去 sources.yml 把它 enabled: false。这是最有效的提质方式——不收垃圾比再写新 fetcher 重要得多。
仓库里附带了一个 MCP 服务器(mcp_server/),把 Redis 里的代理状态作为只读工具暴露给任何 MCP 兼容的 LLM 客户端——Claude Desktop / Claude Code / Cursor 等。
挂载之后,你就能直接问:
- "现在池子状况怎么样?" → 模型调
pool_stats,自然语言总结 - "为什么 cn 池 avg_score 这么低?" →
count_by_source+list_recheck综合分析,告诉你哪个 source 该禁 - "给我一个上海电信的 socks5" →
list_proxies(protocol="socks5", region_contains="上海电信")
7 个只读工具,不暴露任何写操作(防 LLM 误删)。详细配置见 mcp_server/README.md。
如果你想给团队其他成员一个更傻瓜的接口,在 test/ 目录下加一个 client.py,封装 /get + 自动重试:
# test/client.py
import urllib.request, json, random
class XproxyClient:
def __init__(self, api="http://localhost:12563"): self.api = api
def get(self, pool="cn", protocol="http", retries=3):
for _ in range(retries):
try:
r = urllib.request.urlopen(f"{self.api}/get?pool={pool}&protocol={protocol}", timeout=5)
return json.loads(r.read())
except Exception:
continue
return None
# 业务方调用
client = XproxyClient()
proxy = client.get(pool="intl")- IP 归属地数据:lionsoul2014/ip2region
- 22 个公开代理源的维护者(详见
sources.yml)
MIT License — 自由使用、修改、商用,无需署名(但欢迎)。