diff --git a/README.md b/README.md index 04b40107..616f2927 100644 --- a/README.md +++ b/README.md @@ -8,88 +8,278 @@ The Fleet Python SDK provides programmatic access to Fleet's environment infrast ## Installation -Install the Fleet SDK using pip: - ```bash pip install fleet-python ``` -### Alpha/Pre-release Versions +## API Key Setup -To install the latest alpha or pre-release version: +Get your API key from the [Fleet Dashboard](https://fleetai.com/dashboard/api-keys), then set it as an environment variable: ```bash -pip install --pre fleet-python +export FLEET_API_KEY="sk_your_key_here" ``` -To install a specific alpha version: +## Tasks -```bash -pip install fleet-python==0.2.64-alpha1 +A **Task** represents a unit of work for an agent to complete within an environment. Each task combines: + +- **Prompt**: Instructions describing what needs to be done +- **Environment**: The environment configuration (`env_key`, `data_key`, `env_variables`) +- **Verifier**: Code that validates task completion and returns a score (0.0 to 1.0) + +Tasks provide all the configuration needed to spin up an environment and verify an agent's work. + +### Task Properties + +| Property | Description | +| --------------- | ----------------------------------------------------- | +| `key` | Unique task identifier | +| `prompt` | Instructions for the agent | +| `env_key` | Environment identifier (e.g., `"hubspot:v1.2"`) | +| `data_key` | Data configuration identifier | +| `env_variables` | Environment variables for the task | +| `metadata` | Additional info (e.g., `avg_steps` for task difficulty) | + +## Quick Start + +```python +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 ... + + # Verify task completion + result = await task.verify_detailed_async(env.instance_id) + print(result) + + # Clean up + await env.close() ``` -## API Key Setup +## Loading Tasks -Fleet requires an API key for authentication. You can obtain one from the [Fleet Platform](https://fleetai.com/dashboard/api-keys). +### By Task Keys -Set your API key as an environment variable: +```python +tasks = await fleet.load_tasks_async( + keys=["task_abcdef"] +) +``` -```bash -export FLEET_API_KEY="sk_your_key_here" +### By Project Key + +```python +tasks = await fleet.load_tasks_async(project_key="my-project") ``` -## Basic Usage +## Creating Environments ```python -import fleet -import datetime +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", +) +``` -# Create environment by key -env = fleet.env.make("fira") +### With Heartbeats (Optional) -# Reset environment with seed and options -env.reset( - seed=42, - timestamp=int(datetime.datetime.now().timestamp()) +Optionally enable heartbeats to keep environments alive during long-running operations: + +```python +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: + +```python +# Via the environment object +await env.heartbeat() -# Access environment state ('current' is the resource id for a sqlite database) -sql = env.state("sqlite://current") -sql.exec("UPDATE customers SET status = 'active' WHERE id = 123") +# 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 + +```python +# 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 -# Clean up -env.close() +```python +# 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") ``` -## Environment Management +`"self"` is an alias for the profile associated with your `FLEET_API_KEY`. -### Creating Instances +## Account Information + +View your current account details including team info, instance limits, and profile ID: ```python -# Create environment instance with explicit version -env = fleet.env.make("fira:v1.2.5") +account = await fleet.env.account_async() +``` -# Create environment instance with default (latest) version -env = fleet.env.make("fira") +Returns: +```json +{ + "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" +} ``` -### Connecting to Existing Instances +## Run Tracking + +Track active and past runs: ```python -# Connect to a running instance -env = fleet.env.get("env_instance_id") +# 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") +``` -# List all running instances -instances = fleet.env.list_instances() -for instance in instances: - print(f"Instance: {instance.instance_id}") - print(f"Type: {instance.environment_type}") - print(f"Status: {instance.status}") +Returns: + +```json +[ + { + "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" + } +] +``` + +## Task Verification + +Verify task completion and get detailed results: + +```python +result = await task.verify_detailed_async(env.instance_id) +print(result) +``` + +Returns: + +```json +{ + "key": "task_abcdef", + "version": 4, + "success": true, + "result": 1.0, + "error": null, + "execution_time_ms": 2291, + "stdout": "" +} +``` + +On failure, `stdout` contains detailed verification errors: + +```json +{ + "key": "task_abcdef", + "version": 4, + "success": true, + "result": 0, + "error": null, + "execution_time_ms": 2291, + "stdout": "Verification errors: [\"Expected field to be 'value', got None\", \"Form not marked as complete\"]" +} +``` + +## Complete Example + +```python +import fleet +import asyncio -# Filter instances by status (running, pending, stopped, error) -running_instances = fleet.env.list_instances(status_filter="running") +async def main(): + # Load tasks from a project + tasks = await fleet.load_tasks_async(project_key="my-project") + + for task in tasks: + # Create environment + 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", + ) + + try: + # Access the environment URL + print(env.urls.app[0]) + + # ... run your agent ... + + # Verify task completion + result = await task.verify_detailed_async(env.instance_id) + print(f"Task {task.key}: score={result['result']}") + + finally: + await env.close() + + # Clean up all instances from this run + await fleet.env.close_all_async(run_id="my-evaluation-run") -# List available environment types -available_envs = fleet.env.list_envs() +if __name__ == "__main__": + asyncio.run(main()) ``` diff --git a/examples/example_task.py b/examples/example_task.py index 7840bc69..996e51e6 100644 --- a/examples/example_task.py +++ b/examples/example_task.py @@ -112,7 +112,7 @@ async def main(): print(f"Task definition:") print(f" Key: {task.key}") print(f" Prompt: {task.prompt}") - print(f" Environment: {task.env_id}") + print(f" Environment: {task.env_key}") print( f" Verifier: {task.verifier.key if hasattr(task.verifier, 'key') else 'create_bug_issue'}" ) diff --git a/examples/export_tasks.py b/examples/export_tasks.py index 1632ff96..9582cf2b 100644 --- a/examples/export_tasks.py +++ b/examples/export_tasks.py @@ -38,6 +38,12 @@ def main(): help="Output JSON filename (defaults to {team_id}.json)", default=None, ) + parser.add_argument( + "--skip-missing-verifiers", + "-s", + action="store_true", + help="Skip tasks missing verifier_func instead of erroring", + ) args = parser.parse_args() @@ -91,16 +97,32 @@ def main(): missing_verifier.append(task.key) if missing_verifier: - print(f"\n✗ Error: {len(missing_verifier)} task(s) missing verifier_func:") - for key in missing_verifier[:10]: # Show first 10 - print(f" - {key}") - if len(missing_verifier) > 10: - print(f" ... and {len(missing_verifier) - 10} more") - raise ValueError( - "All tasks must have a verifier_func. Cannot export tasks without verifiers." - ) - - print("✓ All tasks have verifier_func") + if args.skip_missing_verifiers: + print( + f"\n⚠ Skipping {len(missing_verifier)} task(s) missing verifier_func:" + ) + for key in missing_verifier[:10]: # Show first 10 + print(f" - {key}") + if len(missing_verifier) > 10: + print(f" ... and {len(missing_verifier) - 10} more") + # Filter out tasks without verifiers + tasks = [task for task in tasks if task.verifier_func] + print(f"\n{len(tasks)} task(s) remaining after filtering") + if not tasks: + raise ValueError( + "No tasks remaining after filtering. Nothing to export." + ) + else: + print(f"\n✗ Error: {len(missing_verifier)} task(s) missing verifier_func:") + for key in missing_verifier[:10]: # Show first 10 + print(f" - {key}") + if len(missing_verifier) > 10: + print(f" ... and {len(missing_verifier) - 10} more") + raise ValueError( + "All tasks must have a verifier_func. Cannot export tasks without verifiers." + ) + else: + print("✓ All tasks have verifier_func") # Determine output filename output_file = args.output or f"{account.team_id}.json" diff --git a/examples/export_tasks_filtered.py b/examples/export_tasks_filtered.py deleted file mode 100644 index 40066ad2..00000000 --- a/examples/export_tasks_filtered.py +++ /dev/null @@ -1,245 +0,0 @@ -""" -Export tasks to JSON, excluding tasks from targets marked as 'unused'. - -This script filters out tasks whose task_project_target has status='unused', -ensuring that broken/invalid targets don't pollute exports. - -Usage: - python export_tasks_filtered.py --task-project-key my-project - python export_tasks_filtered.py --project-key my-project --output tasks.json - python export_tasks_filtered.py --env-key my-env -""" - -import argparse -import json -import os -from typing import List, Set - -import fleet -from dotenv import load_dotenv -from supabase import create_client, Client - -load_dotenv() - - -def get_unused_target_ids(supabase: Client, team_id: str) -> Set[str]: - """Fetch all target IDs that have status='unused' for the given team.""" - # Get all task_projects for this team first - projects_response = ( - supabase.table("task_projects") - .select("id") - .eq("team_id", team_id) - .execute() - ) - - if not projects_response.data: - return set() - - project_ids = [p["id"] for p in projects_response.data] - - # Get all targets with status='unused' for these projects - targets_response = ( - supabase.table("task_project_targets") - .select("id") - .in_("project_id", project_ids) - .eq("status", "unused") - .execute() - ) - - if not targets_response.data: - return set() - - return {t["id"] for t in targets_response.data} - - -def get_task_target_mapping(supabase: Client, task_keys: List[str], team_id: str) -> dict: - """Fetch task_project_target_id for each task key.""" - if not task_keys: - return {} - - # Batch the queries to avoid hitting limits - BATCH_SIZE = 100 - mapping = {} - - for i in range(0, len(task_keys), BATCH_SIZE): - batch_keys = task_keys[i:i + BATCH_SIZE] - response = ( - supabase.table("eval_tasks") - .select("key, task_project_target_id") - .in_("key", batch_keys) - .eq("team_id", team_id) - .execute() - ) - - for row in response.data or []: - mapping[row["key"]] = row.get("task_project_target_id") - - return mapping - - -def main(): - parser = argparse.ArgumentParser( - description="Export tasks to JSON, excluding tasks from 'unused' targets" - ) - parser.add_argument( - "--project-key", - "-p", - help="Optional project key to filter tasks", - default=None, - ) - parser.add_argument( - "--task-keys", - "-t", - nargs="+", - help="Optional list of task keys to export (space-separated)", - default=None, - ) - parser.add_argument( - "--task-project-key", - "-tpk", - help="Optional task project key to filter tasks", - default=None, - ) - parser.add_argument( - "--env-key", - "-e", - help="Optional environment key to filter tasks", - default=None, - ) - parser.add_argument( - "--output", - "-o", - help="Output JSON filename (defaults to {team_id}_filtered.json)", - default=None, - ) - parser.add_argument( - "--include-unused", - action="store_true", - help="Include tasks from 'unused' targets (disables filtering)", - ) - - args = parser.parse_args() - - # Validate that only one filter is specified - filters_specified = sum( - [ - args.project_key is not None, - args.task_keys is not None, - args.task_project_key is not None, - args.env_key is not None, - ] - ) - - if filters_specified > 1: - parser.error( - "Cannot specify multiple filters. Use only one of --project-key, --task-keys, --task-project-key, or --env-key." - ) - - # Get account info - account = fleet.env.account() - print(f"Exporting from team: {account.team_name}") - - # Initialize Supabase client for filtering - supabase_url = os.getenv("SUPABASE_URL") or os.getenv("NEXT_PUBLIC_SUPABASE_URL") - supabase_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY") or os.getenv("SUPABASE_KEY") - - if not supabase_url or not supabase_key: - print("⚠ Warning: SUPABASE_URL/SUPABASE_KEY not set - cannot filter by target status") - print(" Falling back to unfiltered export") - supabase = None - else: - supabase = create_client(supabase_url, supabase_key) - - # Load tasks - if args.project_key: - print(f"Loading tasks from project: {args.project_key}") - tasks = fleet.load_tasks(project_key=args.project_key) - elif args.task_keys: - print(f"Loading {len(args.task_keys)} specific task(s): {', '.join(args.task_keys)}") - tasks = fleet.load_tasks(keys=args.task_keys) - elif args.task_project_key: - print(f"Loading tasks from task project: {args.task_project_key}") - tasks = fleet.load_tasks(task_project_key=args.task_project_key) - elif args.env_key: - print(f"Loading tasks from environment: {args.env_key}") - tasks = fleet.load_tasks(env_key=args.env_key) - else: - print("Loading all tasks") - tasks = fleet.load_tasks() - - print(f"\nFound {len(tasks)} task(s) before filtering") - - # Filter out tasks from unused targets - filtered_tasks = tasks - excluded_count = 0 - - if supabase and not args.include_unused: - print("\nFiltering out tasks from 'unused' targets...") - - # Get unused target IDs - unused_target_ids = get_unused_target_ids(supabase, account.team_id) - - if unused_target_ids: - print(f" Found {len(unused_target_ids)} unused target(s)") - - # Get task -> target mapping - task_keys = [t.key for t in tasks] - task_target_map = get_task_target_mapping(supabase, task_keys, account.team_id) - - # Filter tasks - filtered_tasks = [] - for task in tasks: - target_id = task_target_map.get(task.key) - if target_id in unused_target_ids: - excluded_count += 1 - else: - filtered_tasks.append(task) - - print(f" Excluded {excluded_count} task(s) from unused targets") - else: - print(" No unused targets found - all tasks included") - - tasks = filtered_tasks - print(f"\n{len(tasks)} task(s) after filtering") - - # Validate that all tasks have verifier_func - print("\nValidating tasks have verifier_func...") - missing_verifier = [] - for task in tasks: - if not task.verifier_func: - missing_verifier.append(task.key) - - if missing_verifier: - print(f"\n✗ Error: {len(missing_verifier)} task(s) missing verifier_func:") - for key in missing_verifier[:10]: # Show first 10 - print(f" - {key}") - if len(missing_verifier) > 10: - print(f" ... and {len(missing_verifier) - 10} more") - raise ValueError( - "All tasks must have a verifier_func. Cannot export tasks without verifiers." - ) - - print("✓ All tasks have verifier_func") - - # Determine output filename - output_file = args.output or f"{account.team_id}_filtered.json" - - # Export to JSON - print(f"\nExporting to: {output_file}") - with open(output_file, "w", encoding="utf-8") as f: - json.dump( - [task.model_dump() for task in tasks], - f, - indent=2, - ensure_ascii=False, - ) - - print(f"✓ Successfully exported {len(tasks)} task(s) to {output_file}") - if excluded_count > 0: - print(f" ({excluded_count} task(s) excluded from unused targets)") - - -if __name__ == "__main__": - main() - - diff --git a/examples/import_tasks.py b/examples/import_tasks.py index bf0d744a..7e91913a 100644 --- a/examples/import_tasks.py +++ b/examples/import_tasks.py @@ -35,7 +35,7 @@ async def run_verifier_sanity_check( instance_groups = defaultdict(list) for task in tasks: # Build the instance key - env_key = task.env_id or "" + env_key = task.env_key or "" env_version = task.version or "" data_key = task.data_id or "" data_version = task.data_version or "" @@ -238,7 +238,6 @@ async def main(): task_count = len(tasks_data) task_keys = [] missing_verifier = [] - tasks_with_output_schema = [] for task_data in tasks_data: task_key = task_data.get("key") or task_data.get("id") if task_key: @@ -251,10 +250,6 @@ async def main(): if not verifier_code: missing_verifier.append(task_key or "(no key)") - # Check for output_json_schema - if task_data.get("output_json_schema"): - tasks_with_output_schema.append(task_key or "(no key)") - # Validate all tasks have verifier_func if missing_verifier: print(f"✗ Error: {len(missing_verifier)} task(s) missing verifier_func:") @@ -344,10 +339,6 @@ async def main(): tasks = passed_tasks passed_keys = {t.key for t in passed_tasks} tasks_data = [td for td in tasks_data if td.get("key") in passed_keys] - # Also filter tasks_with_output_schema - tasks_with_output_schema = [ - k for k in tasks_with_output_schema if k in passed_keys - ] print(f"\nProceeding with {len(tasks)} tasks that passed sanity check") else: @@ -397,45 +388,6 @@ async def main(): print(f"✓ Team: {account.team_name}") - # Print HUGE warning if any tasks have output_json_schema - if tasks_with_output_schema: - print("\n") - print("!" * 80) - print("!" * 80) - print("!" * 80) - print( - "!!! !!!" - ) - print( - "!!! ⚠️ WARNING WARNING WARNING ⚠️ !!!" - ) - print( - "!!! !!!" - ) - print( - f"!!! {len(tasks_with_output_schema)} TASK(S) HAVE OUTPUT_JSON_SCHEMA THAT NEED MANUAL COPYING! !!!" - ) - print( - "!!! !!!" - ) - print( - "!!! The output_json_schema field is NOT automatically imported! !!!" - ) - print( - "!!! You MUST manually copy the output schemas to each task! !!!" - ) - print( - "!!! !!!" - ) - print("!" * 80) - print("!" * 80) - print("!" * 80) - print("\nTasks with output_json_schema:") - for i, key in enumerate(tasks_with_output_schema[:20], 1): - print(f" {i}. {key}") - if len(tasks_with_output_schema) > 20: - print(f" ... and {len(tasks_with_output_schema) - 20} more") - print("\n⚠️ REMEMBER TO MANUALLY COPY OUTPUT SCHEMAS! ⚠️\n") except Exception as e: print(f"\n✗ Error importing tasks: {e}") sys.exit(1) diff --git a/examples/iterate_verifiers.py b/examples/iterate_verifiers.py index 2c73ecea..1c09e18f 100755 --- a/examples/iterate_verifiers.py +++ b/examples/iterate_verifiers.py @@ -677,12 +677,12 @@ def main(): epilog=""" Examples: # Extract verifiers from JSON to Python file - %(prog)s extract xai-day-10-batch.json verifiers.py + %(prog)s extract batch.json verifiers.py # Edit verifiers.py... # Apply changes back to JSON file (updates in-place) - %(prog)s apply xai-day-10-batch.json verifiers.py + %(prog)s apply batch.json verifiers.py # Validate verifiers file %(prog)s validate verifiers.py diff --git a/examples/task_bundle_editing/README.md b/examples/task_bundle_editing/README.md new file mode 100644 index 00000000..8873077f --- /dev/null +++ b/examples/task_bundle_editing/README.md @@ -0,0 +1,97 @@ +# Task Bundle Editing + +Download existing tasks, edit them locally, and upload as new tasks. + +## Setup + +```bash +pip install requests python-dotenv +``` + +Set your API key: + +```bash +export FLEET_API_KEY=your_api_key_here +``` + +Or create a `.env` file: + +``` +FLEET_API_KEY=your_api_key_here +``` + +By default, scripts talk to production (`https://orchestrator.fleetai.com`). +Override with `FLEET_BASE_URL` for staging/local: + +```bash +export FLEET_BASE_URL=https://staging.fleetai.com +``` + +## Workflow + +### 1. Download a task + +```bash +python download_task.py --task-key my_existing_task --output-dir ./my_task +``` + +This creates: + +``` +my_task/ + task.json # task metadata, prompt, verifier + files/ # data files (may be empty) + notebook.ipynb + data.csv +``` + +### 2. Edit the task + +Edit `task.json` to change the prompt, verifier, metadata, etc. +Add/remove/modify files in `files/`. + +### 3. Validate before upload + +```bash +python validate_task.py ./my_task --new-key my_new_task +``` + +Checks: valid JSON, required fields, verifier syntax, file sizes, key format. + +### 4. Upload as a new task + +```bash +python upload_task.py --dir ./my_task --key my_new_task +``` + +This validates the bundle, uploads files to a new file-set, then creates the task. + +The `--key` must differ from the original task key (safety check). + +## CLI Reference + +### download_task.py + +| Flag | Description | +|------|-------------| +| `--task-key` | Task key to download (required) | +| `--output-dir` | Output directory (default: `./`) | +| `--api-key` | API key (default: `FLEET_API_KEY` env var) | +| `--team-id` | Team ID override (default: auto-resolved from API key) | + +### upload_task.py + +| Flag | Description | +|------|-------------| +| `--dir` | Path to task bundle directory (required) | +| `--key` | New task key (required, must differ from original) | +| `--api-key` | API key (default: `FLEET_API_KEY` env var) | +| `--team-id` | Team ID override (default: auto-resolved from API key) | +| `--allow-overwrite` | Allow overwriting existing files in S3 | + +### validate_task.py + +| Flag | Description | +|------|-------------| +| `bundle_dir` | Path to task bundle directory (positional, required) | +| `--new-key` | New key to validate against | diff --git a/examples/task_bundle_editing/download_task.py b/examples/task_bundle_editing/download_task.py new file mode 100644 index 00000000..5b59d61a --- /dev/null +++ b/examples/task_bundle_editing/download_task.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +""" +Download a task and its data files to a local directory. + +Usage: + python download_task.py --task-key --output-dir ./my_task + +Creates: + my_task/ + task.json # task metadata, prompt, verifier + files/ # data files from file-set (may be empty) + notebook.ipynb + solution.py + +Requires: FLEET_API_KEY env var (or --api-key) +""" + +import argparse +import json +import os +import sys +from pathlib import Path + +import requests + +try: + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + pass + +DEFAULT_BASE_URL = "https://orchestrator.fleetai.com" + + +def get_api_base() -> str: + base = os.environ.get("FLEET_BASE_URL", DEFAULT_BASE_URL).rstrip("/") + return f"{base}/v1" + + +def headers(api_key: str) -> dict: + return { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + +def resolve_team_id(api_key: str) -> str: + """GET /v1/account → team_id for this API key.""" + resp = requests.get(f"{get_api_base()}/account", headers=headers(api_key)) + if not resp.ok: + print(f" ERROR resolving team: {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + account = resp.json() + team_id = account["team_id"] + team_name = account.get("team_name", "unknown") + print(f" Authenticated as team: {team_name} ({team_id})") + return team_id + + +def download_task(api_key: str, task_key: str, team_id: str | None = None) -> dict: + """GET /v1/tasks/{task_key} → task JSON.""" + print(f"\n1. Downloading task: {task_key}") + params = {} + if team_id: + params["team_id"] = team_id + resp = requests.get( + f"{get_api_base()}/tasks/{task_key}", + headers=headers(api_key), + params=params, + ) + if not resp.ok: + print(f" ERROR {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + task = resp.json() + print(f" prompt: {len(task['prompt'])} chars") + print(f" environment_id: {task['environment_id']}") + print(f" version: {task.get('version')}") + print(f" verifier_id: {task.get('verifier_id')}") + return task + + +def download_files(api_key: str, task_key: str, dest_dir: Path) -> list[Path]: + """POST /v1/file-sets/{key}/download-urls → download each file.""" + print(f"\n2. Fetching download URLs for file-set: {task_key}") + resp = requests.post( + f"{get_api_base()}/file-sets/{task_key}/download-urls", + headers=headers(api_key), + json={"expires_in": 3600}, + ) + if resp.status_code == 404: + print(" No file-set found for this task key (no data files).") + return [] + if not resp.ok: + print(f" ERROR {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + data = resp.json() + + urls = data.get("urls", []) + print(f" Found {len(urls)} files") + + downloaded = [] + for item in urls: + filename = item["filename"] + url = item["url"] + local_path = dest_dir / filename + local_path.parent.mkdir(parents=True, exist_ok=True) + + print(f" Downloading: {filename}") + file_resp = requests.get(url) + file_resp.raise_for_status() + local_path.write_bytes(file_resp.content) + downloaded.append(local_path) + print(f" -> {local_path} ({len(file_resp.content)} bytes)") + + return downloaded + + +def main(): + parser = argparse.ArgumentParser( + description="Download a task + data files to a local directory" + ) + parser.add_argument("--task-key", required=True, help="Task key to download") + parser.add_argument( + "--output-dir", + default=None, + help="Output directory (default: ./)", + ) + parser.add_argument( + "--api-key", + default=os.environ.get("FLEET_API_KEY"), + help="API key (default: FLEET_API_KEY env var)", + ) + parser.add_argument( + "--team-id", + help="Team ID override (default: auto-resolved from API key)", + ) + args = parser.parse_args() + + if not args.api_key: + print("Error: FLEET_API_KEY env var or --api-key required") + sys.exit(1) + + # Only pass team_id to task GET if explicitly provided (requires admin). + # Otherwise, resolve team info just for display. + if args.team_id: + team_id = args.team_id + else: + resolve_team_id(args.api_key) + team_id = None + + output_dir = Path(args.output_dir) if args.output_dir else Path(args.task_key) + output_dir.mkdir(parents=True, exist_ok=True) + files_dir = output_dir / "files" + files_dir.mkdir(exist_ok=True) + + print(f"Output directory: {output_dir}") + + # Download task metadata + task = download_task(args.api_key, args.task_key, team_id=team_id) + + # Save task JSON + task_path = output_dir / "task.json" + task_path.write_text(json.dumps(task, indent=2)) + print(f" Saved to: {task_path}") + + # Download data files — use TASK_KEY env variable as file-set key if available, + # since the file-set key may differ from the task key (e.g., without version suffix) + file_set_key = (task.get("env_variables") or {}).get("TASK_KEY", args.task_key) + downloaded = download_files(args.api_key, file_set_key, files_dir) + + print(f"\n-- Download complete --") + print(f" Task JSON: {task_path}") + print(f" Files ({len(downloaded)}): {files_dir}") + + +if __name__ == "__main__": + main() diff --git a/examples/task_bundle_editing/launch_job.py b/examples/task_bundle_editing/launch_job.py new file mode 100644 index 00000000..546fbed0 --- /dev/null +++ b/examples/task_bundle_editing/launch_job.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +""" +Launch a job for one or more existing tasks. + +Usage: + # Single task: + python launch_job.py --task-key my-task + + # Multiple tasks: + python launch_job.py --task-key task-a task-b task-c + + # Custom models and pass_k: + python launch_job.py --task-key my-task --models anthropic/claude-opus-4.6 --pass-k 3 + +Requires: FLEET_API_KEY env var (or --api-key) +""" + +import argparse +import os +import sys + +import requests + +try: + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + pass + +DEFAULT_BASE_URL = "https://orchestrator.fleetai.com" + +DEFAULT_MODELS = [ + "google/gemini-3.1-pro-preview", + "anthropic/claude-opus-4.6", + "openai/gpt-5.2", +] + + +def get_api_base() -> str: + base = os.environ.get("FLEET_BASE_URL", DEFAULT_BASE_URL).rstrip("/") + return f"{base}/v1" + + +def headers(api_key: str) -> dict: + return { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + +def resolve_team_id(api_key: str) -> str: + """GET /v1/account → team_id for this API key.""" + resp = requests.get(f"{get_api_base()}/account", headers=headers(api_key)) + if not resp.ok: + print(f" ERROR resolving team: {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + account = resp.json() + team_id = account["team_id"] + team_name = account.get("team_name", "unknown") + print(f" Authenticated as team: {team_name} ({team_id})") + return team_id + + +def verify_tasks_exist(api_key: str, task_keys: list[str]) -> None: + """Check that all task keys exist on the server.""" + print(f"\n1. Verifying {len(task_keys)} task(s) exist...") + for key in task_keys: + resp = requests.get( + f"{get_api_base()}/tasks/{key}", headers=headers(api_key) + ) + if resp.status_code == 404: + print(f" ERROR: Task '{key}' not found.") + sys.exit(1) + if not resp.ok: + print(f" ERROR checking task '{key}': {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + task = resp.json() + print(f" {key} (version: {task.get('version')}, env: {task.get('environment_id')})") + + +def launch_job( + api_key: str, task_keys: list[str], models: list[str], pass_k: int +) -> dict: + """POST /v1/jobs → launch a job for the given tasks.""" + print(f"\n2. Launching job...") + print(f" Tasks: {', '.join(task_keys)}") + print(f" Models: {', '.join(models)}") + print(f" pass_k: {pass_k}") + payload = { + "task_keys": task_keys, + "models": models, + "pass_k": pass_k, + } + resp = requests.post( + f"{get_api_base()}/jobs", + headers=headers(api_key), + json=payload, + ) + if not resp.ok: + print(f" ERROR {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + result = resp.json() + job_id = result.get("job_id", result.get("id", "N/A")) + status = result.get("status", "N/A") + print(f" Job launched: {job_id} (status: {status})") + return result + + +def main(): + parser = argparse.ArgumentParser( + description="Launch a job for existing task(s)" + ) + parser.add_argument( + "--task-key", + nargs="+", + required=True, + help="Task key(s) to launch a job for", + ) + parser.add_argument( + "--api-key", + default=os.environ.get("FLEET_API_KEY"), + help="API key (default: FLEET_API_KEY env var)", + ) + parser.add_argument( + "--models", + nargs="+", + default=DEFAULT_MODELS, + help=f"Models for the job (default: {' '.join(DEFAULT_MODELS)})", + ) + parser.add_argument( + "--pass-k", + type=int, + default=1, + help="pass_k for the job (default: 1)", + ) + args = parser.parse_args() + + if not args.api_key: + print("Error: FLEET_API_KEY env var or --api-key required") + sys.exit(1) + + resolve_team_id(args.api_key) + + # Verify all tasks exist before launching + verify_tasks_exist(args.api_key, args.task_key) + + # Launch the job + result = launch_job(args.api_key, args.task_key, args.models, args.pass_k) + + job_id = result.get("job_id", result.get("id", "N/A")) + print(f"\n-- Job launched --") + print(f" Job ID: {job_id}") + print(f" Status: {result.get('status', 'N/A')}") + if job_id != "N/A": + print(f" URL: https://fleetai.com/dashboard/jobs/{job_id}") + + +if __name__ == "__main__": + main() diff --git a/examples/task_bundle_editing/upload_task.py b/examples/task_bundle_editing/upload_task.py new file mode 100644 index 00000000..2c9e543d --- /dev/null +++ b/examples/task_bundle_editing/upload_task.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 +""" +Upload a task bundle (task.json + files/) as a new task, then launch a job. + +Usage: + # Auto-generate key, upload, and launch job: + python upload_task.py --dir ./my_task + + # Explicit key: + python upload_task.py --dir ./my_task --key my_custom_key + + # Skip job launch: + python upload_task.py --dir ./my_task --no-launch-job + + # Custom pass_k: + python upload_task.py --dir ./my_task --pass-k 3 + +Steps: + 1. Validates the bundle (same checks as validate_task.py) + 2. Checks task key doesn't already exist on server + 3. Uploads files (creates file-set, gets presigned upload URLs, POSTs files) + 4. Creates the task via POST /v1/tasks + 5. Launches a job via POST /v1/jobs (unless --no-launch-job) + +Requires: FLEET_API_KEY env var (or --api-key) +""" + +import argparse +import json +import os +import sys +import uuid +from pathlib import Path + +import requests + +try: + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + pass + +from validate_task import validate + +DEFAULT_BASE_URL = "https://orchestrator.fleetai.com" + +DEFAULT_MODELS = [ + "google/gemini-3.1-pro-preview", + "anthropic/claude-opus-4.6", + "openai/gpt-5.2", +] + + +def get_api_base() -> str: + base = os.environ.get("FLEET_BASE_URL", DEFAULT_BASE_URL).rstrip("/") + return f"{base}/v1" + + +def headers(api_key: str) -> dict: + return { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + +def resolve_team_id(api_key: str) -> str: + """GET /v1/account → team_id for this API key.""" + resp = requests.get(f"{get_api_base()}/account", headers=headers(api_key)) + if not resp.ok: + print(f" ERROR resolving team: {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + account = resp.json() + team_id = account["team_id"] + team_name = account.get("team_name", "unknown") + print(f" Authenticated as team: {team_name} ({team_id})") + return team_id + + +def check_task_exists(api_key: str, key: str) -> bool: + """GET /v1/tasks/{key} — returns True if the task already exists.""" + resp = requests.get(f"{get_api_base()}/tasks/{key}", headers=headers(api_key)) + if resp.status_code == 200: + return True + if resp.status_code == 404: + return False + # Unexpected status — surface the error rather than silently skipping the check + print(f" ERROR checking task existence: {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + return False # unreachable, but keeps the type checker happy + + +def upload_files( + api_key: str, + new_key: str, + files_dir: Path, + allow_overwrite: bool = False, +) -> None: + """Create file-set + upload files via presigned URLs.""" + all_files = [p for p in files_dir.rglob("*") if p.is_file()] + if not all_files: + print("\n3. No files to upload, skipping file-set creation.") + return + + filenames = [str(p.relative_to(files_dir)) for p in all_files] + print(f"\n3. Uploading {len(filenames)} files to file-set: {new_key}") + + # Create file-set + resp = requests.post( + f"{get_api_base()}/file-sets", + headers=headers(api_key), + json={"key": new_key, "description": f"Data files for task {new_key}"}, + ) + if resp.status_code == 409: + print(f" File-set '{new_key}' already exists, reusing.") + elif not resp.ok: + print(f" ERROR {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + else: + print(f" Created file-set: {new_key}") + + # Get upload URLs + resp = requests.post( + f"{get_api_base()}/file-sets/{new_key}/upload-urls", + headers=headers(api_key), + json={"filenames": filenames, "expires_in": 3600}, + params={"allow_overwrite": str(allow_overwrite).lower()}, + ) + if resp.status_code == 409: + detail = resp.json().get("detail", {}) + existing = detail.get("existing_files", []) + print(f"\n ERROR: {len(existing)} file(s) already exist in S3:") + for f in existing: + print(f" - {f}") + print(f"\n Use --allow-overwrite to replace them.") + sys.exit(1) + resp.raise_for_status() + upload_data = resp.json() + + # Upload each file via presigned POST + for item in upload_data["urls"]: + filename = item["filename"] + local_path = files_dir / filename + print(f" Uploading: {filename}") + with open(local_path, "rb") as fh: + upload_resp = requests.post( + item["url"], + data=item["fields"], + files={"file": fh}, + ) + upload_resp.raise_for_status() + print(f" -> uploaded ({local_path.stat().st_size} bytes)") + + +def upload_task(api_key: str, task: dict, new_key: str) -> dict: + """POST /v1/tasks → create task with new key.""" + print(f"\n4. Creating task with key: {new_key}") + payload = { + "key": new_key, + "prompt": task["prompt"], + "env_id": task["environment_id"], + "version": task.get("version"), + "env_variables": {"TASK_KEY": new_key}, + "metadata": task.get("metadata"), + "data_id": task.get("data_id"), + "data_version": task.get("data_version"), + } + # Include verifier code if present + verifier = task.get("verifier") + if verifier and verifier.get("code"): + payload["verifier_func"] = verifier["code"] + + resp = requests.post( + f"{get_api_base()}/tasks", + headers=headers(api_key), + json=payload, + ) + if not resp.ok: + print(f" ERROR {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + result = resp.json() + print(f" Created task: {result['key']}") + return result + + +def launch_job(api_key: str, task_key: str, models: list[str], pass_k: int) -> dict: + """POST /v1/jobs → launch a job for the task.""" + print(f"\n5. Launching job for task: {task_key}") + print(f" Models: {', '.join(models)}") + print(f" pass_k: {pass_k}") + payload = { + "task_keys": [task_key], + "models": models, + "pass_k": pass_k, + } + resp = requests.post( + f"{get_api_base()}/jobs", + headers=headers(api_key), + json=payload, + ) + if not resp.ok: + print(f" ERROR {resp.status_code}: {resp.text[:500]}") + resp.raise_for_status() + result = resp.json() + job_id = result.get("job_id", result.get("id", "N/A")) + status = result.get("status", "N/A") + print(f" Job launched: {job_id} (status: {status})") + if job_id != "N/A": + print(f" Dashboard: https://fleetai.com/dashboard/jobs/{job_id}") + return result + + +def main(): + parser = argparse.ArgumentParser( + description="Upload a task bundle as a new task and launch a job" + ) + parser.add_argument( + "--dir", required=True, help="Path to the task bundle directory" + ) + parser.add_argument( + "--key", + help="New task key (default: auto-generated from task.json key + UUID)", + ) + parser.add_argument( + "--api-key", + default=os.environ.get("FLEET_API_KEY"), + help="API key (default: FLEET_API_KEY env var)", + ) + parser.add_argument( + "--team-id", + help="Team ID override (default: auto-resolved from API key)", + ) + parser.add_argument( + "--allow-overwrite", + action="store_true", + help="Allow overwriting existing files in S3", + ) + parser.add_argument( + "--no-launch-job", + action="store_true", + help="Skip launching a job after upload", + ) + parser.add_argument( + "--models", + nargs="+", + default=DEFAULT_MODELS, + help=f"Models for the job (default: {' '.join(DEFAULT_MODELS)})", + ) + parser.add_argument( + "--pass-k", + type=int, + default=1, + help="pass_k for the job (default: 1)", + ) + args = parser.parse_args() + + if not args.api_key: + print("Error: FLEET_API_KEY env var or --api-key required") + sys.exit(1) + + # Resolve team_id from API key unless overridden + team_id = args.team_id or resolve_team_id(args.api_key) + + bundle_dir = Path(args.dir) + if not bundle_dir.is_dir(): + print(f"Error: {bundle_dir} is not a directory") + sys.exit(1) + + # Load task.json + task_path = bundle_dir / "task.json" + if not task_path.exists(): + print(f"Error: {task_path} not found") + sys.exit(1) + + task = json.loads(task_path.read_text()) + original_key = task.get("key", "") + + # Derive key if not provided + if args.key: + new_key = args.key + else: + suffix = uuid.uuid4().hex[:8] + new_key = f"{original_key}_{suffix}" + print(f"Auto-generated task key: {new_key}") + + # Step 1: Validate + print("1. Validating bundle...") + errors = validate(bundle_dir, new_key=new_key) + if errors: + print("Bundle validation failed. Fix errors above before uploading.") + sys.exit(1) + + # Step 2: Check task doesn't already exist on server + print(f"\n2. Checking if task key '{new_key}' already exists...") + if check_task_exists(args.api_key, new_key): + print( + f"Error: Task with key '{new_key}' already exists. " + "Use a different --key." + ) + sys.exit(1) + print(f" Key '{new_key}' is available.") + + files_dir = bundle_dir / "files" + + # Step 3: Upload files first (so a failure here doesn't leave a half-created task) + if files_dir.exists(): + upload_files(args.api_key, new_key, files_dir, allow_overwrite=args.allow_overwrite) + else: + print("\n3. No files/ directory, skipping file-set creation.") + + # Step 4: Create the task + result = upload_task(args.api_key, task, new_key) + + print(f"\n-- Upload complete --") + print(f" Original key: {original_key}") + print(f" New key: {new_key}") + print(f" Task ID: {result.get('id', 'N/A')}") + + # Step 5: Launch job (unless --no-launch-job) + if not args.no_launch_job: + job_result = launch_job(args.api_key, new_key, args.models, args.pass_k) + job_id = job_result.get("job_id", job_result.get("id", "N/A")) + print(f"\n-- Job launched --") + print(f" Job ID: {job_id}") + print(f" Status: {job_result.get('status', 'N/A')}") + if job_id != "N/A": + print(f" URL: https://fleetai.com/dashboard/jobs/{job_id}") + else: + print("\n Skipping job launch (--no-launch-job)") + + +if __name__ == "__main__": + main() diff --git a/examples/task_bundle_editing/validate_task.py b/examples/task_bundle_editing/validate_task.py new file mode 100644 index 00000000..3dc1928c --- /dev/null +++ b/examples/task_bundle_editing/validate_task.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +""" +Validate a downloaded task bundle before uploading. + +Checks: + 1. task.json exists and is valid JSON + 2. Required fields present and non-empty + 3. Verifier code is syntactically valid Python + 4. Verifier uses expected function signature + 5. files/ directory exists (may be empty) + 6. No files exceed size limit + 7. Data files are under files/notebooks/ (agent workspace) and match + the list_workspace_files() pattern in the prompt + 8. Key format is valid + +Usage: + python validate_task.py ./my_task + python validate_task.py ./my_task --new-key my_new_key +""" + +import argparse +import ast +import glob +import json +import re +import sys +from pathlib import Path + +MAX_FILE_SIZE_MB = 50 +MAX_TOTAL_SIZE_MB = 200 + + +def validate(bundle_dir: Path, new_key: str | None = None) -> list[str]: + """Validate a task bundle directory. Returns list of errors (empty = valid).""" + errors = [] + warnings = [] + + # -- 1. task.json existence and parse -- + task_path = bundle_dir / "task.json" + if not task_path.exists(): + errors.append("task.json not found") + return errors + + try: + task = json.loads(task_path.read_text()) + except json.JSONDecodeError as e: + errors.append(f"task.json is not valid JSON: {e}") + return errors + + # -- 2. Required fields -- + required_fields = { + "key": str, + "prompt": str, + "environment_id": str, + } + for field, expected_type in required_fields.items(): + val = task.get(field) + if val is None: + errors.append(f"Missing required field: {field}") + elif not isinstance(val, expected_type): + errors.append( + f"Field '{field}' should be {expected_type.__name__}, " + f"got {type(val).__name__}" + ) + elif isinstance(val, str) and not val.strip(): + errors.append(f"Field '{field}' is empty") + + # Check prompt length + prompt = task.get("prompt", "") + if isinstance(prompt, str): + if len(prompt) < 20: + errors.append(f"Prompt suspiciously short ({len(prompt)} chars)") + elif len(prompt) > 50000: + warnings.append(f"Prompt very long ({len(prompt)} chars)") + + # Optional but expected fields + if not task.get("version"): + warnings.append("No 'version' specified (will use latest)") + + # -- 3. env_variables -- + env_vars = task.get("env_variables") + if env_vars: + task_key_var = env_vars.get("TASK_KEY") + if task_key_var and new_key and task_key_var != new_key: + warnings.append( + f"env_variables.TASK_KEY='{task_key_var}' doesn't match " + f"new key '{new_key}' (upload_task will override this)" + ) + + # -- 4. Verifier code -- + verifier = task.get("verifier") + if verifier: + code = verifier.get("code") + if not code: + warnings.append("Verifier present but has no code") + else: + # Check syntax + try: + tree = ast.parse(code) + except SyntaxError as e: + errors.append(f"Verifier code has syntax error: {e}") + tree = None + + if tree: + # Check for expected function + func_names = [ + node.name + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) + ] + valid_names = {"verify", "verifier", "grade"} + if not any(name in valid_names for name in func_names): + warnings.append( + f"Verifier defines functions {func_names}, " + f"expected one of {valid_names}" + ) + + # Check function signature has 'env' param + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name in valid_names: + arg_names = [a.arg for a in node.args.args] + if "env" not in arg_names: + errors.append( + f"Verifier function '{node.name}' missing 'env' parameter" + ) + + # -- 4b. Verify S3 solution paths reference the correct task key -- + # Verifiers often load gold-reference images via Image.s3() with + # URLs like .../<TASK_KEY>/solutions/gold_plot.png. The TASK_KEY + # env variable must appear as a path segment in every such URL, + # otherwise the verifier will silently load the wrong solutions. + task_key_var = (task.get("env_variables") or {}).get("TASK_KEY") + if code and task_key_var: + expected_key = task_key_var + s3_urls = re.findall( + r'https?://[^"\']+\.s3[^"\']*\.amazonaws\.com/[^"\'\s]+', + code, + ) + solutions_urls = [u for u in s3_urls if "/solutions/" in u] + for url in solutions_urls: + path_segments = url.split("/") + if expected_key not in path_segments: + errors.append( + f"Verifier S3 solutions path does not contain " + f"expected key '{expected_key}' as a path segment: " + f"{url}" + ) + else: + warnings.append("No verifier in task.json") + + # -- 5. Files directory -- + files_dir = bundle_dir / "files" + if not files_dir.exists(): + warnings.append("No 'files/' directory (task has no data files)") + all_files = [] + else: + all_files = [p for p in files_dir.rglob("*") if p.is_file()] + if not all_files: + warnings.append("files/ directory exists but is empty") + + # -- 6. File sizes -- + total_size = 0 + for f in all_files: + size = f.stat().st_size + total_size += size + size_mb = size / (1024 * 1024) + if size_mb > MAX_FILE_SIZE_MB: + errors.append( + f"File too large ({size_mb:.1f} MB): {f.relative_to(bundle_dir)}" + ) + + total_mb = total_size / (1024 * 1024) + if total_mb > MAX_TOTAL_SIZE_MB: + errors.append( + f"Total file size {total_mb:.1f} MB exceeds {MAX_TOTAL_SIZE_MB} MB limit" + ) + + # -- 7. Data file location checks -- + # In Carlisle, files under files/notebooks/ are unpacked to /app/workspace/ + # at startup. The prompt tells agents to use list_workspace_files(pattern=...) + # to find data. Warn if the prompt references a path that doesn't match + # any files, or if data files are placed outside the notebooks/ tree. + if all_files: + notebooks_dir = files_dir / "notebooks" + + # Check that data files live under files/notebooks/ (the only path + # that gets unpacked into the agent workspace) + files_outside_notebooks = [ + f for f in all_files + if not str(f.relative_to(files_dir)).startswith("notebooks/") + and not str(f.relative_to(files_dir)).startswith("solutions/") + ] + if files_outside_notebooks: + warnings.append( + f"{len(files_outside_notebooks)} file(s) outside notebooks/ and " + f"solutions/ — these won't be visible in the agent workspace. " + f"e.g. {files_outside_notebooks[0].relative_to(files_dir)}" + ) + + # Extract list_workspace_files(pattern="...") from the prompt and + # verify matching files exist under files/notebooks/ + workspace_patterns = re.findall( + r'list_workspace_files\(pattern=["\']([^"\']+)["\']\)', prompt + ) + for wp in workspace_patterns: + # The agent sees /app/workspace/{wp}, which maps to + # files/notebooks/{wp} in the bundle + expected_glob = str(notebooks_dir / wp) + matches = glob.glob(expected_glob, recursive=True) + matches = [m for m in matches if Path(m).is_file()] + if not matches: + errors.append( + f"Prompt references list_workspace_files(pattern=\"{wp}\") " + f"but no files match files/notebooks/{wp}" + ) + else: + # Informational — not a warning, just context for the report + pass + + if not workspace_patterns and notebooks_dir.exists(): + warnings.append( + "Prompt has no list_workspace_files() call — agent may not " + "know how to find the data files" + ) + + # -- 8. Key format -- + key = new_key or task.get("key", "") + if key: + if " " in key: + errors.append(f"Key contains spaces: '{key}'") + if len(key) > 200: + errors.append(f"Key too long ({len(key)} chars, max 200)") + + # -- Print results -- + print(f"\n{'=' * 60}") + print(f" Task Bundle Validation: {bundle_dir.name}") + print(f"{'=' * 60}") + print(f" Key: {task.get('key', 'N/A')}") + if new_key: + print(f" New key: {new_key}") + print(f" Environment: {task.get('environment_id', 'N/A')}") + print(f" Version: {task.get('version', 'N/A')}") + print(f" Prompt: {len(task.get('prompt', ''))} chars") + print(f" Verifier: {'yes' if verifier and verifier.get('code') else 'no'}") + print(f" Files: {len(all_files)} ({total_mb:.2f} MB total)") + if all_files: + for f in all_files: + rel = f.relative_to(files_dir) + size_kb = f.stat().st_size / 1024 + print(f" - {rel} ({size_kb:.0f} KB)") + + if warnings: + print(f"\n Warnings ({len(warnings)}):") + for w in warnings: + print(f" [!] {w}") + + if errors: + print(f"\n ERRORS ({len(errors)}):") + for e in errors: + print(f" [X] {e}") + print(f"\n RESULT: FAIL") + else: + print(f"\n RESULT: PASS") + + print(f"{'=' * 60}\n") + return errors + + +def main(): + parser = argparse.ArgumentParser( + description="Validate a task bundle before upload" + ) + parser.add_argument("bundle_dir", help="Path to the task bundle directory") + parser.add_argument( + "--new-key", help="New key to use (checks consistency)" + ) + args = parser.parse_args() + + bundle_dir = Path(args.bundle_dir) + if not bundle_dir.is_dir(): + print(f"Error: {bundle_dir} is not a directory") + sys.exit(1) + + errors = validate(bundle_dir, new_key=args.new_key) + sys.exit(1 if errors else 0) + + +if __name__ == "__main__": + main() diff --git a/fleet/_async/__init__.py b/fleet/_async/__init__.py index 529934da..4203c472 100644 --- a/fleet/_async/__init__.py +++ b/fleet/_async/__init__.py @@ -215,7 +215,7 @@ async def import_task(task, project_key: Optional[str] = None): project_key: Optional project key to associate with the task Example: - task = fleet.Task(key="my-task", prompt="Do something", env_id="my-env") + task = fleet.Task(key="my-task", prompt="Do something", env_key="my-env") await fleet.import_task(task) await fleet.import_task(task, project_key="my-project") """ diff --git a/fleet/_async/client.py b/fleet/_async/client.py index 93eac32f..69fd3f63 100644 --- a/fleet/_async/client.py +++ b/fleet/_async/client.py @@ -50,6 +50,11 @@ JobSessionsResponse, SessionTranscriptResponse, ) +from .models import ( + ScenarioResponse, + ScenarioListResponse, + ScenarioCreateRequest, +) from .tasks import Task if TYPE_CHECKING: @@ -586,7 +591,7 @@ async def make( return instance async def make_for_task(self, task: Task) -> AsyncEnv: - return await self.make(env_key=f"{task.env_id}:{task.version}") + return await self.make(env_key=task.env_spec) async def instances( self, status: Optional[str] = None, region: Optional[str] = None, run_id: Optional[str] = None, profile_id: Optional[str] = None @@ -927,9 +932,7 @@ async def load_task_from_json( task = Task( key=task_json.get("key", task_json.get("id")), prompt=task_json["prompt"], - env_id=task_json.get( - "env_id", task_json.get("env_key") - ), # Use env_id or fallback to env_key + env_key=task_json.get("env_key") or task_json.get("environment_id") or task_json.get("env_id"), # Prefer env_key, fallback to environment_id (API) or env_id (legacy) created_at=task_json.get("created_at"), version=task_json.get("version"), data_id=task_json.get("data_id"), @@ -942,6 +945,10 @@ async def load_task_from_json( verifier_runtime_version=verifier_runtime_version, # Set verifier_runtime_version metadata=task_json.get("metadata", {}), # Default empty metadata output_json_schema=task_json.get("output_json_schema"), # JSON schema for output + factual_answer=task_json.get("factual_answer"), # Expected answer for research/factual tasks + task_modality=task_json.get("task_modality"), # Task modality (computer_use, tool_use, browser) + task_scenario_id=task_json.get("task_scenario_id"), # Task scenario ID + task_lifecycle_status=task_json.get("task_lifecycle_status"), # Task lifecycle status ) return task @@ -1103,7 +1110,7 @@ async def create_verifier_with_fallback(tr, emb_code, is_error): task = Task( key=task_response.key, prompt=task_response.prompt, - env_id=task_response.environment_id, # Map environment_id -> env_id + env_key=task_response.env_key, created_at=task_response.created_at, version=task_response.version, data_id=getattr(task_response, "data_id", None), # Get data_id if available @@ -1116,6 +1123,10 @@ async def create_verifier_with_fallback(tr, emb_code, is_error): verifier_runtime_version=verifier_runtime_version, # Set verifier_runtime_version metadata=metadata, output_json_schema=getattr(task_response, "output_json_schema", None), # Get output_json_schema if available + factual_answer=getattr(task_response, "factual_answer", None), # Get factual_answer if available + task_modality=getattr(task_response, "task_modality", None), # Get task_modality if available + task_scenario_id=getattr(task_response, "task_scenario_id", None), # Get task_scenario_id if available + task_lifecycle_status=getattr(task_response, "task_lifecycle_status", None), # Get task_lifecycle_status if available ) tasks.append(task) @@ -1181,26 +1192,32 @@ async def import_single_task(self, task: Task, project_key: Optional[str] = None project_key: Optional project key to associate with the task Returns: - Response from the API, or None if the import failed - """ - try: - # Validate that verifier_func exists - if not task.verifier_func: - raise ValueError( - f"Task {task.key} is missing verifier_func. " - "All tasks must have a verifier_func to be imported." - ) + Response from the API - params = {} - if project_key: - params["project_key"] = project_key - response = await self.client.request( - "POST", "/v1/tasks", json=task.model_dump(), params=params + Raises: + ValueError: If task is missing verifier_func + Exception: If the API request fails + """ + # Validate that verifier_func exists + if not task.verifier_func: + raise ValueError( + f"Task {task.key} is missing verifier_func. " + "All tasks must have a verifier_func to be imported." ) - return response - except Exception as e: - # logger.error(f"Failed to import task {task.key}: {e}") - return None + + params = {} + if project_key: + params["project_key"] = project_key + + # Convert task to dict and map env_key -> env_id for API compatibility + task_data = task.model_dump() + if "env_key" in task_data: + task_data["env_id"] = task_data.pop("env_key") + + response = await self.client.request( + "POST", "/v1/tasks", json=task_data, params=params + ) + return response async def import_tasks(self, filename: str, project_key: Optional[str] = None): """Import tasks from a JSON file. @@ -1255,6 +1272,76 @@ async def import_with_semaphore(task): # Filter out None values (failed imports) return [r for r in responses if r is not None] + # Scenario methods + async def get_scenario(self, scenario_id: str) -> ScenarioResponse: + """Get a scenario by ID. + + Args: + scenario_id: The scenario ID to retrieve + + Returns: + ScenarioResponse containing the scenario details + """ + response = await self.client.request("GET", f"/v1/scenarios/{scenario_id}") + return ScenarioResponse(**response.json()) + + async def list_scenarios(self, project_id: Optional[str] = None) -> ScenarioListResponse: + """List scenarios, optionally filtered by project. + + Args: + project_id: Optional project ID to filter scenarios by + + Returns: + ScenarioListResponse containing the list of scenarios + """ + params = {} + if project_id: + params["project_id"] = project_id + response = await self.client.request("GET", "/v1/scenarios", params=params) + return ScenarioListResponse(**response.json()) + + async def create_scenario( + self, + scenario_title: str, + project_id: str, + output_json_schema: Optional[Dict[str, Any]] = None, + task_complexity_tier: Optional[str] = None, + user_story: Optional[str] = None, + entrypoint: Optional[str] = None, + is_research_based: Optional[bool] = None, + is_action_based: Optional[bool] = None, + scenario_verifier_prompt: Optional[str] = None, + ) -> ScenarioResponse: + """Create a new scenario. + + Args: + scenario_title: Title for the scenario + project_id: Project ID to associate the scenario with + output_json_schema: Optional JSON schema for expected output + task_complexity_tier: Optional complexity tier + user_story: Optional user story description + entrypoint: Optional entrypoint URL + is_research_based: Whether the scenario is research-based + is_action_based: Whether the scenario is action-based + scenario_verifier_prompt: Optional verifier prompt + + Returns: + ScenarioResponse containing the created scenario + """ + request = ScenarioCreateRequest( + scenario_title=scenario_title, + project_id=project_id, + output_json_schema=output_json_schema, + task_complexity_tier=task_complexity_tier, + user_story=user_story, + entrypoint=entrypoint, + is_research_based=is_research_based, + is_action_based=is_action_based, + scenario_verifier_prompt=scenario_verifier_prompt, + ) + response = await self.client.request("POST", "/v1/scenarios", json=request.model_dump(exclude_none=True)) + return ScenarioResponse(**response.json()) + async def account(self) -> AccountResponse: """Get account information including instance limits and usage. @@ -1270,6 +1357,8 @@ async def update_task( prompt: Optional[str] = None, verifier_code: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, + writer_metadata: Optional[Dict[str, Any]] = None, + qa_metadata: Optional[Dict[str, Any]] = None, ) -> TaskResponse: """Update an existing task. @@ -1278,11 +1367,19 @@ async def update_task( prompt: New prompt text for the task (optional) verifier_code: Python code for task verification (optional) metadata: Additional metadata for the task (optional) + writer_metadata: Metadata filled by task writer (optional) + qa_metadata: Metadata filled by QA reviewer (optional) Returns: TaskResponse containing the updated task details """ - payload = TaskUpdateRequest(prompt=prompt, verifier_code=verifier_code, metadata=metadata) + payload = TaskUpdateRequest( + prompt=prompt, + verifier_code=verifier_code, + metadata=metadata, + writer_metadata=writer_metadata, + qa_metadata=qa_metadata, + ) response = await self.client.request( "PUT", f"/v1/tasks/{task_key}", json=payload.model_dump(exclude_none=True) ) diff --git a/fleet/_async/models.py b/fleet/_async/models.py index 2159b932..b76a961e 100644 --- a/fleet/_async/models.py +++ b/fleet/_async/models.py @@ -7,7 +7,7 @@ from enum import Enum from typing import Any, Dict, List, Optional, Union -from pydantic import BaseModel, Field, conint +from pydantic import BaseModel, ConfigDict, Field, conint class CDPDescribeResponse(BaseModel): @@ -150,13 +150,18 @@ class TableSchema(BaseModel): class TaskRequest(BaseModel): + model_config = ConfigDict(populate_by_name=True) + key: str = Field(..., title="Key") prompt: str = Field(..., title="Prompt") - environment_id: str = Field(..., title="Environment Id") + env_key: str = Field(..., title="Env Key", alias="environment_id") verifier_id: Optional[str] = Field(None, title="Verifier Id") version: Optional[str] = Field(None, title="Version") env_variables: Optional[Dict[str, Any]] = Field(None, title="Env Variables") metadata: Optional[Dict[str, Any]] = Field(None, title="Metadata") + writer_metadata: Optional[Dict[str, Any]] = Field( + None, title="Writer Metadata", description="Metadata filled by task writer" + ) output_json_schema: Optional[Dict[str, Any]] = Field(None, title="Output Json Schema") @@ -164,6 +169,12 @@ class TaskUpdateRequest(BaseModel): prompt: Optional[str] = Field(None, title="Prompt") verifier_code: Optional[str] = Field(None, title="Verifier Code") metadata: Optional[Dict[str, Any]] = Field(None, title="Metadata") + writer_metadata: Optional[Dict[str, Any]] = Field( + None, title="Writer Metadata", description="Metadata filled by task writer" + ) + qa_metadata: Optional[Dict[str, Any]] = Field( + None, title="QA Metadata", description="Metadata filled by QA reviewer" + ) class VerifierData(BaseModel): @@ -177,10 +188,12 @@ class VerifierData(BaseModel): class TaskResponse(BaseModel): + model_config = ConfigDict(populate_by_name=True) + key: str = Field(..., title="Key") prompt: str = Field(..., title="Prompt") team_id: str = Field(..., title="Team Id") - environment_id: str = Field(..., title="Environment Id") + env_key: str = Field(..., title="Env Key", alias="environment_id") created_at: str = Field(..., title="Created At") verifier_id: Optional[str] = Field(None, title="Verifier Id") verifier_func: Optional[str] = Field(None, title="Verifier Func") @@ -191,6 +204,45 @@ class TaskResponse(BaseModel): verifier: Optional[VerifierData] = Field(None, title="Verifier") metadata: Optional[Dict[str, Any]] = Field(None, title="Metadata") output_json_schema: Optional[Dict[str, Any]] = Field(None, title="Output Json Schema") + task_modality: Optional[str] = Field(None, title="Task Modality", description="Task modality (computer_use, tool_use, browser)") + factual_answer: Optional[Any] = Field(None, title="Factual Answer", description="Expected answer for research/factual tasks") + task_scenario_id: Optional[int] = Field(None, title="Task Scenario ID", description="ID of the task scenario this task belongs to") + task_lifecycle_status: Optional[str] = Field(None, title="Task Lifecycle Status", description="Task lifecycle status (production, development, staging, etc.)") + + +class ScenarioResponse(BaseModel): + """Response model for a task scenario.""" + + id: int = Field(..., title="ID", description="Scenario ID") + scenario_title: Optional[str] = Field(None, title="Scenario Title") + output_json_schema: Optional[Dict[str, Any]] = Field(None, title="Output JSON Schema") + task_complexity_tier: Optional[str] = Field(None, title="Task Complexity Tier") + user_story: Optional[str] = Field(None, title="User Story") + entrypoint: Optional[str] = Field(None, title="Entrypoint") + is_research_based: Optional[bool] = Field(None, title="Is Research Based") + is_action_based: Optional[bool] = Field(None, title="Is Action Based") + scenario_verifier_prompt: Optional[str] = Field(None, title="Scenario Verifier Prompt") + + +class ScenarioListResponse(BaseModel): + """Response model for listing scenarios.""" + + scenarios: List["ScenarioResponse"] = Field(..., title="Scenarios") + total: int = Field(..., title="Total") + + +class ScenarioCreateRequest(BaseModel): + """Request model for creating a scenario.""" + + scenario_title: str = Field(..., title="Scenario Title") + project_id: str = Field(..., title="Project ID", description="Project to associate the scenario with") + output_json_schema: Optional[Dict[str, Any]] = Field(None, title="Output JSON Schema") + task_complexity_tier: Optional[str] = Field(None, title="Task Complexity Tier") + user_story: Optional[str] = Field(None, title="User Story") + entrypoint: Optional[str] = Field(None, title="Entrypoint") + is_research_based: Optional[bool] = Field(None, title="Is Research Based") + is_action_based: Optional[bool] = Field(None, title="Is Action Based") + scenario_verifier_prompt: Optional[str] = Field(None, title="Scenario Verifier Prompt") class ValidationError(BaseModel): diff --git a/fleet/_async/tasks.py b/fleet/_async/tasks.py index aa7190ac..da527ee5 100644 --- a/fleet/_async/tasks.py +++ b/fleet/_async/tasks.py @@ -19,7 +19,7 @@ class Task(BaseModel): key: str = Field(..., description="Unique task key identifier") prompt: str = Field(..., description="Task prompt or instruction") - env_id: str = Field(..., description="Environment identifier") + env_key: str = Field(..., description="Environment key") env_variables: Optional[Dict[str, Any]] = Field( default_factory=dict, description="Environment variables" ) @@ -41,6 +41,18 @@ class Task(BaseModel): output_json_schema: Optional[Dict[str, Any]] = Field( None, description="JSON schema for expected output format" ) + task_modality: Optional[str] = Field( + None, description="Task modality (computer_use, tool_use, browser)" + ) + factual_answer: Optional[Any] = Field( + None, description="Expected answer for research/factual tasks" + ) + task_scenario_id: Optional[int] = Field( + None, description="ID of the task scenario this task belongs to" + ) + task_lifecycle_status: Optional[str] = Field( + None, description="Task lifecycle status (production, development, staging, etc.)" + ) @validator("key") def validate_key_format(cls, v): @@ -57,21 +69,41 @@ def serialize_created_at(self, dt: Optional[datetime], _info): return dt.isoformat() if dt else None @property - def env_key(self) -> str: - """Get the environment key combining env_id and version.""" - if self.version and self.version != "None" and ":" not in self.env_id: - return f"{self.env_id}:{self.version}" - return self.env_id + def env_spec(self) -> str: + """Get the env spec string (env_key:version) for instance creation.""" + if self.version and self.version != "None" and ":" not in self.env_key: + return f"{self.env_key}:{self.version}" + return self.env_key @property - def data_key(self) -> Optional[str]: - """Get the data key combining data_id and data_version.""" + def data_spec(self) -> Optional[str]: + """Get the data spec string (data_id:data_version) for instance creation.""" if self.data_id and self.data_version: return f"{self.data_id}:{self.data_version}" elif self.data_id: return self.data_id return None + @property + def data_key(self) -> Optional[str]: + """Alias for data_spec for backward compatibility.""" + return self.data_spec + + @property + def has_verifier(self) -> bool: + """Whether this task has a verifier function.""" + return self.verifier is not None or self.verifier_func is not None + + @property + def is_research_based(self) -> bool: + """Whether this task is research/factual (has a factual_answer).""" + return self.factual_answer is not None + + @property + def is_action_based(self) -> bool: + """Whether this task is action-based (no factual_answer).""" + return self.factual_answer is None + class Config: """Pydantic model configuration.""" @@ -219,7 +251,7 @@ async def make_env( ): """Create an environment instance for this task's environment. - Alias for make() method. Uses the task's env_id (and version if present) to create the env. + Alias for make() method. Uses the task's env_key (and version if present) to create the env. """ return await self.make( region=region, @@ -240,7 +272,7 @@ async def make( """Create an environment instance with task's configuration. Auto-populates environment creation with: - - env_key (env_id + version) + - env_key (env_key + version) - data_key (data_id + data_version, if present) - env_variables (if present) - run_id (if present) @@ -257,18 +289,18 @@ async def make( Environment instance configured for this task Example: - task = fleet.Task(key="my-task", prompt="...", env_id="my-env", + task = fleet.Task(key="my-task", prompt="...", env_key="my-env", data_id="my-data", data_version="v1.0") env = await task.make(region="us-west-2", run_id="my-batch-123", heartbeat_interval=60) """ - if not self.env_id: - raise ValueError("Task has no env_id defined") + if not self.env_key: + raise ValueError("Task has no env_key defined") # Deferred import to avoid circular dependencies from fleet.env import make_async return await make_async( - env_key=self.env_key, + env_key=self.env_spec, data_key=self.data_key, region=region, env_variables=self.env_variables if self.env_variables else None, @@ -346,8 +378,9 @@ def contains_number(text: str, target_number) -> bool: return False return target in numbers - # Create a local namespace for executing the code - local_namespace = { + # Create a globals namespace with all required imports + exec_globals = globals().copy() + exec_globals.update({ "TASK_SUCCESSFUL_SCORE": TASK_SUCCESSFUL_SCORE, "TASK_FAILED_SCORE": TASK_FAILED_SCORE, "IgnoreConfig": IgnoreConfig, @@ -358,10 +391,17 @@ def contains_number(text: str, target_number) -> bool: "json": json, "re": re, "string": string, - } + }) + + # Create a local namespace for executing the code + local_namespace = {} # Execute the cleaned verifier code in the namespace - exec(cleaned_code, globals(), local_namespace) + exec(cleaned_code, exec_globals, local_namespace) + + # Merge local_namespace into exec_globals so helper functions are accessible + # from the main verifier function when it's called + exec_globals.update(local_namespace) # Find the function that was defined (not imported) # Functions defined via exec have co_filename == '' @@ -448,7 +488,12 @@ async def load_tasks( async def update_task( - task_key: str, prompt: Optional[str] = None, verifier_code: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None + task_key: str, + prompt: Optional[str] = None, + verifier_code: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + writer_metadata: Optional[Dict[str, Any]] = None, + qa_metadata: Optional[Dict[str, Any]] = None, ): """Convenience function to update an existing task. @@ -457,6 +502,8 @@ async def update_task( prompt: New prompt text for the task (optional) verifier_code: Python code for task verification (optional) metadata: Additional metadata for the task (optional) + writer_metadata: Metadata filled by task writer (optional) + qa_metadata: Metadata filled by QA reviewer (optional) Returns: TaskResponse containing the updated task details @@ -465,12 +512,18 @@ async def update_task( response = await fleet.update_task("my-task", prompt="New prompt text") response = await fleet.update_task("my-task", verifier_code="def verify(env): return True") response = await fleet.update_task("my-task", metadata={"seed": 42, "story": "Updated story"}) + response = await fleet.update_task("my-task", writer_metadata={"author": "john"}) """ from .global_client import get_client client = get_client() return await client.update_task( - task_key=task_key, prompt=prompt, verifier_code=verifier_code, metadata=metadata + task_key=task_key, + prompt=prompt, + verifier_code=verifier_code, + metadata=metadata, + writer_metadata=writer_metadata, + qa_metadata=qa_metadata, ) @@ -511,7 +564,7 @@ async def import_task(task: Task, project_key: Optional[str] = None): Response from the API, or None if the import failed Examples: - task = fleet.Task(key="my-task", prompt="Do something", env_id="my-env") + task = fleet.Task(key="my-task", prompt="Do something", env_key="my-env") response = await fleet.import_task(task) response = await fleet.import_task(task, project_key="my-project") """ diff --git a/fleet/agent/gemini_cua/agent.py b/fleet/agent/gemini_cua/agent.py index d0736741..b9fb0ad9 100644 --- a/fleet/agent/gemini_cua/agent.py +++ b/fleet/agent/gemini_cua/agent.py @@ -1,13 +1,13 @@ #!/usr/bin/env python3 """ -Gemini CUA Agent +Gemini CUA Agent (Standalone) Env vars: GEMINI_API_KEY: API key FLEET_MCP_URL: CUA server URL (http://localhost:PORT) FLEET_TASK_PROMPT: Task prompt FLEET_TASK_KEY: Task key - FLEET_MODEL: Model (default: gemini-2.5-pro) + FLEET_MODEL: Model (default: gemini-3-pro-preview) FLEET_MAX_STEPS: Max steps (default: 200) FLEET_VERBOSE: Enable verbose logging (default: false) USE_OAUTH: Use gcloud OAuth instead of API key (default: false) @@ -21,10 +21,13 @@ import sys import time from typing import Any, Dict, List, Optional -from mcp import ClientSession -from mcp.client.streamable_http import streamable_http_client + +from google.genai.types import Content, Part from google import genai from google.genai import types +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client + import fleet from fleet.utils.logging import log_verbose, VERBOSE @@ -38,6 +41,123 @@ GOOG_PROJECT = os.environ.get("GOOG_PROJECT", "gemini-agents-area") USE_OAUTH = os.environ.get("USE_OAUTH", "false").lower() in ("true", "1", "yes") +# Screen dimensions for coordinate denormalization (matches MCP browser) +SCREEN_WIDTH = 1366 +SCREEN_HEIGHT = 768 + +# Gemini 3 tool definitions (0-1000 normalized coordinates) +GEMINI_3_TOOL_DEFINITIONS = [ + { + "name": "click_at", + "description": "Click at the specified screen coordinates. Coordinates are normalized 0-1000.", + "parameters": { + "type": "object", + "properties": { + "x": { + "type": "integer", + "description": "X coordinate (0-1000, where 0 is left edge, 1000 is right edge)", + }, + "y": { + "type": "integer", + "description": "Y coordinate (0-1000, where 0 is top edge, 1000 is bottom edge)", + }, + }, + "required": ["x", "y"], + }, + }, + { + "name": "type_text", + "description": "Type text at the current cursor position. Use click_at first to focus the input field.", + "parameters": { + "type": "object", + "properties": { + "text": { + "type": "string", + "description": "The text to type", + }, + "press_enter": { + "type": "boolean", + "description": "Whether to press Enter after typing (default: false)", + }, + }, + "required": ["text"], + }, + }, + { + "name": "key_press", + "description": "Press a key or key combination (e.g., 'Enter', 'Tab', 'Meta+A', 'Ctrl+C', 'Backspace').", + "parameters": { + "type": "object", + "properties": { + "keys": { + "type": "string", + "description": "Key or key combination to press", + }, + }, + "required": ["keys"], + }, + }, + { + "name": "scroll", + "description": "Scroll the page up or down.", + "parameters": { + "type": "object", + "properties": { + "direction": { + "type": "string", + "description": "Direction to scroll: 'up' or 'down'", + "enum": ["up", "down"], + }, + }, + "required": ["direction"], + }, + }, + { + "name": "wait", + "description": "Wait for a few seconds to allow page to load.", + "parameters": { + "type": "object", + "properties": { + "seconds": { + "type": "integer", + "description": "Number of seconds to wait (1-10)", + }, + }, + "required": ["seconds"], + }, + }, +] + +# Key name normalization for xdotool/X11 keysym compatibility +_KEY_NAME_MAP_LOWER = { + "backspace": "BackSpace", + "arrowleft": "Left", "arrowright": "Right", "arrowup": "Up", "arrowdown": "Down", + "left": "Left", "right": "Right", "up": "Up", "down": "Down", + "esc": "Escape", "escape": "Escape", + "del": "Delete", "delete": "Delete", + "pgup": "Page_Up", "pageup": "Page_Up", + "pgdown": "Page_Down", "pgdn": "Page_Down", "pagedown": "Page_Down", + "enter": "Return", "return": "Return", + "tab": "Tab", "space": "space", + "meta": "super", "command": "super", "cmd": "super", "super": "super", + "ctrl": "ctrl", "control": "ctrl", + "alt": "alt", "shift": "shift", + "f1": "F1", "f2": "F2", "f3": "F3", "f4": "F4", "f5": "F5", "f6": "F6", + "f7": "F7", "f8": "F8", "f9": "F9", "f10": "F10", "f11": "F11", "f12": "F12", + "home": "Home", "end": "End", "insert": "Insert", +} + + +def normalize_key_name(key: str) -> str: + """Normalize key names to xdotool/X11 keysym format.""" + if not key: + return key + if "+" in key: + parts = key.split("+") + normalized_parts = [_KEY_NAME_MAP_LOWER.get(p.lower(), p) for p in parts] + return "+".join(normalized_parts) + return _KEY_NAME_MAP_LOWER.get(key.lower(), key) + def get_oauth_token() -> str: """Get OAuth token from gcloud.""" @@ -53,11 +173,9 @@ def get_gemini_client() -> genai.Client: """Create Gemini client with appropriate auth.""" api_key = os.environ.get("GEMINI_API_KEY") custom_endpoint = os.environ.get("FLEET_MODEL_ENDPOINT") - - # Register endpoint for proxy whitelist + _register_endpoint(custom_endpoint or "generativelanguage.googleapis.com") - - # Build http_options + http_opts = None if USE_OAUTH or custom_endpoint: opts = {} @@ -72,29 +190,75 @@ def get_gemini_client() -> genai.Client: opts["api_version"] = "v1alpha" log_verbose(f"Using OAuth (project: {GOOG_PROJECT})") http_opts = types.HttpOptions(**opts) - + return genai.Client(api_key=api_key, http_options=http_opts) +def convert_gemini_3_to_mcp(function_name: str, args: Dict[str, Any]) -> List[Dict[str, Any]]: + """Convert Gemini 3 custom function calls to MCP computer tool format. + + Coordinates are normalized 0-1000, denormalized to screen dimensions. + Returns a list of MCP actions since some functions expand to multiple steps. + """ + def denormalize_x(x: int) -> int: + return int(x / 1000 * SCREEN_WIDTH) + + def denormalize_y(y: int) -> int: + return int(y / 1000 * SCREEN_HEIGHT) + + mcp_actions = [] + + if function_name == "click_at": + x = denormalize_x(args.get("x", 500)) + y = denormalize_y(args.get("y", 500)) + mcp_actions.append({"action": "left_click", "coordinate": [x, y]}) + + elif function_name == "type_text": + text = args.get("text", "") + press_enter = args.get("press_enter", False) + mcp_actions.append({"action": "type", "text": text}) + if press_enter: + mcp_actions.append({"action": "key", "text": "Return"}) + + elif function_name == "key_press": + keys = args.get("keys", "Return") + mcp_actions.append({"action": "key", "text": normalize_key_name(keys)}) + + elif function_name == "scroll": + direction = args.get("direction", "down") + mcp_actions.append({ + "action": "scroll", + "coordinate": [SCREEN_WIDTH // 2, SCREEN_HEIGHT // 2], + "scroll_direction": direction, + "scroll_amount": 5, + }) + + elif function_name == "wait": + seconds = min(args.get("seconds", 3), 10) + mcp_actions.append({"action": "wait", "duration": seconds}) + + else: + # Unknown function, fallback to screenshot + mcp_actions.append({"action": "screenshot"}) + + return mcp_actions + class MCP: """MCP client using streamable-http transport.""" - + def __init__(self, url: str, log_file: Optional[str] = None): - # Ensure URL ends with /mcp/ for streamable-http self.url = url.rstrip("/") + "/mcp/" self._session: Optional[ClientSession] = None self._client = None - self._tools: List[Dict] = [] self._log_file = log_file or os.environ.get("FLEET_SESSION_LOG") self._log_handle = None if self._log_file: from pathlib import Path Path(self._log_file).parent.mkdir(parents=True, exist_ok=True) self._log_handle = open(self._log_file, "a") - + async def __aenter__(self): - # Connect using streamable-http transport print(f"MCP: Connecting to {self.url}...") try: self._client = streamable_http_client(self.url) @@ -123,7 +287,7 @@ async def __aenter__(self): print(f"MCP: Failed to list tools: {type(e).__name__}: {e}") raise return self - + async def __aexit__(self, *args): if self._session: await self._session.__aexit__(*args) @@ -131,17 +295,16 @@ async def __aexit__(self, *args): await self._client.__aexit__(*args) if self._log_handle: self._log_handle.close() - + def _log(self, entry: dict): """Log an entry to the traffic file.""" if self._log_handle: - import json from datetime import datetime entry["timestamp"] = datetime.now().isoformat() entry["url"] = self.url self._log_handle.write(json.dumps(entry) + "\n") self._log_handle.flush() - + async def call(self, name: str, args: Dict = None) -> Dict: """Call a tool and return the result.""" start_time = time.time() @@ -163,8 +326,7 @@ def _get(item, key, default=None): if isinstance(item, dict): return item.get(key, default) return getattr(item, key, default) - - # Convert MCP result to dict format expected by agent + content = [] for item in result.content: item_type = _get(item, "type") @@ -176,8 +338,7 @@ def _get(item, key, default=None): }) elif item_type == "text": content.append({"type": "text", "text": _get(item, "text", "")}) - - # Log the call (just types, not data) + self._log({ "type": "mcp_call", "tool": name, @@ -187,22 +348,18 @@ def _get(item, key, default=None): "is_error": result.isError if hasattr(result, "isError") else False, }) return {"content": content, "isError": result.isError if hasattr(result, "isError") else False} - - def get_tools(self) -> List[Dict]: - """Return the list of tools from the server.""" - return self._tools -def mcp_tools_to_gemini(mcp_tools: List[Dict]) -> List[types.FunctionDeclaration]: - """Convert MCP tool definitions to Gemini FunctionDeclarations.""" - declarations = [] - for tool in mcp_tools: - declarations.append(types.FunctionDeclaration( +def get_gemini_3_tools() -> List[types.FunctionDeclaration]: + """Return Gemini 3 custom tools as FunctionDeclarations.""" + return [ + types.FunctionDeclaration( name=tool["name"], - description=tool.get("description", ""), - parameters=tool.get("inputSchema", {"type": "object", "properties": {}}), - )) - return declarations + description=tool["description"], + parameters=tool["parameters"], + ) + for tool in GEMINI_3_TOOL_DEFINITIONS + ] def get_image_data(result: Dict) -> Optional[str]: @@ -213,80 +370,126 @@ def get_image_data(result: Dict) -> Optional[str]: return None +def extract_reasoning_from_candidate(candidate) -> Optional[str]: + """Extract reasoning trace from Gemini candidate response.""" + reasoning_parts = [] + + if not candidate or not candidate.content or not candidate.content.parts: + return None + + has_function_calls = any( + hasattr(p, "function_call") and p.function_call for p in candidate.content.parts + ) + + for part in candidate.content.parts: + if hasattr(part, "thought") and part.thought: + if isinstance(part.thought, str): + reasoning_parts.append(part.thought) + elif part.thought is True and hasattr(part, "text") and part.text: + reasoning_parts.append(part.text) + elif hasattr(part, "text") and part.text and has_function_calls: + reasoning_parts.append(part.text) + + if not reasoning_parts: + return None + return "\n\n".join(reasoning_parts) + + class GeminiAgent: """Gemini Computer Use Agent.""" - + def __init__(self, mcp: MCP, model: str, session=None): self.mcp = mcp - # Strip provider prefix if present self.model = model.split("/")[-1] if "/" in model else model self.client = get_gemini_client() self.transcript: List[Dict] = [] - self.session = session # Fleet session for live logging + self.session = session self._consecutive_errors = 0 self._max_consecutive_errors = 5 - - async def _execute_tool(self, name: str, args: Dict) -> Dict: - return await self.mcp.call(name, args) - + + async def _take_screenshot(self) -> Optional[str]: + """Take a screenshot and return base64 data.""" + try: + result = await self.mcp.call("computer", {"action": "screenshot"}) + return get_image_data(result) + except Exception as e: + print(f"Screenshot failed: {e}") + return None + + async def _execute_gemini_function(self, name: str, args: Dict) -> Dict: + """Execute a Gemini function by converting to MCP actions.""" + mcp_actions = convert_gemini_3_to_mcp(name, args) + log_verbose(f" Converting {name} -> {len(mcp_actions)} MCP action(s)") + + last_result = None + for i, action in enumerate(mcp_actions): + log_verbose(f" Action {i+1}: {action}") + last_result = await self.mcp.call("computer", action) + if last_result.get("isError"): + return last_result + + # After executing actions, take a screenshot + screenshot_result = await self.mcp.call("computer", {"action": "screenshot"}) + return screenshot_result + async def run(self, prompt: str, max_steps: int) -> Dict[str, Any]: """Run the agent on a task.""" start_time = time.time() - - system_prompt = f"""You control a browser via tools. -STRICT RULES: -- Text output with no tool calls means task complete. Only output text when fully done. -- When finished: output only "DONE: [what you did]" -""" - - # Get tools from MCP server and convert to Gemini format - mcp_tools = self.mcp.get_tools() - gemini_tools = mcp_tools_to_gemini(mcp_tools) - - # Log system prompt and tools + system_prompt = """You are a helpful agent. Complete the task by interacting with the browser. + +Use the available tools to click, type, scroll, and interact with the page. +Coordinates are normalized 0-1000 (0,0 is top-left, 1000,1000 is bottom-right). + +When done, stop calling tools and provide your final response.""" + + # Get Gemini 3 tools + gemini_tools = get_gemini_3_tools() + log_verbose("\n" + "="*60) log_verbose("SYSTEM PROMPT:") log_verbose("="*60) log_verbose(system_prompt) - - log_verbose("\n" + "="*60) - log_verbose(f"TOOLS ({len(mcp_tools)} total):") - log_verbose("="*60) - for tool in mcp_tools: - log_verbose(f"\n {tool['name']}:") - log_verbose(f" Description: {tool.get('description', '')[:200]}") - schema = tool.get('inputSchema', {}) - props = schema.get('properties', {}) - if props: - log_verbose(f" Parameters:") - for pname, pinfo in props.items(): - ptype = pinfo.get('type', 'any') - pdesc = pinfo.get('description', '')[:80] - log_verbose(f" - {pname} ({ptype}): {pdesc}") - + + log_verbose(f"\nTOOLS ({len(gemini_tools)} total):") + for tool in GEMINI_3_TOOL_DEFINITIONS: + log_verbose(f" {tool['name']}: {tool['description'][:80]}...") + + # Configure Gemini with thinking enabled config = types.GenerateContentConfig( - max_output_tokens=4096, + max_output_tokens=65536, system_instruction=system_prompt, tools=[types.Tool(function_declarations=gemini_tools)], thinking_config=types.ThinkingConfig(include_thoughts=True), ) - + # Set config on session for logging (if session exists) if self.session: self.session.config = config - history: List[types.Content] = [] - - user_prompt = f"""###User instruction: {prompt}""" - history.append(types.Content(role="user", parts=[types.Part(text=user_prompt)])) + # Take initial screenshot + print("Taking initial screenshot...") + initial_screenshot = await self._take_screenshot() + + # Build initial user message with task + screenshot + user_parts = [Part(text=f"Task: {prompt}")] + if initial_screenshot: + user_parts.append(Part(inline_data={ + "mime_type": "image/png", + "data": initial_screenshot, + })) + print("✓ Initial screenshot captured") + else: + print("⚠ Could not capture initial screenshot") + + history: List[Content] = [Content(role="user", parts=user_parts)] self.transcript.append({"role": "user", "content": prompt}) - + log_verbose("\n" + "="*60) log_verbose("USER PROMPT:") log_verbose("="*60) - log_verbose(user_prompt) - + log_verbose(prompt) + for step in range(1, max_steps + 1): print(f"\n{'='*50}") print(f"Step {step}/{max_steps}") @@ -300,40 +503,45 @@ async def run(self, prompt: str, max_steps: int) -> Dict[str, Any]: contents=history, config=config, ) - self._consecutive_errors = 0 # Reset on success + self._consecutive_errors = 0 except Exception as e: self._consecutive_errors += 1 error_type = type(e).__name__ print(f"API error ({error_type}): {e}") print(f" Consecutive errors: {self._consecutive_errors}/{self._max_consecutive_errors}") - + if self._consecutive_errors >= self._max_consecutive_errors: return self._result(False, f"Too many consecutive API errors: {error_type}: {e}", step, start_time) # Check for retryable errors if "429" in str(e) or "quota" in str(e).lower() or "rate" in str(e).lower(): - print(f" Rate limited, waiting 10s...") + print(" Rate limited, waiting 10s...") await asyncio.sleep(10) continue elif "503" in str(e) or "500" in str(e) or "overloaded" in str(e).lower(): - print(f" Server error, waiting 5s...") + print(" Server error, waiting 5s...") await asyncio.sleep(5) continue else: return self._result(False, f"{error_type}: {e}", step, start_time) - + if not response.candidates: print("[WARN] No candidates, retrying...") log_verbose(f" Response: {response}") continue - + candidate = response.candidates[0] if not candidate.content or not candidate.content.parts: print("[WARN] Empty response, retrying...") - log_verbose(f" Candidate: {candidate}") continue - - # Log to Fleet session (live) + + # Extract reasoning trace + reasoning = extract_reasoning_from_candidate(candidate) + if reasoning: + preview = reasoning[:100] + "..." if len(reasoning) > 100 else reasoning + print(f"🧠 Thinking: {preview}") + + # Log to Fleet session if available if self.session: try: await self.session.log(history, response) @@ -359,19 +567,19 @@ async def run(self, prompt: str, max_steps: int) -> Dict[str, Any]: # Extract function calls and text function_calls = [p.function_call for p in candidate.content.parts if p.function_call] - text_parts = [p.text for p in candidate.content.parts if p.text] - + text_parts = [p.text for p in candidate.content.parts if p.text and not getattr(p, "thought", False)] + # Print model output if text_parts: for text in text_parts: display = text[:200] + "..." if len(text) > 200 else text print(f"Model: {display}") - - # Check for completion + + # Check for completion (no function calls) if text_parts and not function_calls: final_text = " ".join(text_parts) self.transcript.append({"role": "assistant", "content": final_text}) - + if final_text.strip().upper().startswith("DONE:"): answer = final_text.strip()[5:].strip() print(f"\n✓ Agent completed: {answer[:100]}") @@ -381,28 +589,33 @@ async def run(self, prompt: str, max_steps: int) -> Dict[str, Any]: print(f"\n✗ Agent failed: {error[:100]}") return self._result(False, error, step, start_time) else: - # Text without DONE/FAILED - treat as completion print(f"\n✓ Agent finished with response") return self._result(True, None, step, start_time, final_text) - + + # Check for thinking-only response (no function calls, no text) + if not function_calls and not text_parts: + print("🧠 Thinking-only response, continuing...") + # Add thinking to history so model has context + history.append(candidate.content) + continue + if function_calls: # Add model's response to history history.append(candidate.content) - + log_verbose(f"\n Executing {len(function_calls)} function call(s):") - - # Execute each function call in series with delays + + # Execute each function call response_parts = [] for i, fc in enumerate(function_calls): name = fc.name args = dict(fc.args) if fc.args else {} print(f" Tool {i+1}/{len(function_calls)}: {name}({json.dumps(args)})") self.transcript.append({"role": "tool_call", "name": name, "args": args}) - + try: - result = await self._execute_tool(name, args) - log_verbose(f" Result: isError={result.get('isError', False)}, content_types={[c.get('type') for c in result.get('content', [])]}") - + result = await self._execute_gemini_function(name, args) + if result.get("isError"): self._consecutive_errors += 1 error_text = "" @@ -410,66 +623,72 @@ async def run(self, prompt: str, max_steps: int) -> Dict[str, Any]: if c.get("type") == "text": error_text = c.get("text", "")[:200] print(f" Tool error: {error_text}") + + # Return error to model + response_parts.append(Part( + function_response={ + "name": name, + "response": {"status": "error", "error": error_text}, + } + )) else: self._consecutive_errors = 0 + img_data = get_image_data(result) + + if img_data: + # Function response with screenshot + response_parts.append(Part( + function_response={ + "name": name, + "response": {"status": "success"}, + } + )) + # Add screenshot as inline_data + response_parts.append(Part( + inline_data={ + "mime_type": "image/png", + "data": img_data, + } + )) + log_verbose(" Response: screenshot captured") + else: + response_parts.append(Part( + function_response={ + "name": name, + "response": {"status": "success"}, + } + )) + log_verbose(" Response: no screenshot") + except Exception as e: self._consecutive_errors += 1 error_type = type(e).__name__ print(f" Tool exception ({error_type}): {e}") - print(f" Consecutive errors: {self._consecutive_errors}/{self._max_consecutive_errors}") - log_verbose(f" Exception: {error_type}: {e}") - - # Check if this is a connection/MCP error that we should fail fast on + if "connection" in str(e).lower() or "closed" in str(e).lower(): - print(f" MCP connection lost, failing task") + print(" MCP connection lost, failing task") return self._result(False, f"MCP connection error: {e}", step, start_time) - - result = {"content": [{"type": "text", "text": str(e)}], "isError": True} - - # Build function response with image embedded (per reference format) - img_data = get_image_data(result) # Base64 string - - if img_data: - log_verbose(f" Response: image (base64 len={len(img_data)})") - # Function response with image in parts - fr_part = types.Part( - function_response=types.FunctionResponse( - name=name, - response={"status": "success" if not result.get("isError") else "error"}, - parts=[ - types.FunctionResponsePart( - inline_data=types.FunctionResponseBlob( - mime_type="image/png", - data=img_data, # Base64 string - ) - ) - ], - ) - ) - else: - log_verbose(f" Response: no image (status only)") - # Function response without image - fr_part = types.Part( - function_response=types.FunctionResponse( - name=name, - response={"status": "error" if result.get("isError") else "success"}, - ) - ) - response_parts.append(fr_part) - - # Small delay between tool calls to let page settle + + response_parts.append(Part( + function_response={ + "name": name, + "response": {"status": "error", "error": str(e)}, + } + )) + + # Small delay between tool calls if i < len(function_calls) - 1: await asyncio.sleep(0.1) - - # Add function responses with role="model" (per reference) - history.append(types.Content(role="model", parts=response_parts)) - log_verbose(f" Added {len(response_parts)} function response(s) to history") - - # Max steps reached - still mark as completed so verification runs - # The agent may have done the task but just didn't say "DONE" - print(f"\n⚠ Max steps ({max_steps}) reached - will still run verification") + + # Add function responses to history as user role + # (Gemini expects function_response in user messages) + history.append(Content(role="user", parts=response_parts)) + log_verbose(f" Added {len(response_parts)} response part(s) to history") + + # Max steps reached + print(f"\n⚠ Max steps ({max_steps}) reached") return self._result(True, "Max steps reached", max_steps, start_time, "Max steps reached - task may be complete") - + def _result(self, completed: bool, error: Optional[str], steps: int, start_time: float, answer: str = None) -> Dict: """Build result dict.""" return { @@ -490,21 +709,21 @@ async def main(): "task_key": os.environ.get("FLEET_TASK_KEY", ""), "job_id": os.environ.get("FLEET_JOB_ID"), "instance_id": os.environ.get("FLEET_INSTANCE_ID"), - "model": os.environ.get("FLEET_MODEL", "gemini-2.5-pro"), + "model": os.environ.get("FLEET_MODEL", "gemini-3-pro-preview"), "max_steps": int(os.environ.get("FLEET_MAX_STEPS", "200")), } - - print(f"Gemini CUA Agent") + + print("Gemini CUA Agent") print(f" Model: {config['model']}") print(f" MCP: {config['url']}") print(f" Verbose: {VERBOSE}") print(f" Task: {config['prompt'][:80]}...") - + if not os.environ.get("GEMINI_API_KEY"): result = {"task_key": config["task_key"], "completed": False, "error": "No GEMINI_API_KEY"} print(json.dumps(result)) return result - + try: # Create Fleet session for live logging session = None @@ -515,15 +734,14 @@ async def main(): task_key=config["task_key"], instance_id=config["instance_id"], ) - + async with MCP(config["url"]) as mcp: agent = GeminiAgent(mcp, config["model"], session=session) result = await agent.run(config["prompt"], config["max_steps"]) result["task_key"] = config["task_key"] - # Include session_id in result so orchestrator can complete it after verification if session and session.session_id: result["session_id"] = session.session_id - + print(json.dumps(result)) return result except Exception as e: diff --git a/fleet/cli.py b/fleet/cli.py index ae135be9..7504dde7 100644 --- a/fleet/cli.py +++ b/fleet/cli.py @@ -408,7 +408,7 @@ def get_session_transcript( # Task info if result.task: console.print(f"[bold]Task:[/bold] {result.task.key}") - console.print(f" Environment: {result.task.env_id}") + console.print(f" Environment: {result.task.env_key}") if result.task.version: console.print(f" Version: {result.task.version}") console.print() diff --git a/fleet/client.py b/fleet/client.py index d7021a87..a4c37770 100644 --- a/fleet/client.py +++ b/fleet/client.py @@ -48,6 +48,9 @@ JobResponse, JobListResponse, JobCreateResponse, + ScenarioResponse, + ScenarioListResponse, + ScenarioCreateRequest, JobSessionsResponse, SessionTranscriptResponse, SessionIngestRequest, @@ -598,7 +601,7 @@ def make( return instance def make_for_task(self, task: Task) -> SyncEnv: - return self.make(env_key=f"{task.env_id}:{task.version}") + return self.make(env_key=task.env_spec) def instances( self, status: Optional[str] = None, region: Optional[str] = None, run_id: Optional[str] = None, profile_id: Optional[str] = None @@ -939,9 +942,7 @@ def load_task_from_json( task = Task( key=task_json.get("key", task_json.get("id")), prompt=task_json["prompt"], - env_id=task_json.get( - "env_id", task_json.get("env_key") - ), # Use env_id or fallback to env_key + env_key=task_json.get("env_key") or task_json.get("environment_id") or task_json.get("env_id"), # Prefer env_key, fallback to environment_id (API) or env_id (legacy) created_at=task_json.get("created_at"), version=task_json.get("version"), data_id=task_json.get("data_id"), @@ -954,6 +955,10 @@ def load_task_from_json( verifier_runtime_version=verifier_runtime_version, # Set verifier_runtime_version metadata=task_json.get("metadata", {}), # Default empty metadata output_json_schema=task_json.get("output_json_schema"), # JSON schema for output + factual_answer=task_json.get("factual_answer"), # Expected answer for research/factual tasks + task_modality=task_json.get("task_modality"), # Task modality (computer_use, tool_use, browser) + task_scenario_id=task_json.get("task_scenario_id"), # Task scenario ID + task_lifecycle_status=task_json.get("task_lifecycle_status"), # Task lifecycle status ) return task @@ -1122,7 +1127,7 @@ def create_verifier_with_fallback(tr, emb_code, is_error): task = Task( key=task_response.key, prompt=task_response.prompt, - env_id=task_response.environment_id, # Map environment_id -> env_id + env_key=task_response.env_key, created_at=task_response.created_at, version=task_response.version, data_id=getattr(task_response, "data_id", None), # Get data_id if available @@ -1135,6 +1140,10 @@ def create_verifier_with_fallback(tr, emb_code, is_error): verifier_runtime_version=verifier_runtime_version, # Set verifier_runtime_version metadata=metadata, output_json_schema=getattr(task_response, "output_json_schema", None), # Get output_json_schema if available + factual_answer=getattr(task_response, "factual_answer", None), # Get factual_answer if available + task_modality=getattr(task_response, "task_modality", None), # Get task_modality if available + task_scenario_id=getattr(task_response, "task_scenario_id", None), # Get task_scenario_id if available + task_lifecycle_status=getattr(task_response, "task_lifecycle_status", None), # Get task_lifecycle_status if available ) tasks.append(task) @@ -1200,26 +1209,32 @@ def import_single_task(self, task: Task, project_key: Optional[str] = None): project_key: Optional project key to associate with the task Returns: - Response from the API, or None if the import failed - """ - try: - # Validate that verifier_func exists - if not task.verifier_func: - raise ValueError( - f"Task {task.key} is missing verifier_func. " - "All tasks must have a verifier_func to be imported." - ) + Response from the API - params = {} - if project_key: - params["project_key"] = project_key - response = self.client.request( - "POST", "/v1/tasks", json=task.model_dump(), params=params + Raises: + ValueError: If task is missing verifier_func + Exception: If the API request fails + """ + # Validate that verifier_func exists + if not task.verifier_func: + raise ValueError( + f"Task {task.key} is missing verifier_func. " + "All tasks must have a verifier_func to be imported." ) - return response - except Exception as e: - # logger.error(f"Failed to import task {task.key}: {e}") - return None + + params = {} + if project_key: + params["project_key"] = project_key + + # Convert task to dict and map env_key -> env_id for API compatibility + task_data = task.model_dump() + if "env_key" in task_data: + task_data["env_id"] = task_data.pop("env_key") + + response = self.client.request( + "POST", "/v1/tasks", json=task_data, params=params + ) + return response def import_tasks(self, filename: str, project_key: Optional[str] = None): """Import tasks from a JSON file. @@ -1266,6 +1281,76 @@ def import_tasks(self, filename: str, project_key: Optional[str] = None): # Filter out None values (failed imports) return [r for r in responses if r is not None] + # Scenario methods + def get_scenario(self, scenario_id: str) -> ScenarioResponse: + """Get a scenario by ID. + + Args: + scenario_id: The scenario ID to retrieve + + Returns: + ScenarioResponse containing the scenario details + """ + response = self.client.request("GET", f"/v1/scenarios/{scenario_id}") + return ScenarioResponse(**response.json()) + + def list_scenarios(self, project_id: Optional[str] = None) -> ScenarioListResponse: + """List scenarios, optionally filtered by project. + + Args: + project_id: Optional project ID to filter scenarios by + + Returns: + ScenarioListResponse containing the list of scenarios + """ + params = {} + if project_id: + params["project_id"] = project_id + response = self.client.request("GET", "/v1/scenarios", params=params) + return ScenarioListResponse(**response.json()) + + def create_scenario( + self, + scenario_title: str, + project_id: str, + output_json_schema: Optional[Dict[str, Any]] = None, + task_complexity_tier: Optional[str] = None, + user_story: Optional[str] = None, + entrypoint: Optional[str] = None, + is_research_based: Optional[bool] = None, + is_action_based: Optional[bool] = None, + scenario_verifier_prompt: Optional[str] = None, + ) -> ScenarioResponse: + """Create a new scenario. + + Args: + scenario_title: Title for the scenario + project_id: Project ID to associate the scenario with + output_json_schema: Optional JSON schema for expected output + task_complexity_tier: Optional complexity tier + user_story: Optional user story description + entrypoint: Optional entrypoint URL + is_research_based: Whether the scenario is research-based + is_action_based: Whether the scenario is action-based + scenario_verifier_prompt: Optional verifier prompt + + Returns: + ScenarioResponse containing the created scenario + """ + request = ScenarioCreateRequest( + scenario_title=scenario_title, + project_id=project_id, + output_json_schema=output_json_schema, + task_complexity_tier=task_complexity_tier, + user_story=user_story, + entrypoint=entrypoint, + is_research_based=is_research_based, + is_action_based=is_action_based, + scenario_verifier_prompt=scenario_verifier_prompt, + ) + response = self.client.request("POST", "/v1/scenarios", json=request.model_dump(exclude_none=True)) + return ScenarioResponse(**response.json()) + def account(self) -> AccountResponse: """Get account information including instance limits and usage. @@ -1281,6 +1366,8 @@ def update_task( prompt: Optional[str] = None, verifier_code: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, + writer_metadata: Optional[Dict[str, Any]] = None, + qa_metadata: Optional[Dict[str, Any]] = None, ) -> TaskResponse: """Update an existing task. @@ -1289,11 +1376,19 @@ def update_task( prompt: New prompt text for the task (optional) verifier_code: Python code for task verification (optional) metadata: Additional metadata for the task (optional) + writer_metadata: Metadata filled by task writer (optional) + qa_metadata: Metadata filled by QA reviewer (optional) Returns: TaskResponse containing the updated task details """ - payload = TaskUpdateRequest(prompt=prompt, verifier_code=verifier_code, metadata=metadata) + payload = TaskUpdateRequest( + prompt=prompt, + verifier_code=verifier_code, + metadata=metadata, + writer_metadata=writer_metadata, + qa_metadata=qa_metadata, + ) response = self.client.request( "PUT", f"/v1/tasks/{task_key}", json=payload.model_dump(exclude_none=True) ) diff --git a/fleet/models.py b/fleet/models.py index 30439f33..2becd004 100644 --- a/fleet/models.py +++ b/fleet/models.py @@ -7,7 +7,7 @@ from enum import Enum from typing import Any, Dict, List, Optional, Union -from pydantic import BaseModel, Field, conint +from pydantic import BaseModel, ConfigDict, Field, conint class CDPDescribeResponse(BaseModel): @@ -154,22 +154,36 @@ class TableSchema(BaseModel): class TaskRequest(BaseModel): + model_config = ConfigDict(populate_by_name=True) + key: str = Field(..., title="Key") prompt: str = Field(..., title="Prompt") - environment_id: str = Field(..., title="Environment Id") + env_key: str = Field(..., title="Env Key", alias="environment_id") verifier_id: Optional[str] = Field(None, title="Verifier Id") version: Optional[str] = Field(None, title="Version") env_variables: Optional[Dict[str, Any]] = Field(None, title="Env Variables") metadata: Optional[Dict[str, Any]] = Field(None, title="Metadata") + writer_metadata: Optional[Dict[str, Any]] = Field( + None, title="Writer Metadata", description="Metadata filled by task writer" + ) output_json_schema: Optional[Dict[str, Any]] = Field( None, title="Output Json Schema" ) + factual_answer: Optional[Any] = Field( + None, title="Factual Answer", description="Expected answer for research/factual tasks" + ) class TaskUpdateRequest(BaseModel): prompt: Optional[str] = Field(None, title="Prompt") verifier_code: Optional[str] = Field(None, title="Verifier Code") metadata: Optional[Dict[str, Any]] = Field(None, title="Metadata") + writer_metadata: Optional[Dict[str, Any]] = Field( + None, title="Writer Metadata", description="Metadata filled by task writer" + ) + qa_metadata: Optional[Dict[str, Any]] = Field( + None, title="QA Metadata", description="Metadata filled by QA reviewer" + ) class VerifierData(BaseModel): @@ -183,10 +197,12 @@ class VerifierData(BaseModel): class TaskResponse(BaseModel): + model_config = ConfigDict(populate_by_name=True) + key: str = Field(..., title="Key") prompt: str = Field(..., title="Prompt") team_id: str = Field(..., title="Team Id") - environment_id: str = Field(..., title="Environment Id") + env_key: str = Field(..., title="Env Key", alias="environment_id") created_at: str = Field(..., title="Created At") verifier_id: Optional[str] = Field(None, title="Verifier Id") verifier_func: Optional[str] = Field(None, title="Verifier Func") @@ -199,6 +215,53 @@ class TaskResponse(BaseModel): output_json_schema: Optional[Dict[str, Any]] = Field( None, title="Output Json Schema" ) + factual_answer: Optional[Any] = Field( + None, title="Factual Answer", description="Expected answer for research/factual tasks" + ) + task_modality: Optional[str] = Field( + None, title="Task Modality", description="Task modality (computer_use, tool_use, browser)" + ) + task_scenario_id: Optional[int] = Field( + None, title="Task Scenario ID", description="ID of the task scenario this task belongs to" + ) + task_lifecycle_status: Optional[str] = Field( + None, title="Task Lifecycle Status", description="Task lifecycle status (production, development, staging, etc.)" + ) + + +class ScenarioResponse(BaseModel): + """Response model for a task scenario.""" + + id: int = Field(..., title="ID", description="Scenario ID") + scenario_title: Optional[str] = Field(None, title="Scenario Title") + output_json_schema: Optional[Dict[str, Any]] = Field(None, title="Output JSON Schema") + task_complexity_tier: Optional[str] = Field(None, title="Task Complexity Tier") + user_story: Optional[str] = Field(None, title="User Story") + entrypoint: Optional[str] = Field(None, title="Entrypoint") + is_research_based: Optional[bool] = Field(None, title="Is Research Based") + is_action_based: Optional[bool] = Field(None, title="Is Action Based") + scenario_verifier_prompt: Optional[str] = Field(None, title="Scenario Verifier Prompt") + + +class ScenarioListResponse(BaseModel): + """Response model for listing scenarios.""" + + scenarios: List["ScenarioResponse"] = Field(..., title="Scenarios") + total: int = Field(..., title="Total") + + +class ScenarioCreateRequest(BaseModel): + """Request model for creating a scenario.""" + + scenario_title: str = Field(..., title="Scenario Title") + project_id: str = Field(..., title="Project ID", description="Project to associate the scenario with") + output_json_schema: Optional[Dict[str, Any]] = Field(None, title="Output JSON Schema") + task_complexity_tier: Optional[str] = Field(None, title="Task Complexity Tier") + user_story: Optional[str] = Field(None, title="User Story") + entrypoint: Optional[str] = Field(None, title="Entrypoint") + is_research_based: Optional[bool] = Field(None, title="Is Research Based") + is_action_based: Optional[bool] = Field(None, title="Is Action Based") + scenario_verifier_prompt: Optional[str] = Field(None, title="Scenario Verifier Prompt") class ValidationError(BaseModel): @@ -491,9 +554,11 @@ class SessionInfo(BaseModel): class TaskInfo(BaseModel): """Task information for session transcript.""" + model_config = ConfigDict(populate_by_name=True) + key: str = Field(..., title="Key") prompt: str = Field(..., title="Prompt") - env_id: str = Field(..., title="Env Id") + env_key: str = Field(..., title="Env Key", alias="env_id") env_variables: Optional[Dict[str, Any]] = Field(None, title="Env Variables") created_at: Optional[str] = Field(None, title="Created At") version: Optional[str] = Field(None, title="Version") diff --git a/fleet/tasks.py b/fleet/tasks.py index 46247f87..0b52b61b 100644 --- a/fleet/tasks.py +++ b/fleet/tasks.py @@ -20,7 +20,7 @@ class Task(BaseModel): key: str = Field(..., description="Unique task key identifier") prompt: str = Field(..., description="Task prompt or instruction") - env_id: str = Field(..., description="Environment identifier") + env_key: str = Field(..., description="Environment key") env_variables: Optional[Dict[str, Any]] = Field( default_factory=dict, description="Environment variables" ) @@ -43,6 +43,18 @@ class Task(BaseModel): output_json_schema: Optional[Dict[str, Any]] = Field( None, description="JSON schema for expected output format" ) + factual_answer: Optional[Any] = Field( + None, description="Expected answer for research/factual tasks" + ) + task_modality: Optional[str] = Field( + None, description="Task modality (computer_use, tool_use, browser)" + ) + task_scenario_id: Optional[int] = Field( + None, description="ID of the task scenario this task belongs to" + ) + task_lifecycle_status: Optional[str] = Field( + None, description="Task lifecycle status (production, development, staging, etc.)" + ) @validator("key") def validate_key_format(cls, v): @@ -59,21 +71,41 @@ def serialize_created_at(self, dt: Optional[datetime], _info): return dt.isoformat() if dt else None @property - def env_key(self) -> str: - """Get the environment key combining env_id and version.""" - if self.version and self.version != "None" and ":" not in self.env_id: - return f"{self.env_id}:{self.version}" - return self.env_id + def env_spec(self) -> str: + """Get the env spec string (env_key:version) for instance creation.""" + if self.version and self.version != "None" and ":" not in self.env_key: + return f"{self.env_key}:{self.version}" + return self.env_key @property - def data_key(self) -> Optional[str]: - """Get the data key combining data_id and data_version.""" + def data_spec(self) -> Optional[str]: + """Get the data spec string (data_id:data_version) for instance creation.""" if self.data_id and self.data_version: return f"{self.data_id}:{self.data_version}" elif self.data_id: return self.data_id return None + @property + def data_key(self) -> Optional[str]: + """Alias for data_spec for backward compatibility.""" + return self.data_spec + + @property + def has_verifier(self) -> bool: + """Whether this task has a verifier function.""" + return self.verifier is not None or self.verifier_func is not None + + @property + def is_research_based(self) -> bool: + """Whether this task is research/factual (has a factual_answer).""" + return self.factual_answer is not None + + @property + def is_action_based(self) -> bool: + """Whether this task is action-based (no factual_answer).""" + return self.factual_answer is None + class Config: """Pydantic model configuration.""" @@ -214,7 +246,7 @@ def make_env( ): """Create an environment instance for this task's environment. - Alias for make() method. Uses the task's env_id (and version if present) to create the env. + Alias for make() method. Uses the task's env_key (and version if present) to create the env. """ return self.make( region=region, @@ -235,7 +267,7 @@ def make( """Create an environment instance with task's configuration. Auto-populates environment creation with: - - env_key (env_id + version) + - env_key (env_key + version) - data_key (data_id + data_version, if present) - env_variables (if present) - run_id (if present) @@ -252,18 +284,18 @@ def make( Environment instance configured for this task Example: - task = fleet.Task(key="my-task", prompt="...", env_id="my-env", + task = fleet.Task(key="my-task", prompt="...", env_key="my-env", data_id="my-data", data_version="v1.0") env = task.make(region="us-west-2", run_id="my-batch-123", heartbeat_interval=60) """ - if not self.env_id: - raise ValueError("Task has no env_id defined") + if not self.env_key: + raise ValueError("Task has no env_key defined") # Deferred import to avoid circular dependencies from fleet.env import make return make( - env_key=self.env_key, + env_key=self.env_spec, data_key=self.data_key, region=region, env_variables=self.env_variables if self.env_variables else None, @@ -364,6 +396,10 @@ def contains_number(text: str, target_number) -> bool: # Execute the cleaned verifier code in the namespace exec(cleaned_code, exec_globals, local_namespace) + # Merge local_namespace into exec_globals so helper functions are accessible + # from the main verifier function when it's called + exec_globals.update(local_namespace) + # Find the function that was defined (not imported) # Functions defined via exec have co_filename == '' # Imported functions have their actual module file path @@ -453,7 +489,12 @@ def load_tasks( def update_task( - task_key: str, prompt: Optional[str] = None, verifier_code: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None + task_key: str, + prompt: Optional[str] = None, + verifier_code: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + writer_metadata: Optional[Dict[str, Any]] = None, + qa_metadata: Optional[Dict[str, Any]] = None, ): """Convenience function to update an existing task. @@ -462,6 +503,8 @@ def update_task( prompt: New prompt text for the task (optional) verifier_code: Python code for task verification (optional) metadata: Additional metadata for the task (optional) + writer_metadata: Metadata filled by task writer (optional) + qa_metadata: Metadata filled by QA reviewer (optional) Returns: TaskResponse containing the updated task details @@ -470,12 +513,18 @@ def update_task( response = fleet.update_task("my-task", prompt="New prompt text") response = fleet.update_task("my-task", verifier_code="def verify(env): return True") response = fleet.update_task("my-task", metadata={"seed": 42, "story": "Updated story"}) + response = fleet.update_task("my-task", writer_metadata={"author": "john"}) """ from .global_client import get_client client = get_client() return client.update_task( - task_key=task_key, prompt=prompt, verifier_code=verifier_code, metadata=metadata + task_key=task_key, + prompt=prompt, + verifier_code=verifier_code, + metadata=metadata, + writer_metadata=writer_metadata, + qa_metadata=qa_metadata, ) @@ -514,7 +563,7 @@ def import_task(task: Task, project_key: Optional[str] = None): Response from the API, or None if the import failed Examples: - task = fleet.Task(key="my-task", prompt="Do something", env_id="my-env") + task = fleet.Task(key="my-task", prompt="Do something", env_key="my-env") response = fleet.import_task(task) response = fleet.import_task(task, project_key="my-project") """ diff --git a/pyproject.toml b/pyproject.toml index 03d141e9..3e226cbc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "fleet-python" -version = "0.2.103" +version = "0.2.115" description = "Python SDK for Fleet environments" authors = [ {name = "Fleet AI", email = "nic@fleet.so"}, @@ -132,3 +132,8 @@ testpaths = ["tests"] "httpx.AsyncHTTPTransport" = "httpx.HTTPTransport" "httpx.SyncHTTPTransport" = "httpx.HTTPTransport" "aiohttp" = "httpx" + +[dependency-groups] +dev = [ + "ruff>=0.14.9", +] diff --git a/uv.lock b/uv.lock index 7c408c25..baf2f8e8 100644 --- a/uv.lock +++ b/uv.lock @@ -661,7 +661,7 @@ wheels = [ [[package]] name = "fleet-python" -version = "0.2.100" +version = "0.2.115" source = { editable = "." } dependencies = [ { name = "aiohttp" }, @@ -705,6 +705,11 @@ playwright = [ { name = "playwright" }, ] +[package.dev-dependencies] +dev = [ + { name = "ruff" }, +] + [package.metadata] requires-dist = [ { name = "aiohttp", specifier = ">=3.8.0" }, @@ -733,6 +738,9 @@ requires-dist = [ ] provides-extras = ["cli", "dev", "playwright", "eval"] +[package.metadata.requires-dev] +dev = [{ name = "ruff", specifier = ">=0.14.9" }] + [[package]] name = "frozenlist" version = "1.8.0"