Fleet SDK

The Fleet Python SDK provides programmatic access to Fleet's environment infrastructure.

Installation

pip install fleet-python

API Key Setup

Get your API key from the Fleet Dashboard, then set it as an environment variable:

export FLEET_API_KEY="sk_your_key_here"

Quick Start

import fleet

async def main():
    # Load a task
    tasks = await fleet.load_tasks_async(
        keys=["task_abcdef"]
    )
    task = tasks[0]

    # Create an environment from the task
    env = await fleet.env.make_async(
        env_key=task.env_key,
        data_key=task.data_key,
        env_variables=task.env_variables,
        ttl_seconds=7200,
        run_id="run-123",
    )

    # Access the environment URL
    print(env.urls.app[0])

    # ... interact with the environment ...

    # Clean up
    await env.close()

Loading Tasks

By Task Keys

tasks = await fleet.load_tasks_async(
    keys=["task_abcdef"]
)

By Project Key

tasks = await fleet.load_tasks_async(project_key="my-project")

Creating Environments

env = await fleet.env.make_async(
    env_key=task.env_key,
    data_key=task.data_key,
    env_variables=task.env_variables,
    ttl_seconds=7200,
    run_id="run-123",
)

With Heartbeats (Optional)

Optionally enable heartbeats to keep environments alive during long-running operations:

env = await fleet.env.make_async(
    env_key=task.env_key,
    data_key=task.data_key,
    env_variables=task.env_variables,
    ttl_seconds=10800,
    heartbeat_interval=30,  # seconds
)

Send heartbeats to keep the environment alive:

# Via the environment object
await env.heartbeat()

# Or via instance ID
await fleet.env.heartbeat_async(instance_id)

Heartbeats are optional. If heartbeat_interval is not set, the instance lifetime is controlled solely by ttl_seconds. If heartbeats are enabled and missed 3 consecutive times, the instance will be terminated. Heartbeats take precedence over the TTL.

Instance Management

List Instances

# List all instances for a run
instances = await fleet.env.list_instances_async(run_id="run-123")

# List all instances for your profile
instances = await fleet.env.list_instances_async(profile_id="self")

Close Instances

# Close all instances for a run
await fleet.env.close_all_async(run_id="run-123")

# Close all instances for your profile
await fleet.env.close_all_async(profile_id="self")

# Close a specific instance by ID
await fleet.env.close_async("bc8954c2")

"self" is an alias for the profile associated with your FLEET_API_KEY.

Account Information

View your current account details including team info, instance limits, and profile ID:

account = await fleet.env.account_async()

Returns:

{
  "team_id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
  "team_name": "My Team",
  "instance_limit": 32000,
  "instance_count": 924,
  "profile_id": "11111111-2222-3333-4444-555555555555",
  "profile_name": "Jane Doe"
}

Run Tracking

Track active and past runs:

# List active runs
runs = await fleet.env.list_runs_async()

# List all runs (active and inactive)
runs = await fleet.env.list_runs_async(status="all")

# Filter by profile
runs = await fleet.env.list_runs_async(profile_id="self")

Returns:

[
  {
    "run_id": "run-123",
    "running_count": 0,
    "total_count": 4,
    "first_created_at": "2025-10-24T09:48:47.152387",
    "last_created_at": "2025-10-24T09:55:19.284294",
    "profile_id": "11111111-2222-3333-4444-555555555555"
  }
]

Complete Example

import fleet
import asyncio

async def main():
    # Load tasks from a project
    tasks = await fleet.load_tasks_async(project_key="my-project")
    
    for task in tasks:
        # Create environment with heartbeat support
        env = await fleet.env.make_async(
            env_key=task.env_key,
            data_key=task.data_key,
            env_variables=task.env_variables,
            ttl_seconds=7200,
            run_id="my-evaluation-run",
            heartbeat_interval=30,
        )
        
        try:
            # Send periodic heartbeats during long operations
            await env.heartbeat()
            
            # ... run your evaluation ...
            
        finally:
            await env.close()
    
    # Clean up all instances from this run
    await fleet.env.close_all_async(run_id="my-evaluation-run")

if __name__ == "__main__":
    asyncio.run(main())