Skip to content

API Reference

Jump to bottom
Hades edited this page Aug 23, 2026 · 7 revisions

API 参考

客户端

from ipclick import Downloader, create_client, get_downloader, downloader

ClusterDownloader 不在顶层导出，要直接用它得从子模块导入 （多数情况下用 create_client() 就够了，见下）：

from ipclick.cluster.client import ClusterDownloader

Downloader

Downloader(
    config_path: str | None = None,
    host: str | None = None,
    port: int | None = None,
    token: str | None = None,
    tls: TLSSettings | None = None,
)

每个参数默认都是 None，意思是"这一项去配置里取"，而不是"用某个硬编码值"。 都不传就完全按配置体系解析（最终默认是 127.0.0.1:9528）。 用 with 或显式 close() 关闭。

TLSSettings 不在顶层导出里，要 from ipclick.tls import TLSSettings。 多数时候不用手搓——客户端会自己从 [SECURITY] 配置里解析。

另外两个拿客户端的方式——进程级单例 get_downloader() 和模块级懒加载对象 downloader——见 SDK 用法。

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) + 退避总和。服务端会把它拆成（连接, 读取）两段交给 curl_cffi，两者之和仍等于这个值，连接段的上限由 [DOWNLOADER].connect_timeout 决定。
  • automation_config / automation_script 只有浏览器适配器实现。传给 curl_cffi / niquests 会抛 ValidationError——它们不执行 JavaScript。 报错而不是静默忽略是刻意的：静默忽略的话请求照发、返回一段没跑过脚本的原始 HTML， 调用方从结果里看不出参数根本没生效。
  • 没有 files 参数，gRPC 协议里也没有这个字段；进程内直接传给 curl_cffi 会抛 ValidationError。要上传文件请自己拼好 multipart 体，用 data=<bytes> 加 Content-Type: multipart/form-data; boundary=... 发出去——data 是 bytes 字段， 任意二进制都能原样送达。
  • verify 默认 True，即默认校验证书。协议里它是显式 presence 的 optional bool，所以"显式传 False"和"没传"分得开，服务端对"没传"按 True 处理。
  • impersonate 只有 curl_cffi 支持。 别的适配器收到会报错而不是忽略—— 静默忽略的话你以为伪装开着，实际裸奔。用 curl_cffi 而不传时，SDK 会自动填 impersonate="chrome"。
  • allowed_status_codes 是重试白名单，不是成功判定。 不传时会被填成 [200, 404]。它唯一的作用是把列进来的状态码从重试名单里摘掉，所以只对本来会重试 的码（默认 429/500/502/503/504）有意义；列上 404 是空操作，因为 404 从来就不重试。 摘掉之后 resp.ok 也不会变 True——那个仍然严格只认 2xx。

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 约束。

request_uuid 不会自动帮你配对：DownloadTask.uuid 默认是空字符串，真正的 uuid 只在发送时生成，不会回写到你手上的对象。要用它对应结果，就必须在构造每个 DownloadTask 时自己显式传 uuid=（例如 str(uuid_utils.uuid7())）。

异步

from ipclick.aio import AsyncDownloader

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

接口与同步版一一对应。服务端默认是同步的（一请求一线程）， [SERVER].async_mode = true 可以切到 grpc.aio 协程模式（实验性，默认关）。 客户端用不用协程、服务端跑哪种并发模型，是两件互不相干的事。

其它

from ipclick import get_downloader, close_all_downloaders, downloader

d = get_downloader()          # 进程内共享实例（按连接参数缓存）
close_all_downloaders()       # 退出前清理

downloader.get(url)           # 模块级懒实例，首次用到时才建连
with downloader as d: ...     # 它自己就是上下文管理器

downloader 是一个实例而不是工厂函数——downloader() 会抛 TypeError。 省掉括号是刻意的：脚本里 from ipclick import downloader 之后直接 downloader.get(url) 就能发请求，不用先想清楚连接参数。

数据模型

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 / TRACE
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 也能接住。

客户端拦的和服务端拦的不是一回事。 客户端只看 URL 前缀（不是 http:// / https:// 就抛 ValidationError），协议白名单、内网地址、云元数据地址这些策略全在服务端—— 所以 d.get("file:///etc/passwd") 抛的是 ValidationError，而 d.get("http://169.254.169.254/") 要等服务端回话才抛 URLNotAllowedError。

