feat(tool): implement BuiltinAdapter class and MCP automatic resource slot creation (#964)
CI / build (push) Successful in 18s
CI / lint (push) Successful in 3m20s
CI / typecheck (push) Successful in 3m56s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 4m1s
CI / quality (push) Successful in 4m8s
CI / unit_tests (push) Successful in 6m40s
CI / integration_tests (push) Successful in 6m40s
CI / docker (push) Successful in 1m19s
CI / e2e_tests (push) Failing after 17m17s
CI / coverage (push) Failing after 18m20s
CI / benchmark-publish (push) Successful in 31m37s
CI / status-check (push) Failing after 1s

## Summary

Implement `BuiltinAdapter` class and MCP automatic resource slot creation. Two main changes:

### 1. BuiltinAdapter Class (`tool/builtins/adapter.py`)
Formal adapter implementing the tool adapter lifecycle pattern for built-in tools:
- `discover()` — returns all built-in tool descriptors (file, git, subplan tools)
- `register(registry)` — registers all tools in a ToolRegistry, returns registered names
- `activate()`/`deactivate()` — no-ops (built-in tools are always available)
- Wraps existing `ALL_FILE_TOOLS`, `ALL_GIT_TOOLS`, `ALL_SUBPLAN_TOOLS` lists
- Backward compatible — existing registration functions still work

### 2. MCP Resource Slot Inference (`mcp/adapter.py`)
New `infer_resource_slots()` static method on `MCPToolAdapter`:
- Scans MCP tool parameter schemas for file/directory/repo patterns
- Creates `ResourceSlot` objects with appropriate type, access mode, binding mode
- Mapping: `file_path`→file(rw), `directory`→directory(ro), `repo_path`→git-checkout(rw)
- Slots stored in `source_metadata["resource_slots"]` on registered `ToolSpec` objects

### Quality Gates

| Session | Result |
|---|---|
| `nox -s lint` | PASS |
| `nox -s typecheck` | PASS (0 errors) |
| `nox -s unit_tests` | PASS (10,817 scenarios) |
| `nox -s integration_tests` | PASS (6 new tests) |
| `nox -s coverage_report` | 97% (>= 97%) |

Closes #882

Reviewed-on: #964
Co-authored-by: Brent E. Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent E. Edwards <brent.edwards@cleverthis.com>
This commit was merged in pull request #964.
This commit is contained in:
2026-03-21 05:30:01 +00:00
committed by Forgejo
parent b88bc0ec1b
commit 34c6972a05
9 changed files with 779 additions and 4 deletions
+4
View File
@@ -2,6 +2,10 @@
## Unreleased
- Added BuiltinAdapter class and MCP automatic resource slot creation.
BuiltinAdapter wraps register_file_tools/register_git_tools/register_subplan_tool
into a unified adapter interface. McpAdapter.infer_resource_slots() analyzes
tool input schemas to detect file/directory/repository parameters. (#882)
- Added large-project scaling performance tests with ASV benchmarks for
context assembly, execution throughput, and project scaling. Includes
Behave BDD scenarios (78 scenarios, 200 steps), Robot Framework tests,
+21 -2
View File
@@ -6,6 +6,9 @@ overhead across varying checkpoint counts.
from __future__ import annotations
import shutil
import subprocess
import tempfile
from typing import ClassVar
from cleveragents.application.services.checkpoint_service import CheckpointService
@@ -74,11 +77,27 @@ class RollbackSuite:
"""Benchmark rollback operations."""
def setup(self) -> None:
"""Prepare a service with sandbox and checkpoints."""
"""Prepare a service with a real git sandbox and checkpoints."""
self._tmpdir = tempfile.mkdtemp(prefix="asv-rollback-")
subprocess.run(
["git", "init", self._tmpdir],
check=True,
capture_output=True,
)
subprocess.run(
["git", "commit", "--allow-empty", "-m", "initial"],
check=True,
capture_output=True,
cwd=self._tmpdir,
)
self.svc = CheckpointService()
self.svc.register_sandbox(_PLAN_ID, "sandbox-path")
self.svc.register_sandbox(_PLAN_ID, self._tmpdir)
self.cp = self.svc.create_checkpoint(_PLAN_ID, "commit-abc")
def teardown(self) -> None:
"""Remove temporary sandbox directory."""
shutil.rmtree(self._tmpdir, ignore_errors=True)
def time_rollback(self) -> None:
"""Time a single rollback operation."""
self.svc.rollback_to_checkpoint(_PLAN_ID, self.cp.checkpoint_id)
+82
View File
@@ -0,0 +1,82 @@
Feature: BuiltinAdapter and MCP resource slot inference
As a developer
I want a formal BuiltinAdapter class with lifecycle methods
And MCP resource slot inference from tool parameter schemas
So that built-in tools follow the adapter pattern and MCP tools get automatic resource bindings
# ---- BuiltinAdapter ----
Scenario: BuiltinAdapter.discover returns all built-in tools
Given a builtin adapter
Then the builtin adapter should not be discovered yet
When I call discover on the builtin adapter
Then the builtin adapter should be discovered
And the discovered tools should include all file tools
And the discovered tools should include all git tools
And the discovered tools should include all subplan tools
And the total discovered tool count should be 11
Scenario: BuiltinAdapter.register registers all tools in a ToolRegistry
Given a builtin adapter
And a tool registry
When I call register on the builtin adapter
Then the registry should contain 11 tools
And the registry should contain tool "builtin/file-read"
And the registry should contain tool "builtin/git-status"
And the registry should contain tool "builtin/plan-subplan"
Scenario: BuiltinAdapter.activate and deactivate are no-ops
Given a builtin adapter
When I call activate on the builtin adapter
And I call deactivate on the builtin adapter
Then the builtin adapter lifecycle calls should succeed without error
Scenario: Tools registered through BuiltinAdapter have source builtin
Given a builtin adapter
And a tool registry
When I call register on the builtin adapter
Then all registered tools should have source "builtin"
# ---- MCP Resource Slot Inference ----
Scenario: MCP resource slot inference for file_path parameters
Given an MCP tool input schema with a "file_path" property
When I infer resource slots for the schema
Then the inferred slots should contain a "file" slot
And the "file" slot should have resource_type "file"
And the "file" slot should have access "read_write"
And the "file" slot should have binding "parameter"
Scenario: MCP resource slot inference for directory parameters
Given an MCP tool input schema with a "directory" property
When I infer resource slots for the schema
Then the inferred slots should contain a "directory" slot
And the "directory" slot should have resource_type "directory"
And the "directory" slot should have access "read_only"
And the "directory" slot should have binding "parameter"
Scenario: MCP resource slot inference for repo_path parameters
Given an MCP tool input schema with a "repo_path" property
When I infer resource slots for the schema
Then the inferred slots should contain a "repository" slot
And the "repository" slot should have resource_type "git-checkout"
And the "repository" slot should have access "read_write"
And the "repository" slot should have binding "contextual"
# ---- Boundary / negative cases ----
Scenario: MCP resource slot inference returns empty list for empty schema
Given an MCP tool with an empty input schema
When I infer resource slots for the schema
Then the inferred slots should be empty
Scenario: MCP resource slot inference returns empty list for schema with no properties key
Given an MCP tool with an input schema missing the properties key
When I infer resource slots for the schema
Then the inferred slots should be empty
Scenario: MCP resource slot inference deduplicates when multiple file params present
Given an MCP tool with both "file_path" and "filepath" properties
When I infer resource slots for the schema
Then the inferred slots should contain a "file" slot
And the inferred slot count should be 1
+240
View File
@@ -0,0 +1,240 @@
"""Step definitions for features/builtin_adapter.feature."""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from cleveragents.domain.models.core.tool import ResourceSlot
from cleveragents.mcp.adapter import MCPToolAdapter
from cleveragents.tool.builtins import (
ALL_FILE_TOOLS,
ALL_GIT_TOOLS,
ALL_SUBPLAN_TOOLS,
BuiltinAdapter,
)
from cleveragents.tool.runtime import ToolSpec
__all__: list[str] = []
# ---------------------------------------------------------------------------
# Givens
# ---------------------------------------------------------------------------
@given("a builtin adapter")
def step_given_builtin_adapter(context: Any) -> None:
context.builtin_adapter = BuiltinAdapter()
@given('an MCP tool input schema with a "{param_name}" property')
def step_given_mcp_schema(context: Any, param_name: str) -> None:
context.mcp_tool_name = f"test-tool-{param_name}"
context.mcp_input_schema = {
"type": "object",
"properties": {
param_name: {"type": "string", "description": f"The {param_name}"},
},
"required": [param_name],
}
@given("an MCP tool with an empty input schema")
def step_given_empty_schema(context: Any) -> None:
context.mcp_tool_name = "test-tool-empty"
context.mcp_input_schema = {}
@given("an MCP tool with an input schema missing the properties key")
def step_given_no_properties_schema(context: Any) -> None:
context.mcp_tool_name = "test-tool-no-props"
context.mcp_input_schema = {"type": "object"}
@given('an MCP tool with both "file_path" and "filepath" properties')
def step_given_duplicate_file_params(context: Any) -> None:
context.mcp_tool_name = "test-tool-dup-files"
context.mcp_input_schema = {
"type": "object",
"properties": {
"file_path": {"type": "string"},
"filepath": {"type": "string"},
},
}
# ---------------------------------------------------------------------------
# Whens
# ---------------------------------------------------------------------------
@when("I call discover on the builtin adapter")
def step_when_discover(context: Any) -> None:
context.discovered_tools = context.builtin_adapter.discover()
@when("I call register on the builtin adapter")
def step_when_register(context: Any) -> None:
context.registered_names = context.builtin_adapter.register(context.registry)
@when("I call activate on the builtin adapter")
def step_when_activate(context: Any) -> None:
context.activate_exception = None
try:
context.builtin_adapter.activate()
except Exception as exc:
context.activate_exception = exc
@when("I call deactivate on the builtin adapter")
def step_when_deactivate(context: Any) -> None:
context.deactivate_exception = None
try:
context.builtin_adapter.deactivate()
except Exception as exc:
context.deactivate_exception = exc
@when("I infer resource slots for the schema")
def step_when_infer_slots(context: Any) -> None:
context.inferred_slots = MCPToolAdapter.infer_resource_slots(
context.mcp_tool_name,
context.mcp_input_schema,
)
# ---------------------------------------------------------------------------
# Thens
# ---------------------------------------------------------------------------
@then("the builtin adapter should not be discovered yet")
def step_then_not_discovered(context: Any) -> None:
assert not context.builtin_adapter.is_discovered, (
"Expected is_discovered to be False before calling discover()"
)
@then("the builtin adapter should be discovered")
def step_then_is_discovered(context: Any) -> None:
assert context.builtin_adapter.is_discovered, (
"Expected is_discovered to be True after calling discover()"
)
@then("the discovered tools should include all file tools")
def step_then_includes_file_tools(context: Any) -> None:
discovered_names = {t.name for t in context.discovered_tools}
for spec in ALL_FILE_TOOLS:
assert spec.name in discovered_names, (
f"File tool '{spec.name}' not in discovered tools"
)
@then("the discovered tools should include all git tools")
def step_then_includes_git_tools(context: Any) -> None:
discovered_names = {t.name for t in context.discovered_tools}
for spec in ALL_GIT_TOOLS:
assert spec.name in discovered_names, (
f"Git tool '{spec.name}' not in discovered tools"
)
@then("the discovered tools should include all subplan tools")
def step_then_includes_subplan_tools(context: Any) -> None:
discovered_names = {t.name for t in context.discovered_tools}
for spec in ALL_SUBPLAN_TOOLS:
assert spec.name in discovered_names, (
f"Subplan tool '{spec.name}' not in discovered tools"
)
@then("the total discovered tool count should be {count:d}")
def step_then_discovered_count(context: Any, count: int) -> None:
actual = len(context.discovered_tools)
assert actual == count, f"Expected {count} discovered tools, got {actual}"
@then("the builtin adapter lifecycle calls should succeed without error")
def step_then_lifecycle_no_exception(context: Any) -> None:
assert context.activate_exception is None, (
f"activate raised: {context.activate_exception}"
)
assert context.deactivate_exception is None, (
f"deactivate raised: {context.deactivate_exception}"
)
@then('all registered tools should have source "{source}"')
def step_then_all_source(context: Any, source: str) -> None:
tools: list[ToolSpec] = context.registry.list_tools()
for spec in tools:
assert spec.source == source, (
f"Tool '{spec.name}' has source='{spec.source}', expected '{source}'"
)
@then("the inferred slots should be empty")
def step_then_slots_empty(context: Any) -> None:
assert len(context.inferred_slots) == 0, (
f"Expected no inferred slots, got {len(context.inferred_slots)}: "
f"{[s.name for s in context.inferred_slots]}"
)
@then("the inferred slot count should be {count:d}")
def step_then_slot_count(context: Any, count: int) -> None:
actual = len(context.inferred_slots)
assert actual == count, (
f"Expected {count} inferred slot(s), got {actual}: "
f"{[s.name for s in context.inferred_slots]}"
)
@then('the inferred slots should contain a "{slot_name}" slot')
def step_then_slot_exists(context: Any, slot_name: str) -> None:
names = [s.name for s in context.inferred_slots]
assert slot_name in names, (
f"Slot '{slot_name}' not found in inferred slots: {names}"
)
@then('the "{slot_name}" slot should have resource_type "{expected}"')
def step_then_slot_resource_type(context: Any, slot_name: str, expected: str) -> None:
slot = _find_slot(context.inferred_slots, slot_name)
assert slot.resource_type == expected, (
f"Slot '{slot_name}' resource_type='{slot.resource_type}', "
f"expected '{expected}'"
)
@then('the "{slot_name}" slot should have access "{expected}"')
def step_then_slot_access(context: Any, slot_name: str, expected: str) -> None:
slot = _find_slot(context.inferred_slots, slot_name)
assert slot.access.value == expected, (
f"Slot '{slot_name}' access='{slot.access.value}', expected '{expected}'"
)
@then('the "{slot_name}" slot should have binding "{expected}"')
def step_then_slot_binding(context: Any, slot_name: str, expected: str) -> None:
slot = _find_slot(context.inferred_slots, slot_name)
assert slot.binding.value == expected, (
f"Slot '{slot_name}' binding='{slot.binding.value}', expected '{expected}'"
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _find_slot(slots: list[ResourceSlot], name: str) -> ResourceSlot:
"""Find a ResourceSlot by name or raise AssertionError."""
for slot in slots:
if slot.name == name:
return slot
raise AssertionError(f"Slot '{name}' not found in {[s.name for s in slots]}")
+51
View File
@@ -0,0 +1,51 @@
*** Settings ***
Documentation Integration tests for BuiltinAdapter and MCP resource slot inference
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_builtin_adapter.py
*** Test Cases ***
BuiltinAdapter Discover Returns All Tools
[Documentation] BuiltinAdapter.discover() returns all built-in tool specs
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} discover
... cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} discover-ok
BuiltinAdapter Register Populates Registry
[Documentation] BuiltinAdapter.register() populates a ToolRegistry with all tools
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} register
... cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} register-ok
BuiltinAdapter Tools Have Builtin Source
[Documentation] All tools registered via BuiltinAdapter have source=builtin
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} source
... cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} source-ok
MCP Resource Slot Inference For File Path
[Documentation] MCP infer_resource_slots creates file slot from file_path param
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} mcp_file
... cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} mcp-file-ok
MCP Resource Slot Inference For Directory
[Documentation] MCP infer_resource_slots creates directory slot from directory param
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} mcp_dir
... cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} mcp-dir-ok
MCP Resource Slot Inference For Repo Path
[Documentation] MCP infer_resource_slots creates repository slot from repo_path param
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} mcp_repo
... cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} mcp-repo-ok
+122
View File
@@ -0,0 +1,122 @@
"""Helper script for Robot Framework BuiltinAdapter integration tests."""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.mcp.adapter import MCPToolAdapter
from cleveragents.tool.builtins import BuiltinAdapter
from cleveragents.tool.registry import ToolRegistry
def test_discover() -> None:
"""Verify BuiltinAdapter.discover() returns all built-in tools."""
adapter = BuiltinAdapter()
tools = adapter.discover()
assert len(tools) == 11, f"Expected 11 tools, got {len(tools)}"
names = {t.name for t in tools}
assert "builtin/file-read" in names
assert "builtin/git-status" in names
assert "builtin/plan-subplan" in names
# Calling discover again returns same result
tools2 = adapter.discover()
assert len(tools2) == 11
print("discover-ok")
def test_register() -> None:
"""Verify BuiltinAdapter.register() populates a ToolRegistry."""
adapter = BuiltinAdapter()
registry = ToolRegistry()
registered = adapter.register(registry)
assert len(registered) == 11, f"Expected 11, got {len(registered)}"
assert registry.get("builtin/file-read") is not None
assert registry.get("builtin/git-status") is not None
assert registry.get("builtin/plan-subplan") is not None
print("register-ok")
def test_source() -> None:
"""Verify all tools registered via BuiltinAdapter have source=builtin."""
adapter = BuiltinAdapter()
registry = ToolRegistry()
adapter.register(registry)
for spec in registry.list_tools():
assert spec.source == "builtin", (
f"Tool '{spec.name}' has source='{spec.source}'"
)
print("source-ok")
def test_mcp_file() -> None:
"""Verify MCP resource slot inference for file_path parameter."""
schema: dict[str, Any] = {
"type": "object",
"properties": {
"file_path": {"type": "string"},
},
}
slots = MCPToolAdapter.infer_resource_slots("test-read-file", schema)
assert len(slots) == 1, f"Expected 1 slot, got {len(slots)}"
assert slots[0].name == "file"
assert slots[0].resource_type == "file"
assert slots[0].access.value == "read_write"
assert slots[0].binding.value == "parameter"
print("mcp-file-ok")
def test_mcp_dir() -> None:
"""Verify MCP resource slot inference for directory parameter."""
schema: dict[str, Any] = {
"type": "object",
"properties": {
"directory": {"type": "string"},
},
}
slots = MCPToolAdapter.infer_resource_slots("test-list-dir", schema)
assert len(slots) == 1, f"Expected 1 slot, got {len(slots)}"
assert slots[0].name == "directory"
assert slots[0].resource_type == "directory"
assert slots[0].access.value == "read_only"
assert slots[0].binding.value == "parameter"
print("mcp-dir-ok")
def test_mcp_repo() -> None:
"""Verify MCP resource slot inference for repo_path parameter."""
schema: dict[str, Any] = {
"type": "object",
"properties": {
"repo_path": {"type": "string"},
},
}
slots = MCPToolAdapter.infer_resource_slots("test-git-status", schema)
assert len(slots) == 1, f"Expected 1 slot, got {len(slots)}"
assert slots[0].name == "repository"
assert slots[0].resource_type == "git-checkout"
assert slots[0].access.value == "read_write"
assert slots[0].binding.value == "contextual"
print("mcp-repo-ok")
if __name__ == "__main__":
cmd = sys.argv[1] if len(sys.argv) > 1 else "discover"
dispatch: dict[str, Any] = {
"discover": test_discover,
"register": test_register,
"source": test_source,
"mcp_file": test_mcp_file,
"mcp_dir": test_mcp_dir,
"mcp_repo": test_mcp_repo,
}
fn = dispatch.get(cmd)
if fn:
fn()
else:
print(f"Unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
+133 -2
View File
@@ -27,11 +27,25 @@ if TYPE_CHECKING:
import jsonschema
from pydantic import BaseModel, ConfigDict, Field
from cleveragents.domain.models.core.tool import ToolCapability
from cleveragents.domain.models.core.tool import (
BindingMode,
ResourceAccessMode,
ResourceSlot,
ToolCapability,
)
from cleveragents.tool.runtime import ToolSpec
logger = logging.getLogger(__name__)
__all__ = [
"MCPServerConfig",
"MCPToolAdapter",
"MCPToolDescriptor",
"MCPToolFilter",
"MCPToolResult",
"MCPTransport",
]
class MCPServerConfig(BaseModel):
"""Configuration for connecting to an MCP server.
@@ -512,6 +526,23 @@ class MCPToolAdapter:
checkpointable=False,
)
inferred_slots = self.infer_resource_slots(desc.name, desc.input_schema)
slot_dicts: list[dict[str, str]] = [
{
"name": slot.name,
"resource_type": slot.resource_type,
"access": slot.access.value,
"binding": slot.binding.value,
}
for slot in inferred_slots
]
# TODO(#882): resource_slots are stored in source_metadata but
# nothing downstream reads them yet. The ToolRegistry persists
# resource bindings via domain Tool.resource_slots, and the DB
# migration (c1_001) stores them in tool_resource_bindings.
# A follow-up ticket should wire inferred slots into the domain
# Tool objects so the registry and DB actually consume them.
spec = ToolSpec(
name=tool_name,
description=desc.description or f"MCP tool: {desc.name}",
@@ -519,13 +550,113 @@ class MCPToolAdapter:
capabilities=capabilities,
handler=_make_handler(desc.name, adapter_ref),
source="mcp",
source_metadata={"server": self._config.name},
source_metadata={
"server": self._config.name,
"resource_slots": slot_dicts,
},
)
registry.register(spec)
registered.append(tool_name)
return registered
# -- Parameter name patterns for resource slot inference ----------------
#: Parameter names that indicate a file resource.
_FILE_PARAMS: frozenset[str] = frozenset({"file_path", "filepath", "file"})
#: Parameter names that indicate a directory resource.
_DIR_PARAMS: frozenset[str] = frozenset({"directory", "dir", "folder"})
#: Parameter names that indicate a git repository resource.
_REPO_PARAMS: frozenset[str] = frozenset({"repo_path", "repository"})
@staticmethod
def infer_resource_slots(
tool_name: str,
input_schema: dict[str, Any],
) -> list[ResourceSlot]:
"""Infer resource slots from MCP tool parameter schemas.
Scans ``input_schema["properties"]`` for parameter names matching
file, directory, and repository patterns and creates
:class:`~cleveragents.domain.models.core.tool.ResourceSlot` objects
with appropriate metadata.
Parameter -> ResourceSlot mapping:
- ``file_path``, ``filepath``, ``file`` -> file / read_write / parameter
- ``directory``, ``dir``, ``folder`` -> directory / read_only / parameter
- ``repo_path``, ``repository`` -> repository / read_write / contextual
Parameters
----------
tool_name:
Name of the MCP tool (used for logging).
input_schema:
JSON Schema dict for the tool's inputs.
Returns
-------
list[ResourceSlot]
Inferred resource slots (may be empty).
"""
# Mapping from parameter-name sets to slot metadata.
_SLOT_MAP: list[
tuple[frozenset[str], str, str, ResourceAccessMode, BindingMode]
] = [
(
MCPToolAdapter._FILE_PARAMS,
"file",
"file",
ResourceAccessMode.READ_WRITE,
BindingMode.PARAMETER,
),
(
MCPToolAdapter._DIR_PARAMS,
"directory",
"directory",
ResourceAccessMode.READ_ONLY,
BindingMode.PARAMETER,
),
(
MCPToolAdapter._REPO_PARAMS,
"repository",
"git-checkout",
ResourceAccessMode.READ_WRITE,
BindingMode.CONTEXTUAL,
),
]
properties: dict[str, Any] = input_schema.get("properties", {})
slots: list[ResourceSlot] = []
seen_names: set[str] = set()
for param_name in properties:
for param_set, slot_name, resource_type, access, binding in _SLOT_MAP:
if param_name not in param_set:
continue
if slot_name in seen_names:
break
seen_names.add(slot_name)
slots.append(
ResourceSlot(
name=slot_name,
resource_type=resource_type,
access=access,
binding=binding,
required=True,
description=(
f"{slot_name.title()} resource inferred from "
f"'{param_name}' parameter of MCP tool "
f"'{tool_name}'"
),
)
)
break
return slots
@staticmethod
def infer_capabilities(tool_name: str) -> dict[str, bool]:
"""Infer ToolCapability metadata from an MCP tool name.
@@ -3,10 +3,15 @@
Provides convenience functions to register all built-in file, git, and
subplan tools into a ``ToolRegistry``, optionally wrapped with changeset
capture.
Also exports :class:`BuiltinAdapter`, a formal adapter implementing the
discover / register / activate / deactivate lifecycle pattern for all
built-in tools.
"""
from __future__ import annotations
from cleveragents.tool.builtins.adapter import BuiltinAdapter
from cleveragents.tool.builtins.changeset import (
ChangeSet,
ChangeSetCapture,
@@ -86,6 +91,7 @@ __all__ = [
"GIT_LOG_SPEC",
"GIT_STATUS_SPEC",
"PLAN_SUBPLAN_SPEC",
"BuiltinAdapter",
"ChangeSet",
"ChangeSetCapture",
"ChangeSetEntry",
+120
View File
@@ -0,0 +1,120 @@
"""Formal adapter for built-in tools following the adapter lifecycle pattern.
Provides the :class:`BuiltinAdapter` class that wraps the existing
``register_file_tools``, ``register_git_tools``, and
``register_subplan_tool`` functions behind a unified adapter interface
with ``discover``, ``register``, ``activate``, and ``deactivate``
lifecycle methods.
The adapter follows the same lifecycle naming used by
:class:`~cleveragents.skill.loader.AgentSkillLoader` (``discover``,
``register``, ``activate``, ``deactivate``) rather than the connection-
oriented naming of :class:`~cleveragents.mcp.adapter.MCPToolAdapter`
(``connect``, ``discover_tools``, ``register_tools``, ``disconnect``).
Built-in tools are always in the default namespace, so ``register``
does not accept a ``namespace`` parameter.
Usage::
from cleveragents.tool.builtins.adapter import BuiltinAdapter
from cleveragents.tool.registry import ToolRegistry
adapter = BuiltinAdapter()
specs = adapter.discover() # list of all built-in ToolSpecs
registry = ToolRegistry()
names = adapter.register(registry) # registers and returns names
adapter.activate() # no-op — built-ins are always available
adapter.deactivate() # no-op
"""
from __future__ import annotations
from cleveragents.tool.builtins.file_tools import ALL_FILE_TOOLS
from cleveragents.tool.builtins.git_tools import ALL_GIT_TOOLS
from cleveragents.tool.builtins.subplan_tool import ALL_SUBPLAN_TOOLS
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runtime import ToolSpec
__all__ = ["BuiltinAdapter"]
class BuiltinAdapter:
"""Formal adapter for built-in tools following the adapter lifecycle.
Wraps the existing ``ALL_FILE_TOOLS``, ``ALL_GIT_TOOLS``, and
``ALL_SUBPLAN_TOOLS`` lists behind a unified interface.
The lifecycle methods mirror those of
:class:`~cleveragents.skill.loader.AgentSkillLoader`:
- :meth:`discover` enumerate all available built-in tool specs.
- :meth:`register` register all specs into a ``ToolRegistry``.
- :meth:`activate` no-op (built-in tools are always available).
- :meth:`deactivate` no-op.
"""
def __init__(self) -> None:
self._tools: list[ToolSpec] = []
self._discovered: bool = False
@property
def is_discovered(self) -> bool:
"""Whether :meth:`discover` has been called."""
return self._discovered
def discover(self) -> list[ToolSpec]:
"""Discover all built-in tools.
Returns a copy of the internal list so callers cannot mutate it.
The discovery result is cached; subsequent calls return the same
specs without re-scanning.
Returns
-------
list[ToolSpec]
All built-in file, git, and subplan tool specs.
"""
if not self._discovered:
self._tools = [*ALL_FILE_TOOLS, *ALL_GIT_TOOLS, *ALL_SUBPLAN_TOOLS]
self._discovered = True
return list(self._tools)
def register(self, registry: ToolRegistry) -> list[str]:
"""Register all built-in tools in the registry.
Calls :meth:`discover` if not already done, then registers each
spec into *registry*. Already-registered tools are silently
skipped so that calling ``register()`` twice on the same registry
is safe (idempotent).
Parameters
----------
registry:
The ``ToolRegistry`` to register tools into.
Returns
-------
list[str]
Names of successfully registered tools.
Raises
------
ValueError
If *registry* is ``None``.
"""
if registry is None:
raise ValueError("registry must not be None")
tools = self.discover()
registered: list[str] = []
for spec in tools:
if registry.get(spec.name) is not None:
continue
registry.register(spec)
registered.append(spec.name)
return registered
def activate(self) -> None:
"""No-op for built-in tools (always available)."""
def deactivate(self) -> None:
"""No-op for built-in tools."""