> For the complete documentation index, see [llms.txt](https://docs.clore.ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.clore.ai/guides/guides_v2-zh/gao-ji/python-sdk.md).

# Python SDK 指南

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

{% hint style="success" %}
**刚接触 SDK？** 先从 [5 分钟快速上手](/guides/guides_v2-zh/ru-men-zhi-nan/python-quickstart.md) 开始。
{% endhint %}

如需带真实示例的快速入门教程，请参阅 [Clore.ai Python SDK — 5 分钟内自动化你的 GPU 工作流](https://blog.clore.ai/cloreai-python-sdk-automate-your-gpu-workflows-in-5-minutes/)

## 安装

```bash
pip install clore-ai
```

该 SDK 提供两个客户端：

* **`CloreAI`** — 同步（更简单，适合脚本）
* **`AsyncCloreAI`** — 异步（更适合并发操作，速度更快）

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

***

## 同步 vs 异步 — 何时使用各自版本

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

### 同步示例

```python
from clore_ai import CloreAI

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

servers = client.marketplace(gpu="RTX 4090")
print(f"找到 {len(servers)} 台服务器")

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

### 异步示例

```python
import asyncio
from clore_ai import AsyncCloreAI

async def main():
    async with AsyncCloreAI() as client:
        servers = await client.marketplace(gpu="RTX 4090")
        print(f"找到 {len(servers)} 台服务器")

asyncio.run(main())
```

### 上下文管理器（推荐）

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

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

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

***

## 客户端配置

```python
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()` 方法会获取所有可用服务器，并在客户端侧进行筛选：

```python
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` 对象进行筛选：

```python
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[0].id} — ${cheapest[0].price_usd:.4f}/小时")
```

### 服务器模型字段

每个 `MarketplaceServer` 对象都有以下属性和便捷属性：

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

***

## 订单管理

### 创建订单

```python
order = client.create_order(
    server_id=142,
    image="cloreai/ubuntu22.04-cuda12",
    type="on-demand",               # "按需" 或 "现货"
    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 笔记本令牌
)

print(f"订单 ID：{order.id}")
print(f"IP：{order.pub_cluster}")
print(f"端口：{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 笔记本令牌            |
| `command`            | `str`   | —  | 启动命令                     |
| `spot_price`         | `float` | —  | 现货出价                     |
| `required_price`     | `float` | —  | 所需价格                     |
| `autossh_entrypoint` | `str`   | —  | 自动 SSH 入口点               |

### 列出订单

```python
# 仅活动订单
active = client.my_orders()
for o in active:
    print(f"订单 {o.id}：type={o.type}，IP={o.pub_cluster}，状态={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`   | 创建时间戳                     |

### 监控订单

```python
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_id} 在 {timeout} 秒后仍未准备好")

# 用法
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)}")
```

### 取消订单

```python
# 可选地附带原因取消
client.cancel_order(order_id=38, issue="任务完成")

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

***

## 服务器管理（适用于主机提供者）

如果你在 Clore 上托管 GPU，SDK 可让你管理你的服务器：

### 列出你的服务器

```python
my_servers = client.my_servers()
for s in my_servers:
    print(f"服务器 {s.id}：{s.gpu_model} — {s.status}")
```

### 获取服务器配置

```python
config = client.server_config("MyGPU-Rig")
print(f"名称：{config.name}")
print(f"可见性：{config.visibility}")
print(f"在线：{config.online}")
print(f"最低租期：{config.mrl} 小时")
print(f"按需价格：{config.on_demand_price}")
print(f"现货价格：{config.spot_price}")
```

### 更新服务器设置

```python
client.set_server_settings(
    name="MyGPU-Rig",
    availability=True,       # 使服务器可用
    mrl=24,                  # 最低租期 24 小时
    on_demand=0.0001,        # BTC 计价的按需价格
    spot=0.00000113          # BTC 计价的现货价格
)
print("设置已更新")
```

***

## 现货市场

如果有人出更高价，现货订单可能会被中断。大约三分之一的服务器将现货价格设定在低于按需价格的水平（中位数约优惠 13%）；其余服务器则将现货价格设为按需价格。

### 查看现货报价

```python
offers = client.spot_marketplace(server_id=6)
for offer in offers:
    print(f"订单 {offer.get('order_id')}：价格={offer.get('price')}")
```

### 创建现货订单

```python
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"已创建现货订单 {order.id}")
```

### 调整现货价格

```python
# 提高出价，避免被别人出更高价压过
client.set_spot_price(order_id=39, price=0.000003)
```

### 现货出价策略

```python
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("没有现货报价——请以按需价格作为基准")
        return None

    min_price = min(o["price"] for o in offers)
    bid = min_price * (1 + premium_pct / 100)
    print(f"市场最低价：{min_price}，出价：{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
    )
```

***

## 钱包操作

### 检查余额

```python
wallets = client.wallets()
for w in wallets:
    print(f"{w.name}：{w.balance:.8f}")
    if w.deposit:
        print(f"  存款地址：{w.deposit}")
```

### 低余额提醒

```python
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"⚠️  BTC 余额过低：{w.balance:.8f}（最低：{min_btc}）")
            return False

    print("✅ 余额正常")
    return True

