Inspect resources with Python

These clients cover everything beside client.runs. Every method is in the Python SDK reference. The examples run inside async with ValkyrieClient() as client:, as in the quickstart.

Read tasks

from valkyrie.sdk import FetchTasksRequest, TaskStatus

failed = await client.benchmarks.tasks(
    run_id, FetchTasksRequest(status=[TaskStatus.ERROR], limit=25)
)

task = await client.benchmarks.task(run_id, failed.tasks[0].task_id)
print(task.status, task.error_message, task.evaluation_result)

statuses = await client.benchmarks.statuses([run_id, other_run_id])

tasks returns one page. To walk every page with the same filters:

errors = FetchTasksRequest(status=[TaskStatus.ERROR])
async for task in client.benchmarks.iter_tasks(run_id, errors):
    print(task.task_id)

Pages use offsets, so a task whose status changes while you read can move between pages.

Read logs

from datetime import datetime, timezone

run_logs = await client.logs.fetch_run(
    run_id,
    query="rate limit",
    start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    end_time=datetime(2026, 1, 2, tzinfo=timezone.utc),
)

task_logs = await client.logs.fetch_task(run_id, task_id, query="ERROR")

async for event in client.logs.stream_task(run_id, task_id):
    print(event.timestamp, event.message)

query matches text in log messages. Times need a time zone, and logs at exactly those times are included. page_run and page_task load one page at a time.

Download artifacts

page = await client.artifacts.list(run_id, prefix="task-1")

link = await client.artifacts.download_url(run_id, "task-1/agent_output.tar.gz")

await client.artifacts.download(run_id, "./task-1", path="task-1")

list returns one page of paths under prefix; pass its next_cursor as cursor for the next. download_url returns a temporary link to one file, and download saves a file or folder into a new directory.

Page through the sandbox queue

client.scheduler.overview returns what valkyrie queue status shows. The waiting and active lists page separately. Stop reading a list when its next offset is None:

waiting_offset = 0
while True:
    page = await client.scheduler.overview(
        waiting_limit=200, waiting_offset=waiting_offset
    )
    for entry in page.waiting_entries:
        print(
            entry.benchmark_uuid, entry.external_task_id, entry.priority, entry.position
        )
    if page.waiting_next_offset is None:
        break
    waiting_offset = page.waiting_next_offset

Use active_entries, active_offset, and active_next_offset the same way for active tasks. The queue changes between calls, so a task can appear on two pages or on none.

Manage projects

Every run spends for a project. Create one once, then start runs in it by its handle:

from valkyrie.sdk import SpendType

project = await client.projects.create(
    "google-report-xyz",
    name="Google Report XYZ",
    spend_type=SpendType.CUSTOMER,
    subcategory="report",
    customer="google",
    extra_tags={"engagement": "q4"},
)

projects = (await client.projects.list()).projects

The fields match valkyrie project create. To move a run to another project, see client.runs.retag.

Manage agents

agents = await client.agents.list()

await client.agents.push("./my-agent")

await client.agents.download("my-agent", "./agents")

The methods match valkyrie agent, including install from GitHub and remove.

Store secrets and choose providers

Store a secret, then use the reference it returns wherever a run or mapping names a secret:

entry = await client.secrets.set("daytona", daytona_json)

await client.providers.set("daytona", entry.reference)

names = [secret.name for secret in (await client.secrets.list()).secrets]

Values are never returned, and a secret set or removed in the last few minutes might not appear in list yet. Your default provider and saved benchmark services are your own settings in the current organization:

await client.settings.set_default_provider("modal")

await client.settings.set_benchmark_service(
    "swebench", "https://benchmark.example.com", auth_secret="swebench-token"
)

settings = await client.settings.get()

Changing settings needs a login. An access key reads its creator’s settings but can’t change them. See Configure sandbox providers and Benchmark authentication.

Query benchmark services

services = await client.services.list()

catalog = await client.services.catalog()

task_ids = await client.services.task_ids("swebench", dataset="default")

list checks that every service responds, and catalog returns the hosted services without checking them. task_ids uses the URL you saved for the benchmark, if any. To ask the hosted service, pass ignore_custom_services=True. To send extra headers, pass service_headers.

Get started

Agents

Runs

Benchmarks

Integrations

Python SDK

CLI reference

Python SDK reference

Self-hosting

Inspect resources with Python

Read tasks

Read logs

Download artifacts

Page through the sandbox queue

Manage projects

Manage agents

Store secrets and choose providers

Query benchmark services

​Read tasks

​Read logs

​Download artifacts

​Page through the sandbox queue

​Manage projects

​Manage agents

​Store secrets and choose providers

​Query benchmark services

Read tasks

Read logs

Download artifacts

Page through the sandbox queue

Manage projects

Manage agents

Store secrets and choose providers

Query benchmark services