Python SDK 指南

完整的 Python SDK 指南——同步/异步客户端、市场过滤、订单生命周期、现货市场、钱包操作与错误处理

第一次使用 SDK？ 从 5 分钟快速入门 开始。

有关带有真实示例的快速入门教程，请参见 Clore.ai Python SDK — 在 5 分钟内自动化您的 GPU 工作流程

安装

pip install clore-ai

该 SDK 提供两个客户端：

• CloreAI — 同步（更简单，适合脚本）

• AsyncCloreAI — 异步（并发操作更快）

两者共享相同的方法并返回相同的 Pydantic 模型。

---

同步 vs 异步 — 何时使用各自方式

使用场景	客户端	原因
简单脚本，一次性任务	CloreAI	代码更简单，无需 async/await
监控循环	CloreAI	顺序检查即可
批量市场查询	AsyncCloreAI	并发请求 = 更快
批量订单创建	AsyncCloreAI	并行创建多个订单
Web 应用	AsyncCloreAI	非阻塞 I/O

同步示例

from clore_ai import CloreAI

client = CloreAI()  # 使用 CLORE_API_KEY 环境变量

servers = client.marketplace(gpu="RTX 4090")
print(f"Found {len(servers)} servers")

client.close()  # 或使用上下文管理器

异步示例

import asyncio
from clore_ai import AsyncCloreAI

async def main():
    async with AsyncCloreAI() as client:
        servers = await client.marketplace(gpu="RTX 4090")
        print(f"Found {len(servers)} servers")

asyncio.run(main())

上下文管理器（推荐）

两个客户端都支持上下文管理器以便自动清理：

# 同步
with CloreAI() as client:
    wallets = client.wallets()

# 异步
async with AsyncCloreAI() as client:
    wallets = await client.wallets()

---

客户端配置

client = CloreAI(
    api_key="your_key",      # 或设置 CLORE_API_KEY 环境变量
    base_url="https://api.clore.ai/v1",  # 自定义 API 端点
    timeout=30.0,            # 请求超时（秒）
    max_retries=3            # 在速率限制/网络错误时重试次数
)

SDK 包含内置的速率限制器：

• 常规请求： 1 次请求/秒

• create_order: 调用之间 5 秒冷却时间

• 速率限制错误（代码 5）： 自动指数退避

---

市场过滤

该 marketplace() 方法获取所有可用服务器并在客户端进行过滤：

from clore_ai import CloreAI

client = CloreAI()

# 所有可用服务器
all_servers = client.marketplace()

# 按 GPU 型号过滤（大小写不敏感的子字符串匹配）
rtx_4090s = client.marketplace(gpu="RTX 4090")

# 按多个条件过滤
budget_gpus = client.marketplace(
    gpu="RTX 4090",
    max_price_usd=1.0,       # 最高 $1.00/小时
    min_gpu_count=2,          # 至少 2 块 GPU
    min_ram_gb=64.0,          # 至少 64GB 系统内存
    available_only=True       # 仅可用服务器（默认）
)

高级过滤（客户端侧）

对于方法未内置的过滤器，请对返回的 Server 对象自行过滤：

servers = client.marketplace(gpu="RTX 4090")

# 位于欧盟且可靠性高的服务器
eu_servers = [
    s for s in servers
    if s.location and s.location.upper() in ("DE", "FR", "NL", "FI")
    and s.reliability and s.reliability >= 0.95
]

# 按价格排序
cheapest = sorted(servers, key=lambda s: s.price_usd or float("inf"))
print(f"Cheapest: Server {cheapest[0].id} — ${cheapest[0].price_usd:.4f}/h")

服务器模型字段

每个 MarketplaceServer 对象具有这些属性和便捷属性：

字段	类型	描述
id	int	服务器 ID（在 create_order)
gpu_model	str | None	来自规格的 GPU 描述（属性）
gpu_count	int	GPU 的数量，来自 gpu_array （属性）
ram_gb	float | None	系统内存（GB）（属性，来自 specs.ram)
price_usd	float | None	按需价格（美元）（属性，来自 price.usd.on_demand_usd)
spot_price_usd	float | None	现货价格（美元）（属性）
available	bool	服务器是否未被租用（属性）
location	str | None	来自网络规格的国家代码（属性）
specs	ServerSpecs | None	硬件规格（cpu、内存、磁盘、GPU、网络）
price	ServerPrice | None	完整定价结构
rented	bool | None	服务器当前是否被租用

