feat(bundle): an artifacts/<name>/ folder can be a CLI, installed into agents like a generated one

src/agent_env/a2a_agent/a2a_agent.py

@@ -20,6 +20,7 @@
     NetworkPolicyUnsupportedError,
     port_bindings,
 )
+from agent_env.artifact.artifacts.cli import GATEWAY_ENV_FILE
 from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from agent_env.a2a_agent.object_transfer import (
     TRANSFER_TIMEOUT_SECONDS,
@@ -257,7 +258,7 @@ async def install_cli(deployed: "DeployedA2AAgent", cli_artifact: "CliArtifact",
         env_dir = posixpath.dirname(entrypoint_path)
 
         # write .env first — write_file_from_text creates the parent dir; write_file_from_object does not.
-        await sandbox.write_file_from_text(f"AGENT_ENV_GATEWAY_URL={gateway_url}\n", f"{env_dir}/.env")
+        await sandbox.write_file_from_text(f"AGENT_ENV_GATEWAY_URL={gateway_url}\n", f"{env_dir}/{GATEWAY_ENV_FILE}")
 
         for rel_path, file_artifact in cli_artifact.get_cli_files().get_file_artifacts().items():
             await sandbox.write_file_from_object(file_artifact.object_url, f"{target_root}/{rel_path}")

src/agent_env/artifact/artifacts/cli.py

@@ -1,24 +1,31 @@
 """CliArtifact — packages a CLI tool (entrypoint + supporting files) as a versioned artifact.
 
 A CliArtifact references one FileArtifactUniverse bundling every file in the CLI
-directory tree, colocated under a single object-store prefix (`artifacts/cli/<key_segment(id)>/<version>/`).
+directory tree, colocated under one object-store prefix per write (`ArtifactStore.attempt_prefix`).
 The `entrypoint` field is a relative path within that prefix that consumers
 (e.g. an agent gateway) chmod +x and symlink onto PATH.
 """
 
 from __future__ import annotations
 
+import posixpath
+import re
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Literal, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, Mapping, Optional
 
 from pydantic import ConfigDict, Field, model_serializer
 
 from agent_env.artifact.artifact import Artifact, _write_twin
-from agent_env.config import get_config
-from agent_env.store.ids import derive_id, key_segment
+from agent_env.store.ids import derive_id
 
 if TYPE_CHECKING:
     from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse
+    from agent_env.bundle.authoring import AuthoringContext
+
+# A command a bundle's CLI is installed as, at /opt/cli/<command_name>/.
+_COMMAND_NAME = re.compile(r"[A-Za-z0-9_][A-Za-z0-9._-]*")
+# The file installing a CLI writes the gateway URL to, beside its entrypoint.
+GATEWAY_ENV_FILE = ".env"
 
 
 class CliArtifact(Artifact):
@@ -34,6 +41,8 @@ class CliArtifact(Artifact):
     env_id: Optional[str] = Field(default=None, description="Source env id this CLI was generated from, if any")
     env_version: Optional[int] = Field(default=None, description="Source env version this CLI was generated from, if any")
 
+    toml_keys: ClassVar[dict[str, type]] = {"command_name": str, "entrypoint": str}  # what an artifact.toml sets
+
     # No return annotation: pydantic builds the serialization schema from one, and a dict drops the fields.
     @model_serializer(mode="wrap")
     def _serialize(self, handler: Any):
@@ -49,32 +58,32 @@ def put(
         *,
         command_name: str,
         entrypoint: str,
-        cli_dir: Path,
+        cli_dir: Optional[Path] = None,
+        files: Optional[Mapping[str, Path]] = None,
         env_id: Optional[str] = None,
         env_version: Optional[int] = None,
     ) -> "CliArtifact":
+        """Write the CLI in ``cli_dir``, or the ``files`` given by their paths in the CLI, ``entrypoint`` among them."""
         from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse
         from agent_env.artifact.store import get_artifact_store
 
-        if not cli_dir.is_dir():
-            raise ValueError(f"cli_dir {cli_dir} is not a directory")
-        entrypoint_path = cli_dir / entrypoint
-        if not entrypoint_path.is_file():
-            raise ValueError(f"entrypoint {entrypoint!r} not found at {entrypoint_path}")
-
-        files: dict[str, Path] = {}
-        for p in sorted(cli_dir.rglob("*")):
-            if p.is_file():
-                files[p.relative_to(cli_dir).as_posix()] = p
-        if not files:
-            raise ValueError(f"cli_dir {cli_dir} contains no files")
+        if (cli_dir is None) == (files is None):
+            raise ValueError("specify exactly one of cli_dir or files")
+        if cli_dir is not None:
+            if not cli_dir.is_dir():
+                raise ValueError(f"cli_dir {cli_dir} is not a directory")
+            if not (cli_dir / entrypoint).is_file():
+                raise ValueError(f"entrypoint {entrypoint!r} not found at {cli_dir / entrypoint}")
+            files = {p.relative_to(cli_dir).as_posix(): p for p in sorted(cli_dir.rglob("*")) if p.is_file()}
+        elif entrypoint not in files:
+            raise ValueError(f"entrypoint {entrypoint!r} isn't among the CLI's files")
 
         store = get_artifact_store()
         version = store.next_version(id)
-        key = f"{get_config().get_artifact_key_prefix()}artifacts/cli/{key_segment(id)}/{version}/"
-        cli_object_url = get_config().get_object_store_for(id).object_url(key)
+        cli_object_url = store.attempt_prefix(cls.model_fields["type"].default, id)
 
-        universe = FileArtifactUniverse.put_bundled(id=derive_id(id, "files"), files=files, prefix_url=cli_object_url)
+        universe = FileArtifactUniverse.put_bundled(id=derive_id(id, "files"), files=dict(files),
+                                                    prefix_url=cli_object_url)
 
         instance = cls(
             id=id,
@@ -88,6 +97,38 @@ def put(
         )
         return store.put_document(instance)
 
+    @classmethod
+    def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict:
+        """The keys of ``data``, an artifact.toml, a CLI takes, checked against the folder's files the way installing
+        it reads them. Raises BundleError listing every problem."""
+        fields, problems = ctx.accepted(data, **cls.toml_keys)
+        for key in cls.toml_keys:
+            if key not in data:
+                problems.append(ctx.config_problem(f"{key} is required"))
+        command_name, entrypoint = fields.get("command_name"), fields.get("entrypoint")
+        if command_name is not None and not _COMMAND_NAME.fullmatch(command_name):
+            problems.append(ctx.config_problem(
+                f"command_name must be a plain name, of letters, digits, '.', '_' and '-' and not starting with '.' or "
+                f"'-', since the CLI is installed at /opt/cli/<command_name>; not {command_name!r}"))
+        files = ctx.try_files()
+        if entrypoint is not None and files is not None:
+            gateway_env = posixpath.join(posixpath.dirname(entrypoint), GATEWAY_ENV_FILE)
+            if entrypoint not in files:
+                problems.append(ctx.config_problem(f"entrypoint {entrypoint!r} isn't one of this folder's files"))
+            elif any(name == gateway_env or name.startswith(f"{gateway_env}/") for name in files):
+                problems.append(f"{gateway_env}: installing the CLI writes the gateway URL to this file, beside its "
+                                "entrypoint; rename or remove the folder's own")
+        if problems:
+            ctx.refuse(problems)
+        return fields
+
+    @classmethod
+    def from_toml(cls, data: dict, ctx: AuthoringContext) -> "CliArtifact":
+        """Write the CLI in the bundle folder: every file its folder walk lists."""
+        fields = cls.accept_toml(data, ctx)
+        return cls.put(ctx.id, command_name=fields["command_name"], entrypoint=fields["entrypoint"],
+                       files=ctx.files())
+
     def get_cli_files(self) -> "FileArtifactUniverse":
         from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse
         return FileArtifactUniverse.get(self.cli_files_id)

src/agent_env/bundle/authoring.py

@@ -62,6 +62,13 @@ def dir(self) -> Path:
     def files(self) -> dict[str, Path]:
         return entry_files(self.bundle, self.entry)
 
+    def try_files(self) -> dict[str, Path] | None:
+        """``files``, or None when the folder can't be listed, which the plan's walk reports."""
+        try:
+            return self.files()
+        except BundleError:
+            return None
+
     def file(self) -> tuple[str, Path]:
         return entry_file(self.bundle, self.entry)
 

src/agent_env/bundle/plan.py

@@ -18,6 +18,7 @@
 from agent_env.a2a_agent import A2AAgent
 from agent_env.artifact import (
     Artifact,
+    CliArtifact,
     EnvironmentArtifact,
     EnvironmentUniverseArtifact,
     FileArtifact,
@@ -46,6 +47,7 @@
 from agent_env.store.routing import namespace_routing_enabled
 from agent_env.task import Task
 from agent_env.task_step.registry import get_task_step_registry
+from agent_env.task_step.task_step import TaskStep
 from agent_env.task_step.task_steps.load_artifact import LoadArtifactTaskStep
 
 from ._fs import fold, relative, with_article
@@ -68,7 +70,7 @@
 
 # The artifact types written from their folders, and what each one's write reads there, which the ledger hashes.
 _FOLDER_WALKS = {FileArtifact: one_file, FileArtifactUniverse: entry_files, EnvironmentArtifact: environment_files,
-                 EnvironmentUniverseArtifact: universe_files, SkillArtifact: entry_files}
+                 EnvironmentUniverseArtifact: universe_files, SkillArtifact: entry_files, CliArtifact: entry_files}
 # The env types whose from_toml reads only the toml and what it names, so a type built by one of them is written from
 # its env.toml and the ledger can list what it's made from.
 _ENV_FROM_TOMLS = (Env, MCPServerEnv, WebsiteEnv, MultiEnv)
@@ -132,6 +134,7 @@ def __init__(self, resolved: ResolvedBundle):
         self.outputs = {(output.kind.value, output.id): (entry.entry, output)
                         for entry in resolved.entries for output in entry.outputs}
         self.accepted: dict[_Key, dict] = {}  # each toml a type checks, as it accepted it
+        self.steps: dict[_Key, list[TaskStep]] = {}  # each task's steps that build, as a write will build them
 
     def _add(self, key: _Key, source: ResolvedEntry | BundleImage, needs: list[_Key]) -> None:
         """Record a write in bundle order, a built image just before the entry that builds it."""
@@ -145,6 +148,7 @@ def plan(self, tasks: Sequence[str], evals: Sequence[str]) -> Plan:
         store_refs, store_latest = self._check_store_refs(closure)
         self._build_tasks(closure)
         self._check_files(closure)
+        self._check_cli_loads(closure)
         self._check_env_types(closure)
         self._check_multi_names(closure, store_latest)
         self._check_universes(closure, store_latest)
@@ -163,6 +167,7 @@ def check(self) -> None:
         self._check_store_refs(everything, stores=False)
         self._build_tasks(everything)
         self._check_files(everything)
+        self._check_cli_loads(everything)
         self._check_universes(everything, None)
         self._check_loaded_environments(everything, None)
         if self.problems:
@@ -370,6 +375,41 @@ def _check_files(self, closure: set[_Key]) -> None:
             except BundleError as e:
                 self.problems.extend(e.problems)
 
+    def _check_cli_loads(self, closure: set[_Key]) -> None:
+        """Each load of one of the bundle's CLIs names what installing it needs, an agent and an env whose gateway it
+        reads, and no two of a task's loads install one command into one agent, where the second would replace the
+        first."""
+        registry = get_artifact_registry()
+        clis = {source.entry.id: source.entry for source in self.nodes.values()
+                if isinstance(source, ResolvedEntry) and source.entry.kind is BundleKind.ARTIFACT
+                and issubclass(registry.get(canonical_type(source.entry.type)) or object, CliArtifact)}
+        for key in sorted(closure, key=self.rank.__getitem__):
+            installed: dict[tuple[str, str], tuple[Any, str]] = {}
+            for step in self.steps.get(key, ()):
+                if not isinstance(step, LoadArtifactTaskStep):
+                    continue
+                loaded = [clis[ref["id"]] for ref in step.artifacts if ref.get("id") in clis]
+                if not loaded:
+                    continue
+                where = f"{self._path(self.nodes[key].entry)}: step {step.id!r}"
+                if missing := [name for name in ("agent_name", "env_id") if not getattr(step, name)]:
+                    names, it = " and ".join(repr(cli.name) for cli in loaded), "them" if len(loaded) > 1 else "it"
+                    self.problems.append(f"{where}: loading the CLI{'s' * (it == 'them')} {names} installs {it} into "
+                                         f"an agent and points {it} at an env's gateway, so the step needs "
+                                         f"{' and '.join(missing)}")
+                    continue
+                for cli in loaded:
+                    command = self.accepted.get(_key(cli), {}).get("command_name")
+                    if command is None:
+                        continue  # its artifact.toml's problem is reported by the file check
+                    if (step.agent_name, command) in installed:
+                        first, name = installed[step.agent_name, command]
+                        self.problems.append(f"{where}: installs the CLI {cli.name!r} as {command!r} in agent "
+                                             f"{step.agent_name!r}, where step {first!r} already installed {name!r} "
+                                             "under that command; give one another command_name")
+                    else:
+                        installed[step.agent_name, command] = (step.id, cli.name)
+
     def _env_accept(self, entry: BundleEntry) -> Any:
         """How ``entry``'s env type checks its env.toml, when it's an env written from one and checks it at all."""
         cls = get_env_registry().get(entry.type) if entry.kind is BundleKind.ENV else None
@@ -571,6 +611,7 @@ def _build_tasks(self, closure: set[_Key]) -> None:
                     failed = True
                     self.problems.append(f"{self._path(source.entry)}: step {step.get('id')!r}: "
                                          f"{registry[step['type']].type} can't read it ({type(e).__name__}: {e})")
+            self.steps[key] = steps
             if not failed:
                 try:
                     Task(id=source.entry.id, version=None, steps=steps)

tst/data/a2a_agent/agent.py

@@ -12,12 +12,16 @@
 
 It installs skills under ``~/skills/<name>/``, a bundle's files downloaded through their read grants, and a
 ``list-skills`` line in its prompt adds a ``skill <name>: <path>=<sha256 prefix>, ...`` line for each one.
+A ``run <command>`` line has it run that command in its container and add a ``ran <command>: <output>`` line.
 """
 
 import asyncio
 import base64
+import contextlib
 import hashlib
+import os
 import re
+import signal
 from pathlib import Path
 from typing import Any
 from urllib.parse import urlsplit
@@ -47,6 +51,8 @@
 
 SEND_FILE = "send-file "
 LIST_SKILLS = "list-skills"
+RUN = "run "
+RUN_TIMEOUT_SECONDS = 60
 SKILLS_DIR = Path.home() / "skills"
 LIVE_STEPS = re.compile(r"live-steps (\d+)")
 LIVE_STEP_SECONDS = 1
@@ -79,6 +85,26 @@ async def _read(part: FilePart) -> str:
     return f"read {part.name} over {scheme}: {data.decode(errors='replace').strip()}"
 
 
+async def _ran(command: str) -> str:
+    """``command``'s output, stdout and stderr together, run by the shell in a process group of its own, which is
+    stopped with whatever it started if the command times out or the request is cancelled."""
+    process = await asyncio.create_subprocess_shell(command, stdout=asyncio.subprocess.PIPE,
+                                                    stderr=asyncio.subprocess.STDOUT, start_new_session=True)
+    output = None
+    try:
+        output, _ = await asyncio.wait_for(process.communicate(), RUN_TIMEOUT_SECONDS)
+    except asyncio.TimeoutError:
+        pass
+    finally:
+        if process.returncode is None:
+            with contextlib.suppress(ProcessLookupError):
+                os.killpg(process.pid, signal.SIGKILL)
+            await process.wait()
+    if output is None:
+        return f"ran {command}: timed out after {RUN_TIMEOUT_SECONDS}s"
+    return f"ran {command}: {output.decode(errors='replace').strip()}"
+
+
 def _skills() -> list[str]:
     """One line per installed skill: each file's path and the first 12 hex digits of its sha256."""
     if not SKILLS_DIR.is_dir():
@@ -109,6 +135,8 @@ async def run(self, request: TaskRequest[EchoAgentConfig]) -> TaskResult:
         reply = "\n".join([reply, *[await _read(part) for part in request.parts if isinstance(part, FilePart)]])
         if LIST_SKILLS in prompt.splitlines():
             reply = "\n".join([reply, *_skills()])
+        runs = [line.removeprefix(RUN).strip() for line in prompt.splitlines() if line.startswith(RUN)]
+        reply = "\n".join([reply, *[await _ran(command) for command in runs]])
         sends = [line.removeprefix(SEND_FILE).strip() for line in prompt.splitlines() if line.startswith(SEND_FILE)]
         files = [FilePart(uri=uri, name=uri.rsplit("/", 1)[-1], mime_type="text/plain") for uri in sends]
         request.trajectory.append({"type": "echo", "input": prompt, "output": reply})