Skip to content

API Reference

Jump to bottom
HadesTop edited this page Aug 17, 2026 · 7 revisions

API 参考

客户端

from ipclick import Downloader, ClusterDownloader, create_client

Downloader

Downloader(
    host: str = "127.0.0.1",
    port: int = 9527,
    token: str | None = None,
    ...
)

不传参数时按配置体系解析。用 with 或显式 close()。

create_client()

按 [GENERAL].mode 返回 Downloader 或 ClusterDownloader。推荐用它， 这样代码里不写死单机/集群。

with create_client() as d:
    ...

便捷方法

d.get(url, params=None, **kwargs)
d.post(url, data=None, json=None, **kwargs)
d.put(url, data=None, **kwargs)
d.patch(url, data=None, **kwargs)
d.delete(url, **kwargs)
d.head(url, **kwargs)
d.options(url, **kwargs)

全部返回 DownloadResponse。

request() —— 完整参数

d.request(
    *,
    url: str,                                   # 必填
    method: HttpMethod = HttpMethod.GET,
    adapter: IPClickAdapter | str | None = None,
    headers: dict | None = None,
    cookies: dict | str | None = None,
    params: dict | None = None,
    data: Any = None,                           # bytes / str / dict
    json: dict | None = None,
    proxy: ProxyConfig | str | bool | None = None,
    timeout: float = 60,
    max_retries: int = 3,
    retry_backoff: float = 2.0,
    verify: bool = True,
    allow_redirects: bool = True,
    stream: bool = False,
    impersonate: str | None = None,             # 只有 curl_cffi 支持
    automation_config: str | None = None,
    automation_script: str | None = None,       # 页面内 JS
    allowed_status_codes: list[int] | None = None,
) -> DownloadResponse

几处值得单说的：

  • timeout 是单次尝试的上限，不是总时长。带重试时最坏约 timeout × (max_retries + 1) + 退避总和。
  • verify 默认 True。0.2.0 之前这里默认 None，会被 protobuf 当成"未设置"， 服务端因而收到 False——即默认关闭证书校验。
  • 没有 files 参数。 协议里从来没有这个字段（旧版的 files= 一律抛 NotImplementedError），删掉它只是把 API 说实话。要上传文件请自己拼好 multipart 体， 用 data=<bytes> 加 Content-Type: multipart/form-data; boundary=... 发出去—— data 现在是 bytes 字段，任意二进制都能原样送达。
  • impersonate 只有 curl_cffi 支持。 别的适配器收到会报错而不是忽略—— 静默忽略的话你以为伪装开着，实际裸奔。

stream()

s = d.stream(url, **kwargs) -> StreamedResponse
s.status_code
s.headers
s.content_length
for chunk in s: ...

响应体不进内存。集群里流式请求不转发，由收到请求的节点自己执行。

batch()

d.batch(tasks: Iterable[DownloadTask], timeout: float | None = None) -> Iterator[DownloadResponse]

一次 RPC 发全部，服务端并发执行，按完成顺序返回——必须靠 request_uuid 对应， 不能靠下标。并发度受服务端 max_workers 约束。

异步

from ipclick.aio import AsyncDownloader

async with AsyncDownloader() as d:
    resp = await d.get(url)

接口与同步版一一对应。服务端仍是同步的，异步只在客户端这一侧。

其它

from ipclick import get_downloader, close_all_downloaders, downloader

d = get_downloader()          # 进程内共享实例（按连接参数缓存）
close_all_downloaders()       # 退出前清理
with downloader() as d: ...   # 上下文管理器

数据模型

DownloadResponse

字段 类型 说明
status_code int HTTP 状态码；-1 = 瞬时传输故障
content bytes 响应体原文
text str 解码后的文本
headers dict[str, str] 响应头
url str 最终 URL（跟完重定向）
elapsed_ms int 耗时
error str | None 失败原因
request_uuid str 对应请求的 uuid，批量时靠它配对
adapter_type str 实际用的适配器
trace ResponseTrace 链路信息

ResponseTrace

字段 说明
node_id 集群里实际执行的节点
adapter 实际适配器（browser 已解析成具体引擎）
attempts 内部重试了几次（1 = 一次成功）
forwarded 是否为转发过来的请求
queued_ms 在按 host 限流闸门里排了多久

服务端没带 trace 时是一个全默认值的 ResponseTrace（不是 None）—— 这样 resp.trace.node_id 永远不会 AttributeError。刻意不含任何机密。

DownloadTask

from ipclick import DownloadTask

DownloadTask(uuid="...", url="https://example.com", method=..., adapter=..., ...)

批量用。字段与 request() 的参数一一对应。

枚举

from ipclick import HttpMethod, IPClickAdapter, ProxyConfig

HttpMethod.GET / POST / PUT / PATCH / DELETE / HEAD / OPTIONS
IPClickAdapter.CURL_CFFI / NIQUESTS / BROWSER / CAMOUFOX / PATCHRIGHT / PLAYWRIGHT / DRISSIONPAGE

字符串也认：adapter="niquests" 和 adapter=IPClickAdapter.NIQUESTS 等价。

异常