---

订单管理

创建订单

order = client.create_order(
    server_id=142,
    image="cloreai/ubuntu22.04-cuda12",
    type="on-demand",               # "on-demand" 或 "spot"
    currency="bitcoin",             # 支付货币
    ssh_password="MySecurePass",    # SSH 访问
    ports={"22": "tcp", "8888": "http"},  # 端口映射
    env={"HF_TOKEN": "hf_xxx"},    # 环境变量
    command="bash /start.sh",       # 自定义启动命令
    jupyter_token="my_token"        # Jupyter notebook 令牌
)

print(f"Order ID: {order.id}")
print(f"IP: {order.pub_cluster}")
print(f"Ports: {order.tcp_ports}")

完整 create_order 参数

参数	类型	必需	描述
server_id	int	✅	要租用的服务器
image	str	✅	Docker 镜像
type	str	✅	"on-demand" 或 "spot"
currency	str	✅	支付货币（例如， "bitcoin")
ssh_password	str	—	SSH 密码
ssh_key	str	—	SSH 公钥
ports	dict	—	端口映射（{"22": "tcp"})
env	dict	—	环境变量
jupyter_token	str	—	Jupyter notebook 令牌
command	str	—	启动命令
spot_price	float	—	现货出价价格
required_price	float	—	所需价格
autossh_entrypoint	str	—	自动 SSH 入口点

列出订单

# 仅活动订单
active = client.my_orders()
for o in active:
    print(f"Order {o.id}: type={o.type}, IP={o.pub_cluster}, status={o.status}")

# 包括已完成订单
all_orders = client.my_orders(include_completed=True)

订单模型字段

字段	类型	描述
id	int	订单 ID
server_id	int | None	被租用的服务器 ID
type	str	"on-demand" 或 "spot"
status	str | None	订单状态
image	str | None	Docker 镜像
currency	str | None	支付货币
price	float | None	价格
pub_cluster	str | None	公网 IP / 主机名
tcp_ports	dict | None	端口映射（例如， {"22": 50022})
created_at	str | None	创建时间戳

监控订单

import time

