Python SDK

Programmatically create and manage runs using the Daft Cloud SDK.

What is the Daft Cloud SDK?

The Daft Cloud Python SDK enables you to programmatically interact with Daft Cloud to:

• Create and trigger runs from your Python applications
• Monitor run status and retrieve results
• Automate workflows and CI/CD pipelines
• Integrate Daft Cloud into your existing systems

While the Runs UI is great for manual run creation and monitoring, the SDK allows you to build automated workflows and integrate Daft Cloud into your applications.

Installation

Install the SDK from PyPI:

pip install ev-sdk

Authentication

To use the SDK, you’ll need an access token from Daft Cloud.

Creating an Access Token

1. Navigate to cloud.daft.ai/settings/access-tokens
2. Click Create token
3. Enter a descriptive name
4. Click Create
5. Copy the token immediately - it will only be shown once

Access tokens are workspace-scoped credentials that authenticate your requests to the Daft Cloud API. Unlike API keys which are for model inference, access tokens allow you to create and manage runs.

Using Your Access Token

Store your access token securely as an environment variable:

export DAFT_CLOUD_ACCESS_TOKEN="your-token-here"

Creating and Monitoring Runs

Here’s a complete example showing how to create a run and monitor its progress:

from __future__ import annotations

import logging
import os
import time

from ev.client import Client
from ev.models import (
    CreateRunRequest,
    CreateRunResponse,
    FileEntrypoint,
    FunctionEntrypoint,
    GitSource,
    RunEntrypoint,
    RunEnvironment,
    RunSource,
    RunStatus,
)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)


def main() -> int:
    # Authenticate with your access token
    access_token = os.environ["DAFT_CLOUD_ACCESS_TOKEN"]

    # Create a client pointing to Daft Cloud
    client = Client("https://api.daft.ai", access_token)

    # Get your workspace and project IDs from the Daft Cloud UI.
    # To get your Workspace ID, go to https://cloud.daft.ai/settings
    # To get your Project ID, select your project from https://cloud.daft.ai/projects,
    # then go to Settings.
    workspace_id = "your-workspace-id"
    project_id = "your-project-id"

    # Create a run request
    response: CreateRunResponse = client.create_run(
        workspace_id,
        project_id,
        CreateRunRequest(
            # Specify what to run
            entrypoint=RunEntrypoint({
                "function": FunctionEntrypoint(
                    module="pipeline.py",
                    symbol="my_function",
                    args=[],           # Positional arguments
                    kwargs={},         # Keyword arguments
                )
            }),
            # Specify where the code lives
            source=RunSource({
                "git": GitSource(
                    remote="https://github.com/your-org/your-repo",
                    hash="HEAD",  # or specific commit hash
                    private=False,
                )
            }),
            # Configure the runtime environment
            environment=RunEnvironment(
                python_version="3.12",
                dependencies=[],            # e.g., ["pandas", "numpy>=1.24.0"]
                environment_variables={},   # Additional env vars
            ),
            secrets={},  # Reference project secrets by name
        ),
    )

    # Monitor the run
    run_id = response.run.id
    run_url = f"{client.endpoint.replace('.api.', '.cloud.')}/~/{run_id}"
    logger.info("Created run: %s", run_url)
    logger.info("Status: %s", response.run.status.value)

    # Poll for completion
    status = response.run.status
    while not status.is_complete():
        time.sleep(1)
        status = client.get_run_status(workspace_id, project_id, run_id)
        logger.debug("Status: %s", status.value)

    logger.info("Run completed with status: %s", status.value)

    # Return exit code based on result
    if status == RunStatus.SUCCEEDED:
        logger.info("Run succeeded!")
        return 0
    else:
        logger.error("Run failed with status: %s", status.value)
        return 1


if __name__ == "__main__":
    exit(main())

Run Configuration Options

Entrypoints

The SDK supports the same entrypoint types as the Runs UI:

File Entrypoint: Execute a Python script

entrypoint=RunEntrypoint({
    "file": FileEntrypoint(
        file_path="pipeline.py",
        argv=["--input", "data.csv"],
    )
}),

Function Entrypoint: Call a specific function

entrypoint=RunEntrypoint({
    "function": FunctionEntrypoint(
        module="pipeline.py",
        symbol="my_function",
        args=[],           # Positional arguments
        kwargs={},         # Keyword arguments
    )
}),

Source

Specify where your code is located:

Git Source: Clone from a Git repository

source=RunSource({
    "git": GitSource(
        remote="https://github.com/your-org/your-repo",
        hash="main",        # branch, tag, or commit hash
        private=False,      # True for private repos
    )
})

Environment

Configure the runtime environment:

environment=RunEnvironment(
    python_version="3.12",
    dependencies=[
        "pandas==2.0.0",
        "numpy>=1.24.0",
        "daft",
    ],
    environment_variables={
        "LOG_LEVEL": "INFO",
        "BATCH_SIZE": "1000",
    },
)

Secrets

Reference project secrets by name:

secrets={
    "aws_credentials": "my-aws-secret",
    "api_key": "my-api-key-secret",
}

Finding Your IDs

You’ll need your workspace and project IDs to create runs:

• Workspace ID: Found at cloud.daft.ai/settings
• Project ID: Found at cloud.daft.ai/projects (click into your project and go to Settings)

Next Steps

• Runs - Learn more about run configuration and monitoring
• Secrets - Securely manage credentials for your runs
• Data Sources - Connect to external data systems
• API Keys - Access Daft Cloud models for inference