Benchmarks API Reference¶

from mcpbr.benchmarks import (
    Benchmark,
    BenchmarkTask,
    BENCHMARK_REGISTRY,
    create_benchmark,
    list_benchmarks,
)

def load_tasks(
    self,
    sample_size: int | None = None,
    task_ids: list[str] | None = None,
    level: int | None = None,
    filter_difficulty: list[str] | None = None,
    filter_category: list[str] | None = None,
    filter_tags: list[str] | None = None,
) -> list[dict[str, Any]]

def normalize_task(self, task: dict[str, Any]) -> BenchmarkTask

async def create_environment(
    self,
    task: dict[str, Any],
    docker_manager: DockerEnvironmentManager,
) -> TaskEnvironment

async def evaluate(
    self,
    env: TaskEnvironment,
    task: dict[str, Any],
    solution: str,
) -> dict[str, Any]

def get_prebuilt_image(self, task: dict[str, Any]) -> str | None

def get_prompt_template(self) -> str

from mcpbr.benchmarks import create_benchmark

# Create a benchmark by name
benchmark = create_benchmark("humaneval")

# SWE-bench variants auto-set the dataset
benchmark = create_benchmark("swe-bench-verified")
# Internally sets dataset="SWE-bench/SWE-bench_Verified"

# Pass additional kwargs to the constructor
benchmark = create_benchmark("cybergym", level=2)

from mcpbr.benchmarks import BENCHMARK_REGISTRY

# List all registered benchmarks
for name, cls in BENCHMARK_REGISTRY.items():
    print(f"{name:25s} -> {cls.__name__}")

from typing import Any
from mcpbr.benchmarks.base import Benchmark, BenchmarkTask
from mcpbr.docker_env import DockerEnvironmentManager, TaskEnvironment


class MyBenchmark:
    """Custom benchmark implementation."""

    name: str = "my-benchmark"

    def load_tasks(
        self,
        sample_size: int | None = None,
        task_ids: list[str] | None = None,
        level: int | None = None,
        filter_difficulty: list[str] | None = None,
        filter_category: list[str] | None = None,
        filter_tags: list[str] | None = None,
    ) -> list[dict[str, Any]]:
        """Load tasks from your dataset."""
        tasks = [
            {
                "instance_id": "task-001",
                "problem_statement": "Implement a function that...",
                "repo": "my-org/my-repo",
                "base_commit": "abc123",
                "difficulty": "easy",
            },
            # ... more tasks
        ]

        # Apply filters
        if task_ids:
            tasks = [t for t in tasks if t["instance_id"] in task_ids]
        if filter_difficulty:
            tasks = [t for t in tasks if t.get("difficulty") in filter_difficulty]
        if sample_size:
            tasks = tasks[:sample_size]

        return tasks

    def normalize_task(self, task: dict[str, Any]) -> BenchmarkTask:
        """Convert to normalized format."""
        return BenchmarkTask(
            task_id=task["instance_id"],
            problem_statement=task["problem_statement"],
            repo=task["repo"],
            commit=task["base_commit"],
            metadata={"difficulty": task.get("difficulty")},
        )

    async def create_environment(
        self,
        task: dict[str, Any],
        docker_manager: DockerEnvironmentManager,
    ) -> TaskEnvironment:
        """Create a Docker environment for the task."""
        image = self.get_prebuilt_image(task)
        return await docker_manager.create_environment(
            image=image or "python:3.11-slim",
            instance_id=task["instance_id"],
            repo=task["repo"],
            commit=task["base_commit"],
        )

    async def evaluate(
        self,
        env: TaskEnvironment,
        task: dict[str, Any],
        solution: str,
    ) -> dict[str, Any]:
        """Evaluate the agent's solution."""
        # Apply the solution and run tests
        await env.write_file("/tmp/solution.py", solution)
        exit_code, output = await env.exec_command(
            "cd /workspace && python -m pytest tests/ -v"
        )
        return {
            "resolved": exit_code == 0,
            "test_output": output,
        }

    def get_prebuilt_image(self, task: dict[str, Any]) -> str | None:
        """Return pre-built Docker image if available."""
        return None  # No pre-built images

    def get_prompt_template(self) -> str:
        """Return the prompt template for agents."""
        return (
            "You are a software engineer. Solve the following problem:\n\n"
            "{problem_statement}\n\n"
            "Provide your solution as a Python implementation."
        )

# In mcpbr/benchmarks/__init__.py
from .my_benchmark import MyBenchmark

BENCHMARK_REGISTRY["my-benchmark"] = MyBenchmark

VALID_BENCHMARKS = (
    # ... existing benchmarks ...
    "my-benchmark",
)

from mcpbr.benchmarks import list_benchmarks

names = list_benchmarks()
print(names)
# ['swe-bench-lite', 'swe-bench-verified', 'swe-bench-full', 'cybergym',
#  'humaneval', 'mcptoolbench', 'gsm8k', 'mbpp', 'math', ...]