Convert a benchmark into a service

Valkyrie has no benchmark logic of its own. It creates a sandbox, runs an agent inside it, and calls a benchmark service to get the tasks, set each one up, and grade the result. To convert a benchmark, you wrap the dataset and grader you already have in that service. It answers four questions:

Which tasks exist?
What sandbox does each task run in?
How is a task set up before the agent starts?
How is the agent’s work graded, and turned into one score?

You need Docker to build the images your tasks run on, and a Valkyrie login to test the service. See the quickstart.

Generate the project

create-benchmark-service supplies the web service and the protocol Valkyrie speaks, so you write one BenchmarkService subclass.

uv tool install --exclude-newer "7 days" --exclude-newer-package certifi=false \
  git+https://github.com/vals-ai/create-benchmark-service.git@main

create-benchmark-service swebench

cd swebench-benchmark-service

make install

The generator writes ./NAME-benchmark-service/, where NAME is the name you passed it:

Path	Contents
`main.py`	`app = BenchmarkServiceApp(ExampleBenchmark)`, which serves your class. Leave it as generated.
`src/NAME_benchmark_service/benchmark_service.py`	`ExampleBenchmark`, the class you replace with your benchmark.
`Dockerfile`	The deployable image, serving `main:app` on port `8001`.
`Makefile`	`make install`, `make dev`, `make test`, `make docker-build`, and `make docker-run`.
`pyproject.toml`	Pins the create-benchmark-service version that generated the project.

Implement the service

Replace ExampleBenchmark in benchmark_service.py one method at a time. Each method’s docstring describes what it must do. The examples on this page use these imports:

import json
from collections.abc import AsyncGenerator
from pathlib import Path
from typing import Any

from benchmark_service import BenchmarkService, ImageSource, Resources, Sandbox
from benchmark_service.schemas import (
    BenchmarkEgressPlan,
    EvaluateResponseRequest,
    FinalScoreResult,
    RetrieveTaskResponse,
    StreamChunk,
    StreamMessageChunk,
    StreamResultChunk,
)
from benchmark_service.utils import stream_command

Load the dataset

load_datasets runs once when the service starts, and its result is kept in self.datasets. Key it by dataset name, then by task ID. Each task’s value is whatever your benchmark needs, such as a problem statement, a repository and commit, or an expected answer.

async def load_datasets(self) -> dict[str, dict[str, Any]]:
    tasks = json.loads(Path("data/tasks.json").read_text())
    return {"default": {task["id"]: task for task in tasks}}

Read a dataset back with self.get_dataset(dataset). Several datasets are how one service offers variants, for example a baseline dataset and a with-skills dataset. Runs choose one with --dataset.

Task values usually hold grading data, such as answers and rubrics. None of it reaches the agent unless you return it from retrieve_task or upload it in setup_task.

Describe the sandbox

retrieve_task tells Valkyrie how to build one task’s sandbox: the image, the working directory, where the problem statement goes, the agent’s time limit, and the hardware.

async def retrieve_task(
    self, task_id: str, skip_validation: bool = False, dataset: str | None = None
) -> RetrieveTaskResponse:
    if not skip_validation:
        await self.validate_task_ids([task_id], dataset=dataset)
    task = self.get_dataset(dataset)[task_id]
    return RetrieveTaskResponse(
        source=ImageSource(image=task["image"]),
        problem_path="/tmp/problem_statement.txt",
        cwd="/workspace",
        agent_timeout=1800.0,
        resources=Resources(vcpu=4, memory=8, disk=20),
    )

Image: return the same ImageSource for every task and do per-task work in setup_task, or build one image per task ahead of time and return each task’s own. SnapshotSource starts from a provider snapshot instead, and ComposeSource runs the task as Docker Compose services. Build images for linux/amd64. Private images need registry credentials in your sandbox provider account.
Time limit: agent_timeout is the agent’s generation limit in seconds. See Task time limits.
Hardware: request GPUs through Resources, for example Resources(vcpu=8, memory=32, disk=50, gpu=1, gpu_type="H100").
Docker inside the task: Daytona and Modal sandboxes can run Docker. Use an image with Docker installed and start dockerd during setup.
Install order: Valkyrie installs the agent before setup_task runs. When setup must prepare the runtime first, set agent_install_order="after_setup". ComposeSource tasks always run setup first.

Set network access for each stage with egress:

egress=BenchmarkEgressPlan(
    setup_task=["https://artifacts.example.com"],
    run=[],
    evaluation=[],
)

Each stage takes "*" for unrestricted access, [] to block all outbound traffic, or a list of the hostnames, URLs, IPv4 addresses, and CIDR ranges to allow. Setup and evaluation default to "*". run defaults to None, which leaves the agent unrestricted unless its contract asks for an allowlist.

Set up the task

setup_task runs in the live sandbox before the agent starts: write the problem statement, fetch the repository, install dependencies, or start services. It yields chunks instead of returning, and they stream to whoever is watching the run.

async def setup_task(
    self, task_id: str, sandbox: Sandbox, dataset: str | None = None
) -> AsyncGenerator[StreamChunk, None]:
    task = self.get_dataset(dataset)[task_id]

    yield StreamMessageChunk(type="message", data=f"Preparing {task_id}")
    await sandbox.upload_file("/tmp/problem_statement.txt", task["problem"].encode())

    clone = f"git clone {task['repo']} /workspace"
    async for line in stream_command(sandbox, clone, cwd="/"):
        yield StreamMessageChunk(type="message", data=line)

    yield StreamResultChunk(type="result", data={"status": "ok"})

Chunk	Meaning
`StreamMessageChunk`	Progress text that the person watching the run sees.
`StreamErrorChunk`	A problem worth reporting that doesn’t stop the task.
`StreamResultChunk`	The last chunk. In `setup_task` it reports whether setup worked; in `evaluate_instance` it carries the task’s result.

The sandbox has upload_file, download_file, exec, which returns an ExecResult with exit_code and output, and command, which streams output line by line. stream_command(sandbox, command, cwd) wraps command and raises on a nonzero exit code unless you pass ignore_error=True.

Grade the result

Implement the hook that matches how your benchmark grades, and keep the other one raising an error, as the generated example does:

evaluate_instance grades the agent’s live sandbox: upload the tests, run them, and parse the output. It yields chunks, and its StreamResultChunk is the task’s result.
evaluate_response grades a text answer without a sandbox, and returns the result.

async def evaluate_instance(
    self, task_id: str, sandbox: Sandbox, dataset: str | None = None
) -> AsyncGenerator[StreamChunk, None]:
    task = self.get_dataset(dataset)[task_id]

    yield StreamMessageChunk(type="message", data="Uploading tests")
    await sandbox.upload_file("/tmp/test.sh", task["test_script"].encode())

    result = await sandbox.exec("bash /tmp/test.sh", cwd="/workspace", timeout=600)
    yield StreamMessageChunk(type="message", data=result.output)
    yield StreamResultChunk(type="result", data={"resolved": result.exit_code == 0})


async def evaluate_response(
    self, request: EvaluateResponseRequest, dataset: str | None = None
) -> Any:
    raise ValueError("This benchmark grades in the sandbox")

A task’s result can be any JSON value, because your own service reads it again to compute the score.

Anything you upload in setup_task, the agent can read or change. Upload tests only in evaluate_instance.

Compute the score

calculate_final_score receives {task_id: result} for every task and returns the run’s score, plus any metadata worth reporting. A task that didn’t finish arrives as None.

async def calculate_final_score(
    self, evaluation_results: dict[str, Any], dataset: str | None = None
) -> FinalScoreResult:
    total = len(evaluation_results)
    resolved = sum(
        1 for result in evaluation_results.values() if result and result.get("resolved")
    )
    return FinalScoreResult(
        score=(resolved / total * 100) if total else 0.0,
        metadata={"total_tasks": total, "resolved_tasks": resolved},
    )

Check for reward hacking

Before the first real run, confirm that:

Tests are uploaded in evaluate_instance, never in setup_task.
Reference solutions, patches, and answer keys never enter the sandbox.
The problem statement holds no grading criteria or expected output.
Hints, skills, or scaffolding appear only in the datasets meant to have them.
A live sandbox’s working directory, listed mid-run, contains nothing grading-related.

Test it with Valkyrie

You don’t need to deploy to test. Run the service on your machine, expose it with a reverse tunnel, and point Valkyrie at the tunnel.

Start the service. make dev prints the address it serves on:
make dev

Start a tunnel to that port, and save the tunnel’s address for the benchmark:

valkyrie benchmark service set swebench https://my-tunnel.ngrok-free.dev

Run one task, and watch it stream your setup and grading messages:

valkyrie run start --benchmark swebench --agent my-agent --slice 0:1 --connect --project my-project

Scale up after setup, grading, and scoring all behave. To stop using your copy, run valkyrie benchmark service remove swebench.

Publish the service

A port is finished when it’s listed in the public benchmark services registry: the service lives in its own repository, named like swebench-benchmark-service, added to the registry as a Git submodule with a matching services.yaml entry.

Get started

Agents

Runs

Benchmarks

Integrations

Python SDK

CLI reference

Python SDK reference

Self-hosting

Convert a benchmark into a service

Generate the project

Implement the service

Load the dataset

Describe the sandbox

Set up the task

Grade the result

Compute the score

Check for reward hacking

Test it with Valkyrie

Publish the service

​Generate the project

​Implement the service

​Load the dataset

​Describe the sandbox

​Set up the task

​Grade the result

​Compute the score

​Check for reward hacking

​Test it with Valkyrie

​Publish the service

Generate the project

Implement the service

Load the dataset

Describe the sandbox

Set up the task

Grade the result

Compute the score

Check for reward hacking

Test it with Valkyrie

Publish the service