Task Step Plugins
Write task steps in a package, register them with an entry point and run them in a task, with OpenCiv3's match and recording steps as the example
agentenv_openciv3.steps ships three step types: openciv3_match seats the agents a run deployed,
and any people, in a new game; openciv3_await_game waits for a game people play to end; and
save_env_recording stores the game's recording. This page follows the first and the last, which run
against the openciv3 env, or openciv3-live from Environment Plugins.
When to Write a Step
Write a step when a task needs something done between deploy and teardown that no step in
More task steps does. apply_server_config can start a game with fixed
args, as the smoke task does, but can't seat the deployed agents, and no built-in step stores the
files a recording extension returns.
The Step Contract
A step is a TaskStep subclass, as Your own step shows.
save_env_recording has fields of its own and passes the base fields to super().__init__:
class SaveEnvRecordingTaskStep(TaskStep):
type: ClassVar[str] = "save_env_recording"
entity_refs = (EntityRef.env("env_id"),)
def __init__(self, id: str, version: int | None, env_id: str, extension_uri: str = RECORDING_EXTENSION,
formats: list[str] | None = None, view: str = "spectator", timeout_seconds: int = 300,
depends_on: list | None = None, fail_task_on_error: bool = False):
super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
self.env_id = env_id
# and the other four fields
def to_dict(self) -> dict:
return {**super().to_dict(), "env_id": self.env_id, "extension_uri": self.extension_uri,
"formats": self.formats, "view": self.view, "timeout_seconds": self.timeout_seconds}
@classmethod
def from_dict(cls, data: dict) -> SaveEnvRecordingTaskStep:
return cls(**{**cls._base_from_dict(data), "fail_task_on_error": data.get("fail_task_on_error", False)},
env_id=data["env_id"], extension_uri=data.get("extension_uri", RECORDING_EXTENSION),
formats=data.get("formats"), view=data.get("view", "spectator"),
timeout_seconds=data.get("timeout_seconds", 300))type is the string task JSON names and every stored step document carries, so the class needs its
own. _base_from_dict returns id, version, depends_on and fail_task_on_error, defaulting
the last to true; this step defaults it to false, so a failed recording never fails the task.
Core attaches retry_config itself, so the constructor does not take it.
Preflight
preflight() returns one message per problem it can find before a run, and must not need a sandbox
or a deployed env. The base class returns [], and neither step overrides it: deploy_env's
preflight already loads the env, and the agents openciv3_match seats exist only once deployed, so
its execute checks them:
RuntimeError: agents ['haiku'] are not deployed in this run (deployed: ['opus', 'sonnet'])Declare the Env Reference
entity_refs names the fields that hold an env, agent or artifact id, so a tool that rewrites ids
before a run, such as the bundle resolver, can find them. It is not inherited, and when a bundle
passes one of its own names to a step type that leaves it None, the resolver reports that the type
doesn't declare its entity_refs, so it can't be rewritten. Nothing reads it at run time.
Reach the Deployed Env
Each step finds the env instance in context.deployed_envs, where deploy_env put it, then the card
that advertises the extension it needs: the env's own, or a child's behind a gateway.
def _extension_card(deployed: DeployedEnv, uri: str) -> dict:
own = deployed.environment_card or {}
card = next((c for c in [own, *(own.get("children_environments") or [])] if client.find_extension(c, uri)), None)
if card is None:
raise RuntimeError(f"env {deployed.env_id!r} does not advertise {uri}")
return cardRather than call MCP tools as an agent would, it passes that card to client.invoke_extension, which
reads the extension's endpoint from it. Core derives environment_url from the card URL the
environment provider recorded, so the step knows
nothing about the machine behind it.
Store the Recording
execute asks the urn:openciv3:recording/v1 extension for the game's MP4 and HTML replay, stores
each as a file artifact, and records them in metadata["recordings"] by step id. The bytes go to
the object store and only artifact ids to the context, which is persisted after every step.
async def execute(self, context: TaskStepContext) -> TaskStepContext:
deployed = _deployed_env(context, self.env_id)
card = _extension_card(deployed, self.extension_uri)
args = {"view": self.view, **({"formats": self.formats} if self.formats is not None else {})}
result = await client.invoke_extension(deployed.environment_url, card, self.extension_uri, args,
timeout=self.timeout_seconds)
stem = f"{context.metadata.get('task_id', self.env_id)}-recording-{context.instance_id or uuid.uuid4().hex}"
saved = []
for f in result["files"]:
content = base64.b64decode(f["base64"])
artifact = await asyncio.to_thread(
FileArtifact.put_bytes, f"{stem}.{f['name'].partition('.')[2]}",
description=f"Recording of env {self.env_id!r}: {f['name']}", filename=f["name"],
content=content, content_type=f["content_type"])
saved.append({"name": f["name"], "artifact_id": artifact.id, "version": artifact.version,
"bytes": len(content), "content_type": f["content_type"]})
context.metadata.setdefault("recordings", {})[self.id] = saved
return contextput_bytes is synchronous, so the step runs it in a thread to keep other steps of the run moving.
With the store from Registry Plugins, a registered task's object URLs start cas://openciv3/.
Write a Verifier
A verifier is an ordinary step that writes metadata["verifications"][verifier_id] with a score
from 0 to 1 and the results rows behind it. The plugin writes none: its tasks use the built-in
env_outcome_verifier, which runs verify(mcp_url) from a script the bundle ships as a file
artifact, such as victor-verifier, which names a match's winner.
{"id": "grade", "type": "env_outcome_verifier", "env_id": "openciv3", "file_artifact_id": "victor-verifier",
"verifier_id": "victor", "score_aggregator": "weighted_average", "depends_on": ["opus", "sonnet", "haiku"]}agent-env task run prints the rows and score of every entry in verifications, so a plugin
verifier needs no output code of its own.
Register the Entry Points
Each step is an entry point in the agent_env.task_steps group, named after its type:
[project.entry-points."agent_env.task_steps"]
openciv3_match = "agentenv_openciv3.steps:OpenCiv3MatchTaskStep"
openciv3_await_game = "agentenv_openciv3.steps:OpenCiv3AwaitGameTaskStep"
save_env_recording = "agentenv_openciv3.steps:SaveEnvRecordingTaskStep"The name must equal the type, because documents are written and read by it. A class under another
name, one that keeps the base type, or one that leaves execute or from_dict unimplemented is
reported failed (invalid-plugin) and skipped, with a reason such as
OpenCiv3MatchTaskStep has type 'openciv3_match'; its entry point must be named 'openciv3_match'.
Confirm the Plugin Loaded
agent-env plugin show lists each contribution with its status:
agentenv-openciv3 0.1.0
task step openciv3_await_game agentenv_openciv3.steps:OpenCiv3AwaitGameTaskStep active
task step openciv3_match agentenv_openciv3.steps:OpenCiv3MatchTaskStep active
task step save_env_recording agentenv_openciv3.steps:SaveEnvRecordingTaskStep activeok: 2 plugin package(s), 6 contribution(s), all in effectUse the Steps in a Task
The bundle's three-agents-quick task seats three agents for 10 turns, then grades the game and
saves its recording in parallel. One agent's steps are shown:
[
{"id": "deploy", "type": "deploy_env", "env_id": "openciv3", "ttl_seconds": 7200},
{"id": "agent-opus", "type": "deploy_agent", "agent_name": "opus", "env_ids": ["openciv3"], "depends_on": ["deploy"]},
{"id": "match", "type": "openciv3_match", "env_id": "openciv3", "turns": 10,
"civs": {"opus": "Rome", "sonnet": "Greece", "haiku": "Egypt"}, "depends_on": ["agent-opus", "agent-sonnet", "agent-haiku"]},
{"id": "opus", "type": "prompt_agent", "agent_name": "opus", "model": "anthropic/claude-opus-5-5",
"prompt": "You lead a civilization in a game of OpenCiv3, ...", "depends_on": ["match"]},
{"id": "recording", "type": "save_env_recording", "env_id": "openciv3", "timeout_seconds": 1200,
"depends_on": ["opus", "sonnet", "haiku"]}
]agent-env task create runs each step's preflight(), so an unregistered env stops the create
before anything is saved:
Preflight found 1 problem(s):
- deploy_env 'deploy': env 'openciv3' can't be loaded: Env openciv3 not found
Nothing was saved. Re-run with --skip-validation to save anyway.agent-env run openciv3 --task three-agents-quick runs it from the bundle; it passed with grade 1
in 182.4s. The smoke task needs no model and ends with the same save_env_recording step:
[tasks/smoke.json] step 5/5 recording (save_env_recording)
[tasks/smoke.json] step 5/5 recording done in 4.1s
[tasks/smoke.json] passed in 8.7s
Tasks:
tasks/smoke.json v1: passed (grade: 1), 8.7s, instance @local/agentenv-openciv3/openciv3/smoke-49qpygc4
Tore down 1 sandbox.Its context holds what the step wrote:
{
"recordings": {
"recording": [
{"name": "openciv3-seed7.mp4", "artifact_id": "@local/agentenv-openciv3/openciv3/smoke-recording-@local/agentenv-openciv3/openciv3/smoke-49qpygc4.mp4", "version": 1, "bytes": 184289, "content_type": "video/mp4"},
{"name": "openciv3-seed7.html", "artifact_id": "@local/agentenv-openciv3/openciv3/smoke-recording-@local/agentenv-openciv3/openciv3/smoke-49qpygc4.html", "version": 1, "bytes": 933081, "content_type": "text/html"}
]
}
}Last updated on