feat(agents): select an agent's own tools at deploy, and claude-code's

packages/agentenv-protocol/README.md

@@ -1007,6 +1007,23 @@ class MyAgentConfig(AgentConfig):
     provider_token: WriteOnly[str | None] = None
 ```
 
+An agent that can turn its own tools off, such as its shell, file or web tools, takes `builtin_tools`, typed
+`BuiltinTools` over a `Literal` of its tools' names, so its card's schema names them. A task selects them on
+`deploy_agent`: `{"disable": "*"}` turns them all off, `{"disable": [...]}` the ones named, and `{"only": [...]}` keeps
+the ones named; the agent applies `request.config.builtin_tools` to its runtime. The MCP tools AgentEnv gives it are not
+its own and stay. A selection for an agent whose config has
+no `builtin_tools`, or that names a tool its schema doesn't, is refused before the deploy sends it anything.
+
+```python
+from typing import Literal
+
+from agentenv_protocol.a2a_agent import AgentConfig, BuiltinTools
+
+
+class MyAgentConfig(AgentConfig):
+    builtin_tools: BuiltinTools[Literal["shell", "read_file", "web_fetch"]] | None = None
+```
+
 `output_format` is author-owned in v1 rather than a field on the base
 `AgentConfig`. An agent that supports structured output must declare the field
 on its config subclass, apply it to its model or runtime, and return the value

packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/__init__.py

@@ -102,6 +102,7 @@
 from .staging import STAGING_ENDPOINT, STAGING_V1_URI, StagingStore, staging_routes
 from .tasks.v1 import (
     AgentConfig,
+    BuiltinTools,
     AgentRunResult,
     DataPart,
     FilePart,
@@ -154,6 +155,7 @@
     "AgentIdentity",
     "AgentRunResult",
     "AttributionProbeResponse",
+    "BuiltinTools",
     "BundleSkillRequest",
     "ChangelogIncrement",
     "ContextEventsTrajectoryRequest",

packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/tasks/v1.py

@@ -17,7 +17,7 @@
 from typing import Annotated, Any, Generic, Literal, TypeAlias, TypeVar, Union
 
 from a2a.types import TaskStatusUpdateEvent
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 logger = logging.getLogger(__name__)
 
@@ -127,6 +127,26 @@ class AgentConfig(BaseModel):
     timeout_seconds: int = 600
 
 
+ToolName = TypeVar("ToolName", bound=str)
+
+
+class BuiltinTools(BaseModel, Generic[ToolName]):
+    """Which of an agent's own tools it may use, as a task selects them: ``disable`` turns off the ones it names, or
+    all of them with ``"*"``; ``only`` keeps the ones it names. An agent's config takes it as
+    ``builtin_tools: BuiltinTools[<a Literal of its tools>] | None``, so its card's agent-config schema names them."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    disable: Literal["*"] | list[ToolName] | None = None
+    only: list[ToolName] | None = None
+
+    @model_validator(mode="after")
+    def _one_of(self) -> BuiltinTools:
+        if (self.disable is None) == (self.only is None):
+            raise ValueError("builtin_tools takes exactly one of disable and only")
+        return self
+
+
 ConfigT = TypeVar("ConfigT", bound=AgentConfig)
 
 

packages/agentenv-protocol/tests/test_a2a_agent.py

@@ -6,7 +6,7 @@
 import time
 from dataclasses import FrozenInstanceError
 from functools import wraps
-from typing import Any, Union
+from typing import Any, Literal, Union
 from unittest.mock import MagicMock
 
 import httpx
@@ -44,6 +44,7 @@
     AgentEnvAgent,
     AgentIdentity,
     AttributionProbeResponse,
+    BuiltinTools,
     BundleSkillRequest,
     ContextObjectTrajectoryRequest,
     ContextTrajectoryRequest,
@@ -1774,6 +1775,26 @@ async def run(self, request: TaskRequest[RuntimeConfig]) -> TaskResult:
     assert "default" not in schema["$defs"]["NestedConfig"]["properties"]["token"]
 
 
+def test_builtin_tools_names_the_agents_own_tools_on_its_card_and_takes_one_selection() -> None:
+    class RuntimeConfig(AgentConfig):
+        builtin_tools: BuiltinTools[Literal["Bash", "Read", "WebFetch"]] | None = None
+
+    @a2a_agent(identity=AgentIdentity(name="tooled", description="test", version="1"), config=RuntimeConfig)
+    class Agent(AgentEnvAgent):
+        async def run(self, request: TaskRequest[RuntimeConfig]) -> TaskResult:
+            return TaskResult.text("ok")
+
+    with TestClient(Agent().create_app()) as client:
+        card = client.get("/.well-known/agent-card.json").json()
+        statuses = [_operation(client, card, AGENT_CONFIG_V1.uri, "set", {"builtin_tools": selection}).status_code
+                    for selection in ({"disable": "*"}, {"only": ["Read"]}, {"disable": ["Glob"]},
+                                      {"disable": "*", "only": ["Read"]}, {})]
+
+    request = _extension(card, AGENT_CONFIG_V1.uri)["params"]["methods"]["set"]["request"]
+    assert "builtin_tools" in request["supported"] and '"enum": ["Bash", "Read", "WebFetch"]' in json.dumps(request)
+    assert statuses == [200, 200, 400, 400, 400]
+
+
 def test_validation_errors_do_not_echo_rejected_values() -> None:
     secret = "sk-live-SUPERSECRET"
 

src/agent_env/a2a_agent/validator.py

@@ -185,6 +185,7 @@ async def _run(
             VerifyA2AAgentCardStep,
             VerifyA2AAgentConfigIdentityStep,
             VerifyA2AAgentMCPStep,
+            VerifyA2ABuiltinToolsStep,
             VerifyA2AModalitiesStep,
             VerifyA2APeerAgentsStep,
             VerifyA2ARoleStep,
@@ -279,6 +280,7 @@ async def _run(
             VerifyA2AAgentCardStep(id=f"{task_id}-card", version=None, a2a_agent_id=agent.id, a2a_agent_version=agent.version, depends_on=deploy_agent_dep, fail_task_on_error=False),
             VerifyA2AAgentConfigIdentityStep(id=f"{task_id}-agent-config-identity", version=None, a2a_agent_id=agent.id, a2a_agent_version=agent.version, depends_on=deploy_agent_dep, fail_task_on_error=False),
             VerifyA2ARoleStep(id=f"{task_id}-role", version=None, a2a_agent_id=agent.id, a2a_agent_version=agent.version, depends_on=deploy_agent_dep, fail_task_on_error=False),
+            VerifyA2ABuiltinToolsStep(id=f"{task_id}-builtin-tools", version=None, a2a_agent_id=agent.id, a2a_agent_version=agent.version, depends_on=deploy_agent_dep, fail_task_on_error=False),
             VerifyA2ASystemPromptStep(id=f"{task_id}-system-prompt", version=None, a2a_agent_id=agent.id, a2a_agent_version=agent.version, depends_on=deploy_agent_dep, fail_task_on_error=False),
             PromptAgentTaskStep(id=protocol_prompt_id, version=None, prompt="Hello, respond with OK.", prompt_id=protocol_prompt_id, timeout_seconds=120, depends_on=deploy_agent_dep, fail_task_on_error=False),
             VerifyCoreA2AProtocolStep(id=f"{task_id}-protocol", version=None, a2a_agent_id=agent.id, prompt_id=protocol_prompt_id, depends_on=[TaskStepDependency(task_step_id=protocol_prompt_id)], fail_task_on_error=False),

src/agent_env/agents/README.md

@@ -21,7 +21,12 @@ the folder changes. A run that deploys the agent registers that image when it's
 pulls it by its digest and builds it from this folder only when it can't, and a provider that runs images by name can
 run it. The CLI's version is the Dockerfile's
 `CLAUDE_CODE_VERSION`. The image gives the scripts the model runs
-`node`, without npm or npx, and the Python packages `model-requirements.txt` pins.
+`node`, without npm or npx, and the Python packages `model-requirements.txt` pins. A `deploy_agent` `builtin_tools`
+selection picks the CLI's own tools: `{"disable": "*"}` and `{"only": [...]}` become its `--tools` list, a list to
+turn off becomes `--disallowedTools`, which keeps the CLI's defaults for the rest, and any selection adds
+`--strict-mcp-config`, so its MCP servers are only the task env's. The names are `ClaudeCodeTool`'s in
+`claude_bridge.py`: every tool the pinned CLI has, though which it offers by default depends on the model. A test fails
+when the Dockerfile's `CLAUDE_CODE_VERSION` moves, until `ClaudeCodeTool` and `TOOLS_CLI_VERSION` move with it.
 
 `install_agent` installs it too, into a container a task already runs or onto a `vm` sandbox's Linux host, with
 `install.sh` from its folder: Claude Code's native build for the target's CPU and libc, uv, and the agent's requirements

src/agent_env/agents/agents/claude-code/a2a_server.py

@@ -73,6 +73,7 @@
 )
 from claude_bridge import (
     CLAUDE_STATE_DIR,
+    ClaudeCodeTools,
     SESSION_JSONL_DIR,
     WORKSPACE_DIR,
     kill_running_clis,
@@ -134,6 +135,8 @@ class ClaudeCodeConfig(AgentConfig):
     effort: str | None = None
     max_turns: int | None = None
     max_thinking_tokens: int | None = None
+    # Which of the CLI's own tools it may use. With any selection, its MCP servers are only the ones agent-env gives it.
+    builtin_tools: ClaudeCodeTools | None = None
 
 
 def _rewrite_session_ids(jsonl_text: str, session_id: str) -> tuple[list[str], int]:
@@ -488,6 +491,7 @@ async def _turn(self, request: TaskRequest[ClaudeCodeConfig], cfg: ClaudeCodeCon
                 effort=cfg.effort,
                 max_turns=cfg.max_turns,
                 max_thinking_tokens=cfg.max_thinking_tokens,
+                builtin_tools=cfg.builtin_tools,
                 event_callback=request.trajectory.append,
             )
         except Exception as e:

src/agent_env/agents/agents/claude-code/claude_bridge.py

@@ -16,6 +16,9 @@
 from dataclasses import dataclass, field, replace
 from functools import partial
 from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, model_validator
 
 from kit import in_thread_to_its_end
 
@@ -52,6 +55,33 @@ def project_dir(workspace: Path) -> Path:
 MODEL_HEADERS_ENV = "LITELLM_EXTRA_HEADERS"
 
 
+# The CLI version ClaudeCodeTool names the tools of; a test holds it to the Dockerfile's CLAUDE_CODE_VERSION.
+TOOLS_CLI_VERSION = "2.1.280"
+# The CLI's own tools, by the names its model sees. Which of them it offers by default depends on the model, and its MCP
+# resource tools come only with MCP servers.
+ClaudeCodeTool = Literal["Agent", "Bash", "CronCreate", "CronDelete", "CronList", "Edit", "EnterWorktree",
+                         "ExitWorktree", "ListAgents", "ListMcpResourcesTool", "NotebookEdit", "Read",
+                         "ReadMcpResourceDirTool", "ReadMcpResourceTool", "ReportFindings", "ScheduleWakeup",
+                         "SendMessage", "Skill", "TaskCreate", "TaskGet", "TaskList", "TaskStop", "TaskUpdate",
+                         "WebFetch", "WebSearch", "Workflow", "Write"]
+
+
+class ClaudeCodeTools(BaseModel):
+    """The protocol's ``BuiltinTools[ClaudeCodeTool]``, which the image's protocol floor doesn't have yet: the same
+    fields and checks, so the card's schema is the same."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    disable: Literal["*"] | list[ClaudeCodeTool] | None = None
+    only: list[ClaudeCodeTool] | None = None
+
+    @model_validator(mode="after")
+    def _one_of(self) -> ClaudeCodeTools:
+        if (self.disable is None) == (self.only is None):
+            raise ValueError("builtin_tools takes exactly one of disable and only")
+        return self
+
+
 def session_jsonl_path(session_id: str) -> Path:
     return SESSION_JSONL_DIR / f"{session_id}.jsonl"
 
@@ -383,6 +413,7 @@ async def run_claude(
     effort: str | None = None,
     max_turns: int | None = None,
     max_thinking_tokens: int | None = None,
+    builtin_tools: ClaudeCodeTools | None = None,
     event_callback: Callable[[dict], None] | None = None,
 ) -> ClaudeResult:
     """Run one turn, resuming the CLI session when a transient upstream failure cuts it short; a resume spends what
@@ -395,7 +426,7 @@ async def run_claude(
         _spawn_claude, model=model, allowed_tools=allowed_tools, mcp_servers=mcp_servers,
         system_prompt=system_prompt, output_format=output_format, project_id=project_id, task_id=task_id,
         role=role, effort=effort, max_turns=max_turns, max_thinking_tokens=max_thinking_tokens,
-        event_callback=event_callback,
+        builtin_tools=builtin_tools, event_callback=event_callback,
     )
     attempt_prompt, attempt_resume, attempt_session = prompt, resume, session_id
     trajectory: list[dict] = []
@@ -572,8 +603,11 @@ def _with_1m_window(model: str) -> str:
 
 def _command(session_id: str, resume: bool, *, model: str | None, allowed_tools: list[str] | None,
              mcp_servers: dict[str, dict] | None, system_prompt: str | None, output_format: dict | None,
-             role: str | None, effort: str | None, max_turns: int | None) -> list[str]:
-    """The CLI's argv. The prompt goes on stdin, since one argument over 128 KiB fails exec."""
+             role: str | None, effort: str | None, max_turns: int | None,
+             builtin_tools: ClaudeCodeTools | None = None) -> list[str]:
+    """The CLI's argv. The prompt goes on stdin, since one argument over 128 KiB fails exec. ``builtin_tools`` selects
+    the CLI's own tools: ``{"disable": "*"}`` offers none, ``{"only": [...]}`` just those, and ``{"disable": [...]}``
+    the CLI's defaults but those; with any selection, its MCP servers are only ``mcp_servers``."""
     cmd = ["claude", "-p", "--output-format", "stream-json", "--verbose", "--dangerously-skip-permissions",
            "--setting-sources", "user"]
     cmd += ["--resume", session_id] if resume else ["--session-id", session_id]
@@ -593,6 +627,12 @@ def _command(session_id: str, resume: bool, *, model: str | None, allowed_tools:
         cmd += ["--json-schema", json.dumps(output_format["schema"])]
     if allowed_tools:
         cmd += ["--allowedTools", ",".join(allowed_tools)]
+    if builtin_tools is not None:
+        if builtin_tools.only is not None or builtin_tools.disable == "*":
+            cmd += ["--tools", ",".join(builtin_tools.only or [])]
+        elif builtin_tools.disable:
+            cmd += ["--disallowedTools", ",".join(builtin_tools.disable)]
+        cmd.append("--strict-mcp-config")
     if mcp_servers:
         def entry(info: dict) -> dict:
             server: dict = {"type": "http", "url": info["url"]}
@@ -622,11 +662,12 @@ async def _spawn_claude(
     effort: str | None = None,
     max_turns: int | None = None,
     max_thinking_tokens: int | None = None,
+    builtin_tools: ClaudeCodeTools | None = None,
     event_callback: Callable[[dict], None] | None = None,
 ) -> ClaudeResult:
     cmd = _command(session_id, resume, model=model, allowed_tools=allowed_tools, mcp_servers=mcp_servers,
                    system_prompt=system_prompt, output_format=output_format, role=role, effort=effort,
-                   max_turns=max_turns)
+                   max_turns=max_turns, builtin_tools=builtin_tools)
     logger.info(f"Running claude CLI (resume={resume}, session_id={session_id}, model={model})")
     WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)
     proc = await _started(asyncio.create_subprocess_exec(

src/agent_env/task/episode/rules.py

@@ -139,9 +139,10 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s
     "peer_agents": _peer_agents,
     "rubrics_verifier": _rubrics_verifier,
     **dict.fromkeys((
-        "verify_a2a_agent_card", "verify_a2a_agent_config_identity", "verify_a2a_agent_mcp", "verify_a2a_core_protocol",
-        "verify_a2a_install", "verify_a2a_modalities", "verify_a2a_peer_agents", "verify_a2a_role",
-        "verify_a2a_skill_config", "verify_a2a_snapshot", "verify_a2a_system_prompt", "verify_a2a_trajectory",
+        "verify_a2a_agent_card", "verify_a2a_agent_config_identity", "verify_a2a_agent_mcp", "verify_a2a_builtin_tools",
+        "verify_a2a_core_protocol", "verify_a2a_install", "verify_a2a_modalities", "verify_a2a_peer_agents",
+        "verify_a2a_role", "verify_a2a_skill_config", "verify_a2a_snapshot", "verify_a2a_system_prompt",
+        "verify_a2a_trajectory",
     ), _validates_an_agent),
 }
 

src/agent_env/task_step/registry.py

@@ -69,6 +69,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]:
     from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_skill_config import VerifyA2ASkillConfigStep
     from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_snapshot import VerifyA2ASnapshotStep
     from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_trajectory import VerifyA2ATrajectoryStep
+    from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_builtin_tools import VerifyA2ABuiltinToolsStep
     from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_role import VerifyA2ARoleStep
     from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_system_prompt import VerifyA2ASystemPromptStep
     from agent_env.task_step.task_steps.add_skills import AddSkillsTaskStep
@@ -135,6 +136,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]:
         _get_type(VerifyA2AModalitiesStep): VerifyA2AModalitiesStep,
         _get_type(VerifySandboxTaskStep): VerifySandboxTaskStep,
         _get_type(AggregateVerifiersTaskStep): AggregateVerifiersTaskStep,
+        _get_type(VerifyA2ABuiltinToolsStep): VerifyA2ABuiltinToolsStep,
         _get_type(VerifyA2ARoleStep): VerifyA2ARoleStep,
         _get_type(VerifyA2ASystemPromptStep): VerifyA2ASystemPromptStep,
     }

src/agent_env/task_step/task_steps/a2a_agent_validator/__init__.py

@@ -1,6 +1,7 @@
 from .verify_a2a_agent_card import VerifyA2AAgentCardStep
 from .verify_a2a_agent_config_identity import VerifyA2AAgentConfigIdentityStep
 from .verify_a2a_agent_mcp import VerifyA2AAgentMCPStep
+from .verify_a2a_builtin_tools import VerifyA2ABuiltinToolsStep
 from .verify_a2a_core_protocol import VerifyCoreA2AProtocolStep
 from .verify_a2a_install import VerifyA2AInstallStep
 from .verify_a2a_modalities import VerifyA2AModalitiesStep
@@ -15,6 +16,7 @@
     "VerifyA2AAgentCardStep",
     "VerifyA2AAgentConfigIdentityStep",
     "VerifyA2AAgentMCPStep",
+    "VerifyA2ABuiltinToolsStep",
     "VerifyCoreA2AProtocolStep",
     "VerifyA2AInstallStep",
     "VerifyA2AModalitiesStep",