CloudFactory AI Platform Python Client

Official Python SDK for the CloudFactory AI Platform. Download COCO format annotations from your projects with flexible filtering (batches, work items, timestamps), concurrent workers, automatic retries, and progress tracking.

Prerequisites

Before you start, you will need:

• Python 3.10 or later
• A CloudFactory API Key
• Your Project ID
• Your Space ID

Installation

uv add cf-ai-platform-client

Or with pip:

pip install cf-ai-platform-client

Quick Start

Option A: Synchronous (simpler, no async required)

If you are writing a script, running in a Jupyter notebook, or are not familiar with Python async/await syntax, use SyncCFUserClient:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("YOUR_API_KEY") as client:
    summary = client.download_project(
        project_id="YOUR_PROJECT_ID",
        space_id="YOUR_SPACE_ID",
        destination="./downloads"
    )

print(f"Downloaded {summary.successful} of {summary.total_items} items")
if summary.failed > 0:
    print(f"  {summary.failed} items failed")

Option B: Async script or entry point

If you are writing an async application or prefer async/await syntax, use CFUserClient:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("YOUR_API_KEY") as client:
        summary = await client.download_project(
            project_id="YOUR_PROJECT_ID",
            space_id="YOUR_SPACE_ID",
            destination="./downloads"
        )

    print(f"Downloaded {summary.successful} of {summary.total_items} items")

asyncio.run(main())

Usage Examples

The examples below use SyncCFUserClient for simplicity. Async equivalents use CFUserClient with await and async with.

Basic Download

Download all COCO annotations from a project:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads"
    )

    if summary.failed > 0:
        print("Some downloads failed:")
        for result in summary.results:
            if not result.success:
                print(f"  - Item {result.item_id}: {result.error}")

Filter by Batch IDs

Download only specific batches (up to 100 batch IDs):

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads",
        batch_ids=[1, 2, 3]
    )

Filter by Work Item IDs

Download specific work items by their UUIDs (up to 100):

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads",
        work_item_ids=[
            "550e8400-e29b-41d4-a716-446655440000",
            "6ba7b810-9dad-11d1-80b4-00c04fd430c8"
        ]
    )

Filter by Date (ISO 8601)

Filter by creation date:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads",
        created_at_from="2026-01-01T00:00:00Z",
        created_at_to="2026-01-31T23:59:59Z"
    )

Filter by last updated date:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads",
        updated_at_from="2026-02-11T00:00:00Z",
        updated_at_to="2026-02-18T23:59:59Z"
    )

Note: these filters apply to work items — when they were created and last updated.

Combine Multiple Filters

All filters can be combined freely:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads",
        batch_ids=[1, 2],
        created_at_from="2026-01-01T00:00:00Z",
        created_at_to="2026-01-31T23:59:59Z",
        max_workers=15
    )

Configure Concurrency

By default the SDK downloads 10 work items concurrently. Increase this for faster downloads or decrease it to reduce load:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads",
        max_workers=20  # default: 10
    )

Inspect Download Results

Access detailed information about each downloaded item:

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads"
    )

    for result in summary.results:
        if result.success:
            print(f"✅ {result.item_id} -> {result.file_path}")
        else:
            print(f"❌ {result.item_id}: {result.error}")

Custom Base URL

Use a custom API endpoint (e.g., for staging):

from cf_ai_platform_client import SyncCFUserClient

with SyncCFUserClient.create_with_api_key(
    api_key="MY_API_KEY",
    base_url="https://api.staging.ai.cloudfactory.app/api/v1"
) as client:
    summary = client.download_project(
        project_id="proj_abc123",
        space_id="space_456",
        destination="./my_downloads"
    )

File Organization

Downloaded COCO annotation files are organized by batch name:

destination/
├── Default/
│   ├── <work_item_id>.json
│   └── <work_item_id>.json
└── Training_Batch_001/
    └── <work_item_id>.json

• Files are grouped by batch name (fetched from batch metadata)
• If no batches have been defined, all items will be under the Default batch
• File naming: {work_item_id}.{extension}
• Extension is auto-detected from URL (defaults to .json)
• Each .json file is a COCO-format annotation with an asset_url field injected into the info block so you can trace it back to the original asset
• Only COMPLETED work items are downloaded by default

API Reference

SyncCFUserClient (synchronous)

Synchronous wrapper around CFUserClient. Runs a background event loop in a daemon thread — all methods block until complete. Use with for context management.

Methods

create_with_api_key(api_key: str, base_url: Optional[str] = None) -> SyncCFUserClient

Create a client instance with API key authentication.

Parameters:

• api_key (str): Your CloudFactory API key
• base_url (Optional[str]): Custom API base URL (default: production)

Returns: Configured SyncCFUserClient instance

download_project(project_id, space_id, destination, ...) -> DownloadSummary

Download all COCO annotations from a project with flexible filtering options. Blocks until complete.

All parameters are identical to CFUserClient.download_project.

close() -> None

Close HTTP clients and stop the background event loop. Called automatically when using with.

CFUserClient (async)

Async client for the CloudFactory AI Platform. All methods must be awaited; use async with for context management.

Architecture

The download process follows these steps:

1. Fetch Batches: Retrieve batch metadata (ID → name mapping) from the Platform API
2. Fetch Manifest: Query work items with optional filters (status defaults to "COMPLETED")
3. Parallel Downloads: Use asyncio with semaphore to download annotations concurrently
4. For Each Item:
   • Fetch annotation metadata from Platform API
   • Extract COCO results URL from annotation metadata
   • Generate signed download URL from Space API
   • Download file content with retry logic
   • Save to disk organized by batch name
5. Return Summary: Aggregate statistics and per-item results

Methods

create_with_api_key(api_key: str, base_url: Optional[str] = None) -> CFUserClient

Create a client instance with API key authentication.

Parameters:

• api_key (str): Your CloudFactory API key
• base_url (Optional[str]): Custom API base URL (default: production)

Returns: Configured CFUserClient instance

async download_project(project_id, space_id, destination, ...) -> DownloadSummary

Download all COCO annotations from a project with flexible filtering options. Must be awaited.

Parameters:

• project_id (str): The project ID to download
• space_id (str): The space ID (required for generating signed URLs)
• destination (str): Local directory path to save files
• batch_ids (Optional[list[int]]): Filter by specific batch IDs (max 100). Downloads all if None
• work_item_ids (Optional[list[str]]): Filter by specific work item IDs (UUID format). Downloads all if None
• created_at_from (Optional[str]): Filter items created on or after this timestamp (ISO 8601 format, e.g., "2026-01-15T10:30:00Z")
• created_at_to (Optional[str]): Filter items created on or before this timestamp (ISO 8601 format)
• updated_at_from (Optional[str]): Filter items updated on or after this timestamp (ISO 8601 format)
• updated_at_to (Optional[str]): Filter items updated on or before this timestamp (ISO 8601 format)
• max_workers (int): Maximum concurrent downloads (default: 10)
• limit (Optional[int]): Items per page when fetching the manifest (1–100, default: API default of 20)

Returns: DownloadSummary with statistics and detailed results

Raises:

• httpx.HTTPStatusError: If API requests fail
• OSError: If file system operations fail

async close() -> None

Close HTTP clients and cleanup resources. Must be awaited. Called automatically when using async with.

Models

DownloadSummary

Summary statistics for a download operation.

DownloadResult

Result of a single work item download.

Error Handling

The SDK includes comprehensive error handling:

Automatic Retries

Downloads automatically retry up to 3 times with exponential backoff (2-10 seconds) for:

• Network errors and timeouts (httpx.HTTPError, httpx.TimeoutException)
• All HTTP errors (including 4xx and 5xx status codes)

Retries apply only to the download phase (fetching files from signed URLs). API calls to fetch manifests, batches, and annotations do not have automatic retries.

Graceful Failures

If a download fails after retries:

• Error is logged
• Item is marked as failed in results
• Download continues for remaining items
• Summary includes failure details

Example Error Handling

import logging
from cf_ai_platform_client import SyncCFUserClient

logging.basicConfig(level=logging.INFO)

with SyncCFUserClient.create_with_api_key("MY_API_KEY") as client:
    try:
        summary = client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./downloads"
        )

        if summary.failed > 0:
            print(f"\n⚠️  {summary.failed} items failed to download:")
            for result in summary.results:
                if not result.success:
                    print(f"  - {result.item_id}: {result.error}")

    except Exception as e:
        print(f"❌ Download failed: {e}")
        raise

Requirements

• Python 3.10 to 3.14 (actively supported versions)
• Dependencies:
  • httpx >= 0.27.0 - HTTP client
  • pydantic >= 2.6.0 - Data validation
  • tqdm >= 4.66.0 - Progress bars
  • tenacity >= 8.2.0 - Retry logic

Development

Setup

This project uses uv for fast dependency management. The setup script installs uv with checksum verification for security:

# Clone the repository
git clone https://github.com/cloudfactory/cf-ai-platform-client-python.git
cd cf-ai-platform-client-python

# Run setup (installs uv with checksum verification)
./setup.sh --dev

Alternative uv installation methods:

If you prefer to install uv yourself:

# macOS (recommended)
brew install uv

# Using pipx
pipx install uv

# Using pip
pip install uv

# Then install dependencies
uv sync --all-extras

Security note: The setup script downloads a pinned uv release (v0.10.4) with SHA256 checksum verification to prevent supply-chain attacks. CI/CD workflows use the official astral-sh/setup-uv@v4 GitHub Action.

Running Tests

# Run all tests with coverage
uv run pytest

# Run with verbose output
uv run pytest -v

# Run specific test file
uv run pytest tests/test_client.py

# Generate HTML coverage report
uv run pytest --cov=src --cov-report=html

Code Quality

# Format code
uv run ruff format src tests

# Lint code
uv run ruff check src tests

# Type checking
uv run mypy src

# Run all checks
uv run ruff format src tests && uv run ruff check --fix src tests && uv run mypy src

Building and Publishing

# Build distribution packages
uv build

# Publish to TestPyPI (for testing)
uv publish --publish-url https://test.pypi.org/legacy/

# Publish to PyPI (production)
uv publish

Note: Publishing is automated via GitHub Actions. Push to main triggers the production release workflow.

License

MIT License - see LICENSE file for details.