Developer Guide

Extend AgentCompass with new components.

This guide explains how to extend AgentCompass without crossing runtime boundaries.

Add a Benchmark

A benchmark owns the dataset contract, task preparation, evaluation logic, and aggregate metrics. It should not own model-provider calls, agent loops, sandbox lifecycle, or provider-specific image selection.

Add a benchmark when you are introducing a new task family or a new scoring protocol. If the change only adapts an existing benchmark to a new sandbox image or workspace layout, add a recipe instead.

Benchmark File

Create a module under src/agentcompass/benchmarks/. Most benchmarks define a RuntimeBenchmarkConfig subclass, an optional BenchmarkPlan, and a BaseBenchmark subclass.

from dataclasses import dataclass

from agentcompass.benchmarks.config import RuntimeBenchmarkConfig
from agentcompass.runtime.base import BaseBenchmark, EnvironmentSession
from agentcompass.runtime.models import (
    BenchmarkPlan,
    EnvironmentSpec,
    ExecutionPlan,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskInput,
    TaskOutput,
    TaskSpec,
    TaskStatus,
)
from agentcompass.runtime.registry import BENCHMARKS


@dataclass(slots=True)
class MyBenchmarkConfig(RuntimeBenchmarkConfig):
    dataset_path: str = "data/my_benchmark/tasks.jsonl"


@dataclass(slots=True)
class MyBenchmarkPlan(BenchmarkPlan):
    workspace_root: str = "workspace/"


@BENCHMARKS.register()
class MyBenchmark(BaseBenchmark):
    id = "my_benchmark"
    description = "My Benchmark: one-sentence task and scoring description (paper or project URL)."
    config_class = MyBenchmarkConfig
    evaluation_environment_mode = "none"

    def load_tasks(self, req: RunRequest) -> list[TaskSpec]:
        config = self.build_config(req)
        return [
            TaskSpec(
                task_id="sample-1",
                question="...",
                category="default",
                ground_truth="...",
                metadata={"source": config.dataset_path},
            )
        ]

    def build_plan(self, task: TaskSpec, req: RunRequest, environment: EnvironmentSpec) -> MyBenchmarkPlan:
        return MyBenchmarkPlan()

    async def prepare_task(
        self,
        task: TaskSpec,
        env: EnvironmentSession,
        req: RunRequest,
        plan: MyBenchmarkPlan,
    ) -> PreparedTask:
        return PreparedTask(
            task_id=task.task_id,
            category=task.category,
            ground_truth=task.ground_truth,
            input=TaskInput(prompt=task.question, workspace=plan.workspace_root),
            output=TaskOutput(),
            metadata=dict(task.metadata),
        )

    async def evaluate(
        self,
        task: TaskSpec,
        prepared: PreparedTask,
        result: RunResult,
        req: RunRequest,
        plan: ExecutionPlan,
        env: EnvironmentSession | None = None,
    ) -> RunResult:
        result.correct = result.final_answer == prepared.ground_truth
        result.score = 1.0 if result.correct else 0.0
        result.status = TaskStatus.COMPLETED if not result.error else TaskStatus.EVAL_ERROR
        return result

Benchmark Responsibilities

Hook	Purpose
id	Stable CLI and config id. It is also used by recipes, analyzers, result paths, and docs.
description	User-facing description exported by agentcompass list dump. Benchmarks and harnesses must define a non-empty description.
config_class	Parses --benchmark-params and config defaults. Extend RuntimeBenchmarkConfig when the benchmark needs dataset paths, versions, splits, or sampling controls.
load_tasks()	Loads raw tasks and returns TaskSpec objects. Keep this deterministic and fail early on missing datasets.
select_tasks()	Optional override for custom sampling. The base class already supports sample_ids.
build_plan()	Produces benchmark-side execution state for one task. Recipes may rewrite this plan later.
prepare_task()	Converts a TaskSpec into the benchmark-to-harness PreparedTask contract. Upload files or write prompts here when needed.
evaluate()	Scores a harness RunResult and returns the normalized final RunResult.
aggregate_metrics()	Optional aggregate metric override. The default path aggregates binary correctness.

Use TaskSpec.metadata for source records, image names, docker metadata, expected files, and benchmark-specific IDs. Use PreparedTask.input and PreparedTask.output for the actual harness contract.

Register And Verify

Export the benchmark in src/agentcompass/benchmarks/__init__.py so registry import side effects can discover it:

from .my_benchmark import MyBenchmark  # noqa: F401

Then verify the component list and run a single task:

uv run agentcompass list benchmark