def wait_for_ready(client, order_id, timeout=120):
    """等待订单获得公网 IP。"""
    for _ in range(timeout // 10):
        orders = client.my_orders()
        order = next((o for o in orders if o.id == order_id), None)
        if order and order.pub_cluster:
            return order
        time.sleep(10)
    raise TimeoutError(f"Order {order_id} not ready after {timeout}s")

# 用法
order = client.create_order(server_id=142, image="cloreai/ubuntu22.04-cuda12", type="on-demand", currency="bitcoin")
ready = wait_for_ready(client, order.id)
print(f"SSH: ssh root@{ready.pub_cluster} -p {ready.tcp_ports.get('22', 22)}")

取消订单

# 带可选原因取消
client.cancel_order(order_id=38, issue="Job complete")

# 取消所有活动订单
orders = client.my_orders()
for order in orders:
    client.cancel_order(order.id, issue="Cleanup")
    print(f"Cancelled order {order.id}")

---

服务器管理（针对主机方）

如果您在 Clore 上托管 GPU，SDK 允许您管理您的服务器：

列出您的服务器

my_servers = client.my_servers()
for s in my_servers:
    print(f"Server {s.id}: {s.gpu_model} — {s.status}")

获取服务器配置

config = client.server_config("MyGPU-Rig")
print(f"Name: {config.name}")
print(f"Visibility: {config.visibility}")
print(f"Online: {config.online}")
print(f"Min rental: {config.mrl}h")
print(f"On-demand price: {config.on_demand_price}")
print(f"Spot price: {config.spot_price}")

更新服务器设置

client.set_server_settings(
    name="MyGPU-Rig",
    availability=True,       # 使服务器可用
    mrl=24,                  # 最低 24 小时租期
    on_demand=0.0001,        # 按需价格（BTC）
    spot=0.00000113          # 现货价格（BTC）
)
print("Settings updated")

---

现货市场

现货订单便宜 30–50%，但如果有人出价更高，可能会被中断。

查看现货报价

offers = client.spot_marketplace(server_id=6)
for offer in offers:
    print(f"Order {offer.get('order_id')}: price={offer.get('price')}")

创建现货订单

order = client.create_order(
    server_id=142,
    image="cloreai/ubuntu22.04-cuda12",
    type="spot",
    currency="bitcoin",
    spot_price=0.0001,       # 您的出价
    ssh_password="MyPass"
)
print(f"Spot order {order.id} created")

调整现货价格

# 提高您的出价以避免被超出
client.set_spot_price(order_id=39, price=0.000003)

现货竞价策略

from clore_ai import CloreAI

client = CloreAI()

def smart_spot_bid(server_id, premium_pct=5):
    """以略高于当前最低现货价的价格出价。"""
    offers = client.spot_marketplace(server_id=server_id)
    if not offers:
        print("No spot offers — use on-demand price as baseline")
        return None

    min_price = min(o["price"] for o in offers)
    bid = min_price * (1 + premium_pct / 100)
    print(f"Market min: {min_price}, bidding: {bid:.8f} (+{premium_pct}%)")
    return bid

# 用法
bid = smart_spot_bid(server_id=142, premium_pct=10)
if bid:
    order = client.create_order(
        server_id=142,
        image="cloreai/ubuntu22.04-cuda12",
        type="spot",
        currency="bitcoin",
        spot_price=bid
    )

---

钱包操作

检查余额

wallets = client.wallets()
for w in wallets:
    print(f"{w.name}: {w.balance:.8f}")
    if w.deposit:
        print(f"  Deposit address: {w.deposit}")

低余额提醒

from clore_ai import CloreAI

def check_balance(min_btc=0.001):
    """如果 BTC 余额低于阈值则发出提醒。"""
    client = CloreAI()
    wallets = client.wallets()

    for w in wallets:
        if w.name.lower() == "bitcoin" and w.balance < min_btc:
            print(f"⚠️  Low BTC balance: {w.balance:.8f} (minimum: {min_btc})")
            return False

    print("✅ Balances OK")
    return True

check_balance(min_btc=0.001)

---

错误处理最佳实践

异常层次结构

CloreAPIError（基类）
├── DBError           (代码 1) — 数据库错误
├── InvalidInputError (代码 2) — 错误的输入
├── AuthError         (代码 3) — 无效的 API 密钥
├── InvalidEndpointError (代码 4) — 错误的端点
├── RateLimitError    (代码 5) — 被限流（自动重试）
└── FieldError        (代码 6) — 字段特定错误

基础错误处理

from clore_ai import CloreAI
from clore_ai.exceptions import (
    CloreAPIError,
    AuthError,
    RateLimitError,
    InvalidInputError
)

client = CloreAI()

try:
    order = client.create_order(
        server_id=999999,
        image="cloreai/ubuntu22.04-cuda12",
        type="on-demand",
        currency="bitcoin"
    )
except AuthError:
    print("Invalid API key — check CLORE_API_KEY")
except InvalidInputError as e:
    print(f"Bad input: {e}")
except RateLimitError:
    print("Rate limited — the SDK retries automatically, but max retries exceeded")
except CloreAPIError as e:
    print(f"API error (code {e.code}): {e}")

带退避的重试模式

SDK 对速率限制和网络错误具有内置重试（max_retries=3）。对于应用层的重试：

import time
from clore_ai import CloreAI
from clore_ai.exceptions import CloreAPIError, RateLimitError

def retry_operation(func, max_attempts=3, base_delay=2.0):
    """使用指数退避重试 Clore API 操作。"""
    for attempt in range(max_attempts):
        try:
            return func()
        except RateLimitError:
            if attempt < max_attempts - 1:
                delay = base_delay * (2 ** attempt)
                print(f"Rate limited, retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise
        except CloreAPIError as e:
            if e.code in (1,):  # 数据库错误可能是暂时性的
                if attempt < max_attempts - 1:
                    time.sleep(base_delay)
                    continue
            raise

# 用法
client = CloreAI()
servers = retry_operation(lambda: client.marketplace(gpu="RTX 4090"))

---

性能提示

1. 重用客户端

# ❌ 错误 — 每次都会创建新的 HTTP 连接
for _ in range(10):
    client = CloreAI()
    client.marketplace()
    client.close()

# ✅ 正确 — 重用 HTTP 连接
client = CloreAI()
for _ in range(10):
    client.marketplace()
client.close()

2. 对于并发操作使用异步

import asyncio
from clore_ai import AsyncCloreAI

async def compare_gpus():
    async with AsyncCloreAI() as client:
        # 并行运行 3 个搜索
        rtx4090, rtx3090, a100 = await asyncio.gather(
            client.marketplace(gpu="RTX 4090"),
            client.marketplace(gpu="RTX 3090"),
            client.marketplace(gpu="A100"),
        )

        print(f"RTX 4090: {len(rtx4090)} servers")
        print(f"RTX 3090: {len(rtx3090)} servers")
        print(f"A100: {len(a100)} servers")

asyncio.run(compare_gpus())

3. 异步批量创建订单

import asyncio
from clore_ai import AsyncCloreAI

async def batch_deploy(server_ids):
    async with AsyncCloreAI() as client:
        tasks = [
            client.create_order(
                server_id=sid,
                image="cloreai/ubuntu22.04-cuda12",
                type="on-demand",
                currency="bitcoin",
                ssh_password="BatchPass123",
                ports={"22": "tcp"}
            )
            for sid in server_ids
        ]
        orders = await asyncio.gather(*tasks, return_exceptions=True)

        for sid, result in zip(server_ids, orders):
            if isinstance(result, Exception):
                print(f"Server {sid}: FAILED — {result}")
            else:
                print(f"Server {sid}: Order {result.id} created")

        return orders

# 同时在 3 台服务器上部署
asyncio.run(batch_deploy([142, 305, 891]))

注意： SDK 在调用之间强制执行 5 秒的冷却时间。 create_order 即使在异步模式下，订单也会被间隔开以遵守速率限制。

4. 完成后关闭客户端

# 上下文管理器会自动处理此项
with CloreAI() as client:
    # 工作...
    pass  # client.close() 会被自动调用

# 或手动关闭
client = CloreAI()
try:
    # 工作...
    pass
finally:
    client.close()

---

完整示例：自动扩缩 GPU 工作节点

import asyncio
import time
from clore_ai import AsyncCloreAI
from clore_ai.exceptions import CloreAPIError

async def auto_scale(
    gpu_model="RTX 4090",
    max_price=2.0,
    target_workers=3,
    image="cloreai/ubuntu22.04-cuda12"
):
    """维持一个 GPU 工作节点池。"""
    async with AsyncCloreAI() as client:
        # 1. 检查当前订单
        current_orders = await client.my_orders()
        active_count = len(current_orders)
        print(f"Active workers: {active_count}/{target_workers}")

        if active_count >= target_workers:
            print("Already at target. Nothing to do.")
            return

        # 2. 查找可用服务器
        servers = await client.marketplace(gpu=gpu_model, max_price_usd=max_price)
        servers.sort(key=lambda s: s.price_usd or float("inf"))

        needed = target_workers - active_count
        candidates = servers[:needed]

        if len(candidates) < needed:
            print(f"Only {len(candidates)} servers available (need {needed})")

        # 3. Deploy
        for server in candidates:
            try:
                order = await client.create_order(
                    server_id=server.id,
                    image=image,
                    type="on-demand",
                    currency="bitcoin",
                    ssh_password="WorkerPass123",
                    ports={"22": "tcp"}
                )
                print(f"Deployed on server {server.id} → order {order.id}")
            except CloreAPIError as e:
                print(f"Failed to deploy on {server.id}: {e}")

asyncio.run(auto_scale())

---

下一步

• 命令行自动化 — Bash 脚本、CI/CD、批处理操作

• 批量处理 — 在 Clore GPU 上处理大规模工作负载

• API 集成 — 将 AI 服务连接到您的应用