cage-fleet-sdk 0.1.0

Async Python SDK for Cage Fleet

Cage Fleet SDK

cage-fleet-sdk is an asyncio Python client and rolling-concurrency scheduler for the Cage Fleet control plane. It creates marker-only cyber-range Arenas, waits when capacity is exhausted, and submits on demand or at the authoritative deadline. Scheduler state is kept only in memory; Fleet Web is the source of truth after a user process exits.

Python 3.11 or newer is required.

Install

The package is ready to build but has not been published to PyPI yet. Until the first release, install it from a repository checkout:

uv sync --extra dev

After the first release is uploaded, users will install it with:

python -m pip install cage-fleet-sdk

uv may create a virtual environment without a pip module. That is expected; run project commands with uv run. If using a conventional pip-created environment instead, install the checkout with python -m pip install -e '.[dev]'.

Backend contract

The SDK talks to the Cage Fleet control plane implemented by the cage-fleet project, not directly to a Cage Node. Its source of truth is api/openapi.yaml. The public client covers all seven SDK-facing operations: challenge discovery, capacity, owned Arena listing, Arena creation, individual Arena lookup, terminal submission, and explicit close. Submission and close send empty request bodies. Arena creation reuses one UUID Idempotency-Key across capacity retries.

The default HTTP behavior honors standard proxy environment variables. Set trust_env=False on CageFleetClient for a direct connection, such as a localhost test backend.

For development verification:

uv sync --extra dev
uv run pytest
uv run ruff check .
uv run mypy src
uv run python -m build

uv build creates the source distribution and wheel under dist/. Before publishing a new release, update the version and repeat the API and live-backend checks.

Authentication

Pass a Task Token supplied by Cage Fleet. The SDK does not persist it locally.

export CAGE_FLEET_TASK_TOKEN='replace-with-secret'

Run one Arena

create_arena() waits and retries by default when the control plane returns capacity_exhausted. The same Idempotency-Key is reused for every retry.

import asyncio
import os

from cage_fleet_sdk import CageFleetClient


async def main() -> None:
    async with CageFleetClient(
        base_url=os.environ.get(
            "CAGE_FLEET_BASE_URL", "https://eval.agentcyberrange.io/"
        ),
        task_token=os.environ["CAGE_FLEET_TASK_TOKEN"],
    ) as client:
        challenges = await client.list_challenges()
        if not challenges:
            raise RuntimeError("Fleet has no available Challenges")

        arena = await client.create_arena(challenges[0].challenge_id)
        print(arena.task_prompt)
        print(arena.entry_urls)

        result = await client.submit(arena.arena_id)
        print(result.verdict)


asyncio.run(main())

client.submit(arena_id) is a standalone public API. It does not require a Scheduler.

Run rolling-concurrency Rounds

Each Round has its own max_concurrency. A Work Slot is refilled immediately after an Arena obtains a terminal SubmissionResult; the Scheduler does not wait for a fixed wave to finish. Capacity Wait still occupies the task's Work Slot.

import asyncio
import os

from cage_fleet_sdk import (
    ArenaHandle,
    CageFleetClient,
    ChallengeTask,
    RoundConfig,
    RoundScheduler,
)


async def run_agent(arena: ArenaHandle) -> None:
    print(arena.task_prompt)
    print(arena.entry_urls)

    # Run your Agent here. It may submit early:
    await arena.submit()


async def main() -> None:
    async with CageFleetClient(
        base_url=os.environ.get(
            "CAGE_FLEET_BASE_URL", "https://eval.agentcyberrange.io/"
        ),
        task_token=os.environ["CAGE_FLEET_TASK_TOKEN"],
    ) as client:
        challenges = await client.list_challenges()
        if not challenges:
            raise RuntimeError("Fleet has no available Challenges")

        scheduler = RoundScheduler(client)
        result = await scheduler.run(
            rounds=[
                RoundConfig(
                    challenges=[
                        ChallengeTask(challenge_id=challenge.challenge_id)
                        for challenge in challenges
                    ],
                    max_concurrency=2,
                ),
            ],
            on_arena_ready=run_agent,
        )
        print(result.status)


asyncio.run(main())

If run_agent() returns without calling arena.submit(), the Scheduler submits automatically. If it raises, the Scheduler records agent_callback_failed and still submits to preserve completed markers. At lease_expires_at, it sets arena.cancel_event, cancels the callback task, and submits immediately.

The Scheduler intentionally does not persist or resume an Execution. If the user process exits, inspect the Arena and its result in Fleet Web.

Cancellation

The default preserves scoring:

await scheduler.cancel()  # submit active Arenas

Explicit close mode abandons scoring:

await scheduler.cancel(mode="close")

Cancellation stops new Challenge Tasks from starting. Task Outcomes already submitted retain their SubmissionResult; tasks not started are returned as cancelled.

Development verification

uv run pytest
uv run ruff check .
uv run mypy src
uv run python -m build

scripts/live_backend_smoke.py exercises every low-level client method against a running Fleet. scripts/live_scheduler_smoke.py asserts rolling refill and can assert capacity fallback with --expect-capacity-fallback; pass --expected-running-concurrency N to also prove that it saturates the capacity actually available.

Repository-only live Agent test

The repository's scripts/run_agent_test.py discovers every current Challenge through client.list_challenges(), runs each one once with a concurrency limit of two, and waits two minutes per mock Agent. When a mock Agent returns, the Scheduler submits its Arena and immediately refills the released Work Slot. The script verifies peak concurrency, rolling refill, outcomes, and that no active Arena remains.

uv sync --extra dev
uv run python scripts/run_agent_test.py

The script also adds this checkout's src/ directory to its import path, so it can load the local SDK source directly. Installing editable mode remains recommended because it also installs the SDK's third-party dependencies into the same Python environment.

Pass one or more --challenge-id arguments only when intentionally testing a subset. Use --max-concurrency or --mock-seconds to change the test. Supplying --agent-command 'python /absolute/path/to/my_agent.py' replaces the mock wait with a real Agent subprocess. That process receives the task prompt on standard input and Arena metadata in the AGENTCYBERRANGE_* environment variables. The default base URL is https://eval.agentcyberrange.io/; provide the Task Token through CAGE_FLEET_TASK_TOKEN or --task-token.