只有瞬时传输故障返回 status_code == -1，其余一律抛异常。 这条分界是刻意的： 把用法错误伪装成网络故障会让人去查网络，而实际要改的是代码或部署。被 SSRF 策略拒绝 尤其容易误判——它和"目标站点连不上"的排查方向完全相反（一个改 [SECURITY]，一个查网络）， 所以服务端的 PERMISSION_DENIED 会在客户端还原成 URLNotAllowedError 抛出来， 而不是混进那些返回 -1 的传输故障里。

断点续传

from ipclick.resume import download_to_file

result = download_to_file(d, url, "big.zip")
result.total_bytes
result.attempts         # 总共发了几次传输
result.restarts         # 其中有几次是从头重下
result.resumed          # attempts > 1，即"断过"；不代表断点续上了
result.status_code
result.path

resumed 只说明传输不止一次。真正想知道"断点接上了"要看 attempts > 1 and restarts == 0——服务器不支持 Range 时会从头重下， 那种情况 resumed 同样是 True，但一个字节都没省下。

自动处理 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=9528)

自定义适配器：

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

from ipclick.dto.response import Response

class MyAdapter(DownloaderAdapter):
    adapter_name = "my_adapter"

    def download(self, url: str, **kwargs) -> Response: ...

register_adapter(MyAdapter)

第一个位置参数是 URL 字符串，不是 DownloadTask——本项目里 task 是另一个东西。

注册只影响服务端进程内的注册表。但 gRPC 上 adapter 字段是一个封闭的 protobuf 枚举，没有字符串逃生口，所以 adapter="my_adapter" 这个名字 客户端根本发不出去（SDK 在构造请求时就会抛"未知的适配器名称"，Web 端的 「试一试」走同一条路）。自定义适配器目前只有两条路能真正跑起来：在服务端进程内 直接调用它，或者让它顶替一个已有枚举名（adapter_name = "niquests"）。 要一个全新的名字，得改 dto/proto/task.proto 的 AdapterType 再重新生成代码。

命令行

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

ipclick init [-f] [-d DIR] [-p PORT]   # 生成 ipclick.toml 与 .env
ipclick run [-c FILE] [-p PORT] [--host ADDR] [-v] [-w] [--web-port PORT] [--web-host ADDR] [--web-lan]
ipclick health [-c FILE] [--host H] [-p PORT] [--service NAME] [--timeout S]
ipclick config-info [-c FILE]

上面四条是部署用的。另有一组给脚本和 AI 调用的命令——fetch、status、trace、 node、component、config、skill，全部支持 -J/--json。 完整清单、每个选项、--json 输出契约与退出码都在命令行。

命令 干什么
init 生成配置。.env 是 600 权限并自动加进 .gitignore
run 起服务端。-w 同时起 Web 管理端并把登录信息打到控制台；--web-port 覆盖 [WEB].port（默认 9527）；同目录起多实例时必须岔开，否则第二个起不来
health 查 grpc.health.v1 状态，脚本里做探针用
config-info 显示什么真生效了 —— 排查配置问题的第一步

gRPC 协议

服务名 TaskService，五个方法：

方法 用途
Send 单个请求
SendBatch 批量
SendStream 流式下载（不转发）
Ping 节点探测——验连通性与鉴权，不做任何业务动作
Component 远程管理这台节点上的可选组件（装 / 卸 / 下浏览器本体 / 查状态）。默认关，要被操作的那台自己打开 [CLUSTER].allow_remote_install，关着时返回 PERMISSION_DENIED 并说清要改哪一项

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 单独一位是必要的：探测成功本身分不清"我的令牌对"和"它根本不验"。

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

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

几处协议上的约定：

  • data 是 bytes 而不是 string。两者在 protobuf 上编码完全相同，所以这两个 类型之间的改动是 wire 兼容的。
  • 移除的字段号进 reserved，不复用。旧客户端发来时不会被误解成别的字段。
  • 移除的适配器枚举值标 deprecated 而不是删掉，所以旧客户端传 HTTPX 时能得到 "已移除，请改用 niquests"这样一句有用的话，而不是"未知枚举值"。
  • optional 显式 presence 用来区分"没传"和"传了默认值"。verify_ssl、 allow_redirects、stream 这类布尔字段少了它就分不清"显式传 false"和"没传"， 服务端只能把两者一律当成 false；有了 presence，服务端才能只对"没传"套用自己的 默认值（verify_ssl 未设置时按 true 处理）。

下一步

IPClick

开始

用法

能力

运维

参考

Clone this wiki locally