check_balance(min_btc=0.001)
```

***

## 错误处理最佳实践

### 异常层次结构

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

### 基础错误处理

```python
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("无效的 API 密钥 — 请检查 CLORE_API_KEY")
except InvalidInputError as e:
    print(f"输入错误：{e}")
except RateLimitError:
    print("触发速率限制 — SDK 会自动重试，但已超过最大重试次数")
except CloreAPIError as e:
    print(f"API 错误（代码 {e.code}）：{e}")
```

### 带退避的重试模式

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

```python
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"受到速率限制，{delay} 秒后重试...")
                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. 重用客户端

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

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

### 2. 将异步用于并发操作

```python
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)} 台服务器")
        print(f"RTX 3090: {len(rtx3090)} 台服务器")
        print(f"A100: {len(a100)} 台服务器")

asyncio.run(compare_gpus())
```

### 3. 异步批量创建订单

```python
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"服务器 {sid}：失败 — {result}")
            else:
                print(f"服务器 {sid}：已创建订单 {result.id}")

        return orders

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

{% hint style="warning" %}
**注意：** SDK 强制要求在两次 `create_order` 调用之间有 5 秒冷却时间。即使在异步模式下，订单也会错开以遵守速率限制。
{% endhint %}

### 4. 完成后关闭客户端

```python
# 上下文管理器会自动处理
with CloreAI() as client:
    # work...
    pass  # client.close() 会自动调用

# 或者手动关闭
client = CloreAI()
try:
    # work...
    pass
finally:
    client.close()
```

***

## 完整示例：自动扩缩 GPU 工作器

```python
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_count}/{target_workers}")

        if active_count >= target_workers:
            print("已达到目标。无需操作。")
            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"仅有 {len(candidates)} 台服务器可用（需要 {needed} 台）")

        # 3. 部署
        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"已在服务器 {server.id} 上部署 → 订单 {order.id}")
            except CloreAPIError as e:
                print(f"在 {server.id} 上部署失败：{e}")

asyncio.run(auto_scale())
```

***

## 下一步

* [CLI 自动化](/guides/guides_v2-zh/gao-ji/cli-automation.md) — Bash 脚本、CI/CD、批量操作
* [批量处理](/guides/guides_v2-zh/gao-ji/batch-processing.md) — 在 Clore GPU 上处理大规模工作负载
* [API 集成](/guides/guides_v2-zh/gao-ji/api-integration.md) — 将 AI 服务连接到你的应用


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.clore.ai/guides/guides_v2-zh/gao-ji/python-sdk.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.