from ipclick.exceptions import (
    IPClickError, ConfigError, AdapterError, TransportError,
    ClientClosedError, AuthenticationError, RequestError,
    ValidationError, URLNotAllowedError,
)
异常 什么时候 该做什么
IPClickError 所有异常的基类
ValidationError 参数非法（URL 空、协议不允许、适配器名拼错） 改代码
URLNotAllowedError URL 被 SSRF 策略拦下（ValidationError 的子类） 改 URL 或 [SECURITY]
AdapterError 适配器不存在 / 依赖没装 / 浏览器本体未就绪 装依赖或改配置
ConfigError 配置写错（未知引擎名等） 改配置文件
AuthenticationError 令牌不对 对令牌
TransportError 传输层问题 查网络
ClientClosedError 用了已关闭的客户端 改代码
RequestError 请求构造失败 改代码
HostLimitTimeout 按 host 限流等不到额度 放宽限制或降速

ValidationError 同时继承 ValueError，所以 except ValueError 也能接住。

只有瞬时传输故障返回 status_code == -1，其余一律抛异常。 这条分界是刻意的： 把用法错误伪装成网络故障会让人去查网络，而实际要改的是代码或部署。

断点续传

from ipclick.resume import download_to_file

result = download_to_file(d, url, "big.zip")
result.total_bytes
result.resumed          # 是否从断点接着下的

自动处理 Range 与 If-Range。

链路记录

from ipclick.trace import get_recorder

r = get_recorder()
r.recent(limit=50)
r.query(limit=100, status_class="failed", adapter="browser", keyword="example.com")
r.stats(days=7)

详见链路记录。

服务端 API

from ipclick.server import serve

serve(config_path="ipclick.toml", host="0.0.0.0", port=9527)

自定义适配器：

from ipclick.adapters.base import DownloaderAdapter
from ipclick.adapters.registry import register_adapter

class MyAdapter(DownloaderAdapter):
    adapter_name = "my_adapter"
    def download(self, task, **kwargs): ...

register_adapter(MyAdapter)

注册要在服务端进程里做，客户端只是传一个名字。

命令行

ipclick --version
ipclick -e toml                    # 输出 ipclick.toml 模板
ipclick -e env                     # 输出 .env 模板

ipclick init [-f] [-d DIR]         # 生成 ipclick.toml 与 .env
ipclick run [-c FILE] [-p PORT] [--host ADDR] [-v] [-w] [--web-port PORT]
ipclick health [--host H] [-p PORT] [--service NAME] [--timeout S]
ipclick config-info [-c FILE]
命令 干什么
init 生成配置。.env 是 600 权限并自动加进 .gitignore
run 起服务端。-w 同时起 Web 管理端并把登录信息打到控制台；--web-port 覆盖 [WEB].port（0.4，同目录起多实例时必须岔开）
health 查 grpc.health.v1 状态，脚本里做探针用
config-info 显示什么真生效了 —— 排查配置问题的第一步

gRPC 协议

服务名 TaskService，四个方法：

方法 用途
Send 单个请求
SendBatch 批量
SendStream 流式下载（不转发）
Ping 节点探测（0.4 新增）——验连通性与鉴权，不做任何业务动作

Ping 的存在是因为 grpc.health.v1 回答不了"鉴权对不对"：健康检查刻意免鉴权 （编排系统的探针通常拿不到密钥），于是"连不上"和"连上了但令牌不对"在它眼里长得 一模一样，而这两件事的排查方向完全相反。Ping 走正常鉴权通路，能收到响应就说明 这一跳的令牌是对的。

message PingReq  { string from_node = 1; }        // 发起方 id，纯诊断用
message PingResp {
  string node_id = 1;        // 对端自报的节点 id
  string version = 2;        // 对端的 IPClick 版本
  bool   auth_required = 3;  // 对端**是否启用了鉴权**（false = 谁都能调它）
  bool   forward = 4;        // 对端是否开着服务端转发
  int64  uptime_seconds = 5;
  int32  in_flight = 6;      // 对端在途请求数
}

auth_required 单独一位是必要的：探测成功本身分不清"我的令牌对"和"它根本不验"。

对 0.3 的节点调 Ping 会返回 UNIMPLEMENTED——但能走到方法查找这一步，说明鉴权 拦截器已经放行了，所以那恰恰证明令牌是对的。调用方应当据此区分，而不是当成失败。 Web 管理端的「测试连接」就是这么处理的。

.proto 在 src/ipclick/dto/proto/task.proto。 其它语言可以直接用它生成客户端。

几处协议上的约定：

  • data 是 bytes（0.3.0 从 string 改的）。这个方向的改动是 wire 兼容的—— 两者在 protobuf 上编码完全相同。
  • 移除的字段号进 reserved，不复用。旧客户端发来时不会被误解成别的字段。
  • 移除的适配器枚举值标 deprecated 而不是删掉，所以旧客户端传 HTTPX 时能得到 "已移除，请改用 niquests"这样一句有用的话，而不是"未知枚举值"。
  • optional 显式 presence 用来区分"没传"和"传了默认值"——verify 那个 bug 就是这个区别没处理好造成的。

IPClick

开始

用法

能力

运维

参考

Clone this wiki locally