agentcompass run \
  my_benchmark \
  openai_chat \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["sample-1"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add a Harness

A harness owns the agent or model execution loop. It consumes PreparedTask, RunRequest.model, and an EnvironmentSession, then returns a normalized RunResult. It should not load datasets, score benchmark correctness, or hard-code provider-specific sandbox images.

Add a harness when you are integrating a new agent framework, coding assistant, GUI agent, direct model-call path, or tool-use strategy.

Harness File

Create a module under src/agentcompass/harnesses/. A typical harness defines a user-facing config, a runtime plan, and a BaseHarness subclass.

from dataclasses import dataclass
from typing import Any

from agentcompass.runtime.base import BaseHarness, EnvironmentSession
from agentcompass.runtime.component_config import RuntimeHarnessConfig
from agentcompass.runtime.models import (
    EnvironmentSpec,
    HarnessPlan,
    ModelSpec,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskStatus,
)
from agentcompass.runtime.registry import HARNESSES


@dataclass(slots=True)
class MyHarnessConfig(RuntimeHarnessConfig):
    max_steps: int = 30


@dataclass(slots=True)
class MyHarnessPlan(HarnessPlan):
    max_steps: int = 30


@HARNESSES.register()
class MyHarness(BaseHarness):
    id = "my_harness"
    description = "Runs MyAgent against text, file, or workspace tasks (official website: https://...)."
    config_class = MyHarnessConfig
    plan_class = MyHarnessPlan

    def supports(self, environment: EnvironmentSpec, model: ModelSpec) -> bool:
        return environment.id in {"host_process", "docker", "daytona", "modal"}

    async def start_session(self, env: EnvironmentSession, req: RunRequest, plan: MyHarnessPlan) -> dict[str, Any]:
        return {"env": env, "model": req.model.id, "max_steps": plan.max_steps}

    async def run_task(
        self,
        session: dict[str, Any],
        prepared: PreparedTask,
        req: RunRequest,
        plan: MyHarnessPlan,
    ) -> RunResult:
        return RunResult(
            task_id=prepared.task_id,
            category=prepared.category,
            status=TaskStatus.COMPLETED,
            final_answer="...",
            ground_truth=prepared.ground_truth,
            artifacts={"workspace": prepared.input.workspace},
        )

    async def close_session(self, session: dict[str, Any]) -> None:
        return None

Harness Responsibilities

Hook	Purpose
id / description	Stable CLI id and exported description. Include the agent/framework website when available.
supports()	Validates the environment and model protocol combination before execution starts. Raise a clear error for unsupported protocols.
config_class	Parses --harness-params and config defaults. Keep public knobs here, not in ad hoc metadata.
plan_class	Runtime execution state derived from config. Recipes may inspect or rewrite the final execution plan.
start_session()	Initializes clients, agent config files, persistent processes, or per-run state.
run_task()	Executes exactly one PreparedTask and returns RunResult. Preserve trajectory, artifacts, usage, and errors whenever possible.
close_session()	Releases external processes, clients, temporary sessions, or remote framework state.

Use prepared.input.workspace, prepared.input.files, prepared.input.media, prepared.input.tools, and prepared.input.messages instead of reaching back into benchmark internals. A harness should be reusable across benchmarks that emit the same prepared-task contract.

Register And Verify

Export the harness in src/agentcompass/harnesses/__init__.py:

from .my_harness import MyHarness  # noqa: F401

Then check discovery and a smoke run:

uv run agentcompass list harness

agentcompass run \
  scicode \
  my_harness \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add an Environment

An environment provider owns execution primitives: commands, file transfer, text IO, directory sync, endpoint discovery, and teardown. It should not know benchmark scoring rules or agent behavior.

Add an environment when AgentCompass needs to run the same benchmark/harness matrix on a new local process, container runtime, cloud sandbox, or remote execution provider.

Environment File

Create a module under src/agentcompass/environments/. Each provider usually has a session class, config dataclass, and BaseEnvironment implementation.

from dataclasses import dataclass
from pathlib import Path
from typing import Any

from agentcompass.runtime.base import BaseEnvironment, EnvironmentSession
from agentcompass.runtime.component_config import RuntimeEnvironmentConfig
from agentcompass.runtime.models import ExecResult, ExecutionPlan, RunRequest
from agentcompass.runtime.registry import ENVIRONMENTS


class MyEnvironmentSession(EnvironmentSession):
    default_workspace_root = "/workspace/"

    async def exec(
        self,
        command: list[str] | str,
        *,
        shell: bool = False,
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        timeout: float | None = None,
        detach: bool = False,
        flags: dict[str, Any] | None = None,
    ) -> ExecResult:
        command = self._validate_exec_command(command, shell=shell)
        # Map AgentCompass exec semantics onto the provider SDK here.
        return ExecResult(returncode=0, stdout="", stderr="")

    async def upload(self, src: str, dst: str) -> None: ...
    async def download(self, src: str, dst: str) -> None: ...
    async def write_text(self, path: str, content: str) -> None: ...
    async def read_text(self, path: str) -> str: ...
    async def upload_dir(self, src: Path | str, dst: str) -> None: ...
    async def download_dir(self, src: str, dst: Path | str) -> None: ...
    async def endpoint(self) -> str | None: ...


@dataclass(slots=True)
class MyEnvironmentConfig(RuntimeEnvironmentConfig):
    image: str = "python:3.12-slim"
    default_workspace_root: str = "/workspace/"


@ENVIRONMENTS.register()
class MyEnvironment(BaseEnvironment):
    id = "my_environment"
    config_class = MyEnvironmentConfig
    default_workspace_root = "/workspace/"

    async def open(self, req: RunRequest, plan: ExecutionPlan) -> MyEnvironmentSession:
        config = self.build_config(req, plan)
        return MyEnvironmentSession()

    async def close(self, env: EnvironmentSession) -> None:
        return None

Environment Responsibilities

Surface	Purpose
EnvironmentSession.exec()	Implements subprocess-like execution. Respect shell, cwd, env, timeout, detach, and return ExecResult with stdout, stderr, returncode, and timeout status.
File primitives	Implement upload, download, write_text, read_text, upload_dir, and download_dir so benchmark and harness code can stay provider-neutral.
endpoint()	Returns a reachable URL when the provider exposes one, otherwise None.
RuntimeEnvironmentConfig	Documents public provider params such as image, resources, region, credentials, or workspace root.
BaseEnvironment.open()	Builds the provider session from the recipe-adjusted ExecutionPlan. Always use plan.environment.params, not only the original request.
BaseEnvironment.close()	Terminates or detaches remote resources deterministically. It should tolerate partially initialized sessions.

Provider defaults belong in config/defaults.yaml when they are useful for normal users. Benchmark-specific image, snapshot, workspace, or resource mapping belongs in recipes.

Register And Verify

Export the environment from src/agentcompass/environments/__init__.py. Guard optional SDK imports if the dependency is not always installed:

try:
    from .my_environment import MyEnvironment  # noqa: F401
except ModuleNotFoundError as exc:
    if exc.name != "my_provider_sdk":
        raise

Then compile the provider and run a minimal command-oriented benchmark or harness smoke test:

python -m compileall src/agentcompass/environments/my_environment.py

agentcompass run \
  terminal_bench_2 \
  terminus2 \
  your-model \
  --env my_environment \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add a Recipe

A recipe is a compatibility layer between one benchmark family and one environment provider. It rewrites the per-task ExecutionPlan before the sandbox starts, usually to select images, workspaces, resource requests, snapshots, environment variables, or evaluation layout.

Add a recipe when task metadata already contains provider-relevant information and users should not have to pass it manually on the CLI.

Recipe File

Create a module under src/agentcompass/recipes/<benchmark_family>/. Keep one provider recipe per file when the provider behavior is meaningfully different.

from copy import deepcopy

from agentcompass.runtime.base import BaseRecipe
from agentcompass.runtime.models import ExecutionPlan, RunRequest, TaskSpec
from agentcompass.runtime.registry import RECIPES


@RECIPES.register()
class MyBenchmarkModalRecipe(BaseRecipe):
    id = "my_benchmark_modal"
    enabled_by_default = True

    def matches(self, req: RunRequest, task: TaskSpec, plan: ExecutionPlan) -> bool:
        return req.benchmark.id == "my_benchmark" and req.environment.id == "modal"

    def apply(self, plan: ExecutionPlan, req: RunRequest, task: TaskSpec) -> ExecutionPlan:
        _ = req
        updated = deepcopy(plan)
        params = dict(updated.environment.params)

        user_image = str(params.get("image") or "").strip()
        task_image = str(task.metadata.get("docker_image") or "").strip()
        if not user_image and not task_image:
            raise ValueError(f"{self.id} requires image or task.metadata.docker_image")

        params.setdefault("image", task_image)
        params.setdefault("default_workspace_root", "/workspace")
        updated.environment.params = params
        return updated

Recipe Responsibilities

Rule	Why it matters
Match narrowly	Check benchmark id, environment id, and required task metadata. Avoid recipes that fire for unrelated tasks.
Preserve user overrides	Use setdefault() or explicit compatibility checks when users pass image, resources, workspace, or credentials themselves.
Fail before sandbox startup	If a heavy benchmark requires a task image and none is available, raise a clear ValueError in apply().
Rewrite plans, not configs	Mutate a copied ExecutionPlan; do not mutate the original RunRequest.
Keep execution out	Recipes should not run commands, create sessions, evaluate results, or call model APIs.
Share helpers by benchmark family	Put shared layout helpers in recipes/<benchmark_family>/common.py when multiple providers need the same workspace mapping.

Recipes are applied automatically when execution.enabled_recipes is empty. Users can restrict recipe selection with --enabled-recipes.

Register And Verify

Export the recipe from the package initializer:

# src/agentcompass/recipes/my_benchmark/__init__.py
from .modal import MyBenchmarkModalRecipe  # noqa: F401

# src/agentcompass/recipes/__init__.py
from .my_benchmark import MyBenchmarkModalRecipe  # noqa: F401

Run a one-sample task and confirm the logs contain Recipe matched with the new recipe id:

agentcompass run \
  my_benchmark \
  my_harness \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["sample-1"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add an Analyzer

Analyzers run after a task produces a RunResult. They inspect trajectories, metrics, errors, latency, model output, or tool calls, then attach an AnalysisResult under analysis_result.<AnalyzerId> in each details JSON. Aggregate fields are rendered into analysis_summary.md and analysis_summary.json.

Add an analyzer when the logic is post-execution diagnosis or statistics. Do not put analyzer logic in a benchmark or harness if it can be rerun later with agentcompass analysis.

Analyzer File

Create a new file under src/agentcompass/analyzers/basic/, or create a dedicated sub-package for a larger analyzer family. Each analyzer must subclass BaseAnalyzer, register with @ANALYZERS.register(), and implement analysis().

from agentcompass.runtime.base import BaseAnalyzer
from agentcompass.runtime.models import AnalysisResult, AnalyzerCategory, RunResult
from agentcompass.runtime.registry import ANALYZERS


@ANALYZERS.register()
class MyAnalyzer(BaseAnalyzer):
    id = "MyAnalyzer"
    description = "Detects a custom trajectory pattern."
    category = AnalyzerCategory.BEHAVIOR
    datasets = []
    data_requirements = []
    conf = {"only_incorrect": False, "threshold": 0.0}
    distribution_fields = {
        "total_steps": "numeric_stats",
        "pattern_types": "value_counts",
    }

    async def analysis(self, task, prepared, result: RunResult, req, plan) -> AnalysisResult:
        if result is None or result.trajectory is None:
            return AnalysisResult(
                task_id=task.task_id,
                is_badcase=None,
                error="no trajectory available",
            )

        total_steps = len(result.trajectory.steps or [])

        return AnalysisResult(
            task_id=task.task_id,
            is_badcase=False,
            score=None,
            details={
                "total_steps": total_steps,
                "pattern_types": [],
            },
        )

Analyzer Attributes

Attribute	Purpose
id	Stable analyzer id. Used by the registry, CLI selection, details JSON, and summaries.
description	User-facing description exported by agentcompass list dump.
category	Summary grouping. Use AnalyzerCategory.ERROR, EFFICIENCY, BEHAVIOR, ABILITY, BASIC_BADCASE, or ENV_FRAMEWORK_ERROR.
datasets	Benchmark ids this analyzer supports. Empty list means all benchmarks.
data_requirements	JSONPath requirements checked before analysis. Missing fields skip the sample.
conf	Analyzer config merged from execution.analysis_params.<AnalyzerId>. Common keys include only_incorrect and threshold.
distribution_fields	Details fields to aggregate across tasks. Values are numeric_stats or value_counts.
base_analyzer / priority	Override mechanism for benchmark-specific analyzer variants. Higher priority wins within the same family.

AnalysisResult.is_badcase should be True for detected badcases, False for explicitly clean samples, and None for statistics-only analyzers that should not affect badcase ratios.

Aggregate Fields

Declare every cross-task summary field in distribution_fields:

Method	Use for	Summary output
numeric_stats	Numeric values or numeric lists.	count, min, mean, percentiles, and max.
value_counts	Strings or string lists.	frequency counts and ratios.

Only keys returned inside AnalysisResult.details can be aggregated.

Register And Verify

Export the analyzer from package initializers so registry import side effects can see it:

# src/agentcompass/analyzers/basic/__init__.py
from agentcompass.analyzers.basic.my_analyzer import MyAnalyzer  # noqa: F401

# src/agentcompass/analyzers/__init__.py
from agentcompass.analyzers.basic import MyAnalyzer  # noqa: F401

Then verify it appears in the component list:

uv run agentcompass list analyzer

Run it during evaluation:

agentcompass run \
  terminal_bench_2 \
  terminus2 \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY" \
  --enable-analysis \
  --analysis-params '{"analyzers":["MyAnalyzer"]}'

Or rerun it on existing results:

agentcompass analysis --input results/<benchmark>/<model>/<run_id> --analysis-params '{"analyzers":["MyAnalyzer"]}' --task_concurrency 8

Developer Notes

• Keep analyzer code read-only with respect to task execution. It should inspect persisted results, not mutate benchmark behavior.
• Use datasets for benchmark-specific analyzers instead of branching on unrelated task metadata.
• Use base_analyzer and priority when a benchmark-specific analyzer should override a generic analyzer.
• Return a descriptive error instead of raising when a sample cannot be analyzed because optional trajectory fields are missing.