ragent-oss Python SDK

ragent-oss 存储网关的 Python 客户端库。异步接口，基于 httpx。

安装

pip install ragent-oss

快速开始

from ragent_oss import RagentOssClient, PresignRequest, SignRequest, DeleteRequest

async with RagentOssClient("https://oss.ragents.net", "your-api-key") as client:
    # 获取预签名上传 URL
    presign = await client.presign(
        PresignRequest(
            filename="report.pdf",
            content_type="application/pdf",
            category="knowledge",
        )
    )
    # presign.object_key, presign.upload_url, presign.headers

    # 使用 httpx 上传文件到预签名 URL
    import httpx
    async with httpx.AsyncClient() as http:
        await http.put(presign.upload_url, content=file_bytes, headers=presign.headers)

    # 获取签名下载 URL
    sign = await client.sign(SignRequest(object_key=presign.object_key))
    # sign.url

    # 删除文件
    await client.delete(DeleteRequest(object_key=presign.object_key))

便捷方法

async with RagentOssClient("https://oss.ragents.net", "your-api-key") as client:
    # 一步上传（presign + PUT）
    object_key = await client.upload(
        filename="report.pdf",
        content=file_bytes,
        content_type="application/pdf",
        category="knowledge",
    )

    # 一步下载（sign + GET）
    content = await client.download(object_key)

    # 下载到临时文件（流式写入）
    temp_path = await client.download_to_temp(object_key)

API 参考

RagentOssClient(base_url, api_key)

支持 async with 上下文管理器，也可以手动调用 await client.close()。

await client.health()

检查服务状态，不需要认证。

返回：HealthResponse(status, storage)

await client.presign(req)

获取预签名上传 URL。

返回：PresignResponse(object_key, upload_url, headers)

await client.sign(req)

获取签名下载 URL。

返回：SignResponse(url)

await client.delete(req)

删除对象。

返回：DeleteResponse(success)

await client.upload(filename, content, content_type, category)

便捷方法：presign + PUT。返回 object_key: str。

await client.download(object_key, expires_in?)

便捷方法：sign + GET。返回 bytes。

await client.download_to_temp(object_key, expires_in?)

便捷方法：sign + 流式下载到临时文件。返回 Path。

类型定义

所有类型都是 dataclass，不依赖 Pydantic。

PresignRequest(filename: str, content_type: str, category: str)
PresignResponse(object_key: str, upload_url: str, headers: dict[str, str])
SignRequest(object_key: str, expires_in: int | None = None)
SignResponse(url: str)
DeleteRequest(object_key: str)
DeleteResponse(success: bool)
HealthResponse(status: str, storage: str)

API 使用 camelCase（objectKey），SDK 自动转换为 snake_case（object_key）。

错误处理

from ragent_oss import (
    RagentOssError,
    AuthenticationError,
    ValidationError,
    UploadError,
    NetworkError,
)

try:
    await client.presign(...)
except AuthenticationError:
    # 401 - API 密钥无效
    pass
except ValidationError:
    # 422 - 请求参数错误
    pass
except NetworkError:
    # 网络不可达
    pass

开发

cd sdk/python
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
pytest