cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(mcp): implement lazy start/auto-stop and sandbox path rewriting for MCP tools #1200

Merged
freemo merged 1 commits from feature/mcp-lazy-lifecycle-sandbox into master 2026-04-02 18:18:40 +00:00
Conversation 10 Commits 1 Files Changed 9
+1612
9 changed files with 1612 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/mcp_lifecycle.feature
+160
View File
@@ -0,0 +1,160 @@
Feature: MCP Server Lifecycle Management
As an actor runtime
I want MCP servers to start lazily and stop when idle
So that resources are used efficiently and server management is automatic
# -------------------------------------------------------------------
# Lazy Start
# -------------------------------------------------------------------
Scenario: MCP server starts on first tool call (lazy start)
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
Then the MCP client state should be "idle"
And the MCP client should not be connected
When I call MCP tool "list_repos" with arguments {}
Then the MCP client should be connected
And the MCP client state should be "running"
Scenario: MCP server does not start without tool call when lazy start enabled
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
Then the MCP client should not be connected
Scenario: Explicit start works with lazy start enabled
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I explicitly start the MCP client
Then the MCP client should be connected
And the MCP client state should be "running"
Scenario: Tool call fails when lazy start disabled and not started
Given an MCP client config for "github" with lazy start disabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" expecting a runtime error
Then the MCP client error should mention "not started"
Scenario: Explicit start is idempotent
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I explicitly start the MCP client
And I explicitly start the MCP client
Then the MCP client should be connected
# -------------------------------------------------------------------
# Auto-Stop
# -------------------------------------------------------------------
Scenario: MCP server auto-stops after idle timeout
Given an MCP client config for "github" with idle timeout 0.15 seconds
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
And I wait 0.5 seconds
Then the MCP client state should be "stopped"
And the MCP client should not be connected
Scenario: Activity resets idle timer
Given an MCP client config for "github" with idle timeout 0.5 seconds
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
And I wait 0.25 seconds
And I call MCP tool "list_repos" with arguments {}
And I wait 0.25 seconds
Then the MCP client should be connected
Scenario: Auto-stop disabled when timeout is zero
Given an MCP client config for "github" with idle timeout 0.0 seconds
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
And I wait 0.15 seconds
Then the MCP client should be connected
# -------------------------------------------------------------------
# Shutdown
# -------------------------------------------------------------------
Scenario: Explicit shutdown stops the server
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
And I shutdown the MCP client
Then the MCP client state should be "stopped"
And the MCP client should not be connected
Scenario: Double shutdown is safe
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
And I shutdown the MCP client
And I shutdown the MCP client
Then the MCP client state should be "stopped"
# -------------------------------------------------------------------
# Health Monitoring
# -------------------------------------------------------------------
Scenario: Health check detects crash and auto-restarts
Given an MCP client config for "github" with health check interval 0.15 seconds
And a mock MCP transport with tool "list_repos" that crashes after 1 call
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
And I wait 1.0 seconds
Then the MCP client restart count should be greater than 0
Scenario: Health check disabled when interval is zero
Given an MCP client config for "github" with health check interval 0.0 seconds
And a mock MCP transport with tool "list_repos"
When I create an MCP client
And I call MCP tool "list_repos" with arguments {}
Then the MCP client health check failures should be 0
# -------------------------------------------------------------------
# Registry (Multiple Servers)
# -------------------------------------------------------------------
Scenario: Multiple MCP servers with independent lifecycles
Given an MCP registry
And an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I register the MCP client in namespace "gh"
And I register another MCP client config for "jira" in namespace "jira"
Then the MCP registry should have 2 namespaces
And MCP namespace "gh" should not be connected
And MCP namespace "jira" should not be connected
Scenario: Registry call_tool starts specific server lazily
Given an MCP registry
And an MCP client config for "github" with lazy start enabled
And a mock MCP transport with tool "list_repos"
When I register the MCP client in namespace "gh"
And I call tool "list_repos" on MCP namespace "gh" with arguments {}
Then MCP namespace "gh" should be connected
Scenario: Registry shutdown_all stops all servers
Given an MCP registry with 2 running servers
When I shutdown all MCP servers
Then the MCP registry should have 2 namespaces
And all MCP servers should be stopped
# -------------------------------------------------------------------
# Capability Metadata
# -------------------------------------------------------------------
Scenario: Full capability metadata exposed after connect
Given an MCP client config for "github" with lazy start enabled
And a mock MCP transport with full capabilities
When I create an MCP client
And I explicitly start the MCP client
Then the MCP adapter capability metadata should have tools true
And the MCP adapter capability metadata should have resources true
And the MCP adapter capability metadata should have prompts true
features/mcp_sandbox_rewrite.feature
+66
View File
@@ -0,0 +1,66 @@
Feature: MCP Sandbox Path Rewriting
As an actor runtime executing in a sandboxed workspace
I want file paths in MCP tool arguments and responses rewritten
So that MCP tools operate on correct paths inside the sandbox
# -------------------------------------------------------------------
# Argument Rewriting (Host -> Sandbox)
# -------------------------------------------------------------------
Scenario: Rewrite host path to sandbox path in arguments
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool arguments with key "file_path" and value "/home/user/project/src/main.py"
Then the rewritten arguments should have "file_path" equal to "/workspace/src/main.py"
Scenario: Non-matching paths pass through unchanged
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool arguments with key "file_path" and value "/other/location/file.py"
Then the rewritten arguments should have "file_path" equal to "/other/location/file.py"
Scenario: Non-path string values are not rewritten
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool arguments with key "title" and value "My Document"
Then the rewritten arguments should have "title" equal to "My Document"
Scenario: Nested dict paths are rewritten
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite nested tool arguments with outer key "config" inner key "source" and value "/home/user/project/cfg.yaml"
Then the rewritten nested argument "config.source" should equal "/workspace/cfg.yaml"
Scenario: List of paths are rewritten
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool arguments with key "files" and path list "/home/user/project/a.py,/home/user/project/b.py"
Then the rewritten argument "files" should be a list with 2 rewritten paths
Scenario: Host root path itself maps to sandbox root
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool arguments with key "dir" and value "/home/user/project"
Then the rewritten arguments should have "dir" equal to "/workspace"
# -------------------------------------------------------------------
# Response Rewriting (Sandbox -> Host)
# -------------------------------------------------------------------
Scenario: Rewrite sandbox path to host path in response
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool response with key "output_file" and value "/workspace/build/out.txt"
Then the rewritten response should have "output_file" equal to "/home/user/project/build/out.txt"
Scenario: Non-matching sandbox response paths pass through
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite tool response with key "output_file" and value "/tmp/other.txt"
Then the rewritten response should have "output_file" equal to "/tmp/other.txt"
Scenario: Nested response paths are rewritten
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I rewrite nested tool response with outer key "result" inner key "path" and value "/workspace/dist/app.js"
Then the rewritten nested response "result.path" should equal "/home/user/project/dist/app.js"
# -------------------------------------------------------------------
# Roundtrip
# -------------------------------------------------------------------
Scenario: Roundtrip rewrite preserves original path
Given a sandbox path rewriter with host root "/home/user/project" and sandbox root "/workspace"
When I roundtrip rewrite tool arguments with key "file" and value "/home/user/project/src/main.py"
Then the roundtrip path should equal "/home/user/project/src/main.py"
features/steps/mcp_lifecycle_steps.py
+342
View File
@@ -0,0 +1,342 @@
"""Step definitions for features/mcp_lifecycle.feature."""
from __future__ import annotations
import json
import time
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.mcp.adapter import MCPServerConfig
from cleveragents.mcp.client import McpClient, McpClientConfig
from cleveragents.mcp.registry import McpRegistry
from features.mocks.mock_mcp_transport import MockMCPTransport
def _mock_tool(name: str, desc: str = "", schema: dict | None = None) -> dict:
return {
"name": name,
"description": desc or f"Mock tool {name}",
"inputSchema": schema or {},
}
# ---------------------------------------------------------------
# Given Steps
# ---------------------------------------------------------------
@given('an MCP client config for "{name}" with lazy start enabled')
def step_mcp_client_config_lazy(context: Context, name: str) -> None:
context.mcp_server_config = MCPServerConfig(
name=name, transport="stdio", command="echo"
)
context.mcp_client_config = McpClientConfig(
server=context.mcp_server_config,
lazy_start=True,
idle_timeout_seconds=0,
health_check_interval_seconds=0,
)
@given('an MCP client config for "{name}" with lazy start disabled')
def step_mcp_client_config_no_lazy(context: Context, name: str) -> None:
context.mcp_server_config = MCPServerConfig(
name=name, transport="stdio", command="echo"
)
context.mcp_client_config = McpClientConfig(
server=context.mcp_server_config,
lazy_start=False,
idle_timeout_seconds=0,
health_check_interval_seconds=0,
)
@given('an MCP client config for "{name}" with idle timeout {secs:f} seconds')
def step_mcp_client_config_idle(context: Context, name: str, secs: float) -> None:
context.mcp_server_config = MCPServerConfig(
name=name, transport="stdio", command="echo"
)
context.mcp_client_config = McpClientConfig(
server=context.mcp_server_config,
lazy_start=True,
idle_timeout_seconds=secs,
health_check_interval_seconds=0,
)
@given('an MCP client config for "{name}" with health check interval {secs:f} seconds')
def step_mcp_client_config_health(context: Context, name: str, secs: float) -> None:
context.mcp_server_config = MCPServerConfig(
name=name, transport="stdio", command="echo"
)
context.mcp_client_config = McpClientConfig(
server=context.mcp_server_config,
lazy_start=True,
idle_timeout_seconds=0,
health_check_interval_seconds=secs,
auto_restart=True,
)
@given('a mock MCP transport with tool "{tool_name}"')
def step_mock_transport_with_tool(context: Context, tool_name: str) -> None:
tools = [_mock_tool(tool_name)]
results = {tool_name: {"result": "ok"}}
context.mcp_lifecycle_transport = MockMCPTransport(
tools=tools, invoke_results=results
)
@given('a mock MCP transport with tool "{tool_name}" that crashes after {count:d} call')
def step_mock_transport_crash(context: Context, tool_name: str, count: int) -> None:
"""Transport that works for N calls then starts failing health checks."""
tools = [_mock_tool(tool_name)]
results = {tool_name: {"result": "ok"}}
call_counter: list[int] = [0]
crash_active: list[bool] = [False]
class _CrashAfterNTransport(MockMCPTransport):
def connect(self, config: MCPServerConfig) -> dict[str, Any]:
# Reset crash state and counter on reconnect so the
# server can recover after a restart.
crash_active[0] = False
call_counter[0] = 0
return super().connect(config)
def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
if method == "tools/list":
call_counter[0] += 1
if call_counter[0] > count:
crash_active[0] = True
if crash_active[0]:
msg = "Server crashed"
raise ConnectionError(msg)
return super().call(method, params)
context.mcp_lifecycle_transport = _CrashAfterNTransport(
tools=tools, invoke_results=results
)
@given("a mock MCP transport with full capabilities")
def step_mock_transport_full_caps(context: Context) -> None:
tools = [_mock_tool("test_tool")]
class _FullCapsTransport(MockMCPTransport):
def connect(self, config: MCPServerConfig) -> dict[str, Any]:
self._connected = True
return {
"capabilities": {
"tools": True,
"resources": True,
"prompts": True,
}
}
context.mcp_lifecycle_transport = _FullCapsTransport(tools=tools)
@given("an MCP registry")
def step_mcp_registry(context: Context) -> None:
context.mcp_lifecycle_registry = McpRegistry()
@given("an MCP registry with {count:d} running servers")
def step_mcp_registry_running(context: Context, count: int) -> None:
registry = McpRegistry()
for i in range(count):
server_name = f"server-{i}"
tools = [_mock_tool(f"tool_{i}")]
transport = MockMCPTransport(tools=tools)
config = McpClientConfig(
server=MCPServerConfig(name=server_name, transport="stdio", command="echo"),
lazy_start=True,
idle_timeout_seconds=0,
health_check_interval_seconds=0,
)
client = registry.register(config, namespace=f"ns-{i}", transport=transport)
client.start()
context.mcp_lifecycle_registry = registry
# ---------------------------------------------------------------
# When Steps
# ---------------------------------------------------------------
@when("I create an MCP client")
def step_create_mcp_client(context: Context) -> None:
context.mcp_lifecycle_client = McpClient(
config=context.mcp_client_config,
transport=context.mcp_lifecycle_transport,
)
context.mcp_lifecycle_error = None
@when('I call MCP tool "{tool_name}" with arguments {args_json}')
def step_call_mcp_tool(context: Context, tool_name: str, args_json: str) -> None:
arguments = json.loads(args_json)
context.mcp_lifecycle_result = context.mcp_lifecycle_client.call_tool(
tool_name, arguments
)
@when('I call MCP tool "{tool_name}" expecting a runtime error')
def step_call_mcp_tool_error(context: Context, tool_name: str) -> None:
context.mcp_lifecycle_error = None
try:
context.mcp_lifecycle_client.call_tool(tool_name, {})
except Exception as exc:
context.mcp_lifecycle_error = str(exc)
@when("I explicitly start the MCP client")
def step_start_mcp_client(context: Context) -> None:
context.mcp_lifecycle_client.start()
@when("I shutdown the MCP client")
def step_shutdown_mcp_client(context: Context) -> None:
context.mcp_lifecycle_client.shutdown()
@when("I wait {secs:f} seconds")
def step_wait_seconds(context: Context, secs: float) -> None:
# Use the original sleep if the test environment has patched
# time.sleep (e.g. _install_fast_sleep_patch caps at 10 ms).
# Timer-based lifecycle tests need real wall-clock advancement.
original_sleep = getattr(time, "_original_sleep", time.sleep)
original_sleep(secs)
@when('I register the MCP client in namespace "{namespace}"')
def step_register_mcp_client(context: Context, namespace: str) -> None:
context.mcp_lifecycle_registry.register(
config=context.mcp_client_config,
namespace=namespace,
transport=context.mcp_lifecycle_transport,
)
@when('I register another MCP client config for "{name}" in namespace "{namespace}"')
def step_register_another_mcp_client(
context: Context, name: str, namespace: str
) -> None:
config = McpClientConfig(
server=MCPServerConfig(name=name, transport="stdio", command="echo"),
lazy_start=True,
idle_timeout_seconds=0,
health_check_interval_seconds=0,
)
tools = [_mock_tool("default_tool")]
transport = MockMCPTransport(tools=tools)
context.mcp_lifecycle_registry.register(
config=config, namespace=namespace, transport=transport
)
@when(
'I call tool "{tool_name}" on MCP namespace "{namespace}" with arguments {args_json}'
)
def step_registry_call_tool(
context: Context, tool_name: str, namespace: str, args_json: str
) -> None:
arguments = json.loads(args_json)
context.mcp_lifecycle_result = context.mcp_lifecycle_registry.call_tool(
namespace, tool_name, arguments
)
@when("I shutdown all MCP servers")
def step_shutdown_all(context: Context) -> None:
context.mcp_lifecycle_registry.shutdown_all()
# ---------------------------------------------------------------
# Then Steps
# ---------------------------------------------------------------
@then('the MCP client state should be "{state}"')
def step_mcp_client_state(context: Context, state: str) -> None:
actual = context.mcp_lifecycle_client.state
assert actual == state, f"Expected state '{state}', got '{actual}'"
@then("the MCP client should not be connected")
def step_mcp_client_not_connected(context: Context) -> None:
assert not context.mcp_lifecycle_client.is_connected
@then("the MCP client should be connected")
def step_mcp_client_connected(context: Context) -> None:
assert context.mcp_lifecycle_client.is_connected
@then('the MCP client error should mention "{text}"')
def step_mcp_client_error_mention(context: Context, text: str) -> None:
assert context.mcp_lifecycle_error is not None, "Expected error but none occurred"
assert text.lower() in context.mcp_lifecycle_error.lower(), (
f"Error '{context.mcp_lifecycle_error}' does not mention '{text}'"
)
@then("the MCP client restart count should be greater than {count:d}")
def step_mcp_restart_count_gt(context: Context, count: int) -> None:
actual = context.mcp_lifecycle_client.restart_count
assert actual > count, f"Expected restart count > {count}, got {actual}"
@then("the MCP client health check failures should be {count:d}")
def step_mcp_health_failures(context: Context, count: int) -> None:
actual = context.mcp_lifecycle_client.health_check_failures
assert actual == count, f"Expected {count} health failures, got {actual}"
@then("the MCP registry should have {count:d} namespaces")
def step_registry_namespace_count(context: Context, count: int) -> None:
actual = len(context.mcp_lifecycle_registry.list_namespaces())
assert actual == count, f"Expected {count} namespaces, got {actual}"
@then('MCP namespace "{namespace}" should not be connected')
def step_namespace_not_connected(context: Context, namespace: str) -> None:
client = context.mcp_lifecycle_registry.get(namespace)
assert client is not None, f"Namespace '{namespace}' not found"
assert not client.is_connected
@then('MCP namespace "{namespace}" should be connected')
def step_namespace_connected(context: Context, namespace: str) -> None:
client = context.mcp_lifecycle_registry.get(namespace)
assert client is not None, f"Namespace '{namespace}' not found"
assert client.is_connected
@then("all MCP servers should be stopped")
def step_all_stopped(context: Context) -> None:
for ns in context.mcp_lifecycle_registry.list_namespaces():
client = context.mcp_lifecycle_registry.get(ns)
assert client is not None
assert not client.is_connected, f"Server in namespace '{ns}' still connected"
@then("the MCP adapter capability metadata should have tools true")
def step_cap_tools_true(context: Context) -> None:
meta = context.mcp_lifecycle_client.adapter.capability_metadata
assert meta.tools is True, f"Expected tools=True, got {meta.tools}"
@then("the MCP adapter capability metadata should have resources true")
def step_cap_resources_true(context: Context) -> None:
meta = context.mcp_lifecycle_client.adapter.capability_metadata
assert meta.resources is True, f"Expected resources=True, got {meta.resources}"
@then("the MCP adapter capability metadata should have prompts true")
def step_cap_prompts_true(context: Context) -> None:
meta = context.mcp_lifecycle_client.adapter.capability_metadata
assert meta.prompts is True, f"Expected prompts=True, got {meta.prompts}"
features/steps/mcp_sandbox_rewrite_steps.py
+138
View File
@@ -0,0 +1,138 @@
"""Step definitions for features/mcp_sandbox_rewrite.feature."""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.mcp.sandbox import SandboxPathRewriter
# ---------------------------------------------------------------
# Given Steps
# ---------------------------------------------------------------
@given(
'a sandbox path rewriter with host root "{host_root}" and sandbox root "{sandbox_root}"'
)
def step_sandbox_rewriter(context: Context, host_root: str, sandbox_root: str) -> None:
context.sandbox_rewriter = SandboxPathRewriter(
host_root=host_root, sandbox_root=sandbox_root
)
# ---------------------------------------------------------------
# When Steps
# ---------------------------------------------------------------
@when('I rewrite tool arguments with key "{key}" and value "{value}"')
def step_rewrite_arguments_kv(context: Context, key: str, value: str) -> None:
arguments = {key: value}
context.sandbox_rewritten_args = context.sandbox_rewriter.rewrite_arguments(
arguments
)
@when(
'I rewrite nested tool arguments with outer key "{outer}" inner key "{inner}" and value "{value}"'
)
def step_rewrite_nested_arguments(
context: Context, outer: str, inner: str, value: str
) -> None:
arguments = {outer: {inner: value}}
context.sandbox_rewritten_args = context.sandbox_rewriter.rewrite_arguments(
arguments
)
@when('I rewrite tool arguments with key "{key}" and path list "{paths}"')
def step_rewrite_arguments_list(context: Context, key: str, paths: str) -> None:
path_list = paths.split(",")
arguments = {key: path_list}
context.sandbox_rewritten_args = context.sandbox_rewriter.rewrite_arguments(
arguments
)
@when('I rewrite tool response with key "{key}" and value "{value}"')
def step_rewrite_response_kv(context: Context, key: str, value: str) -> None:
response = {key: value}
context.sandbox_rewritten_resp = context.sandbox_rewriter.rewrite_response(response)
@when(
'I rewrite nested tool response with outer key "{outer}" inner key "{inner}" and value "{value}"'
)
def step_rewrite_nested_response(
context: Context, outer: str, inner: str, value: str
) -> None:
response = {outer: {inner: value}}
context.sandbox_rewritten_resp = context.sandbox_rewriter.rewrite_response(response)
@when('I roundtrip rewrite tool arguments with key "{key}" and value "{value}"')
def step_roundtrip_rewrite(context: Context, key: str, value: str) -> None:
arguments = {key: value}
rewritten = context.sandbox_rewriter.rewrite_arguments(arguments)
# Simulate the sandbox returning the rewritten path
context.sandbox_roundtrip_result = context.sandbox_rewriter.rewrite_response(
rewritten
)
# ---------------------------------------------------------------
# Then Steps
# ---------------------------------------------------------------
@then('the rewritten arguments should have "{key}" equal to "{expected}"')
def step_rewritten_arg_equal(context: Context, key: str, expected: str) -> None:
actual = context.sandbox_rewritten_args.get(key)
assert actual == expected, f"Expected '{expected}', got '{actual}'"
@then('the rewritten nested argument "{path}" should equal "{expected}"')
def step_rewritten_nested_arg(context: Context, path: str, expected: str) -> None:
parts = path.split(".")
value: Any = context.sandbox_rewritten_args
for part in parts:
value = value[part]
assert value == expected, f"Expected '{expected}', got '{value}'"
@then('the rewritten argument "{key}" should be a list with {count:d} rewritten paths')
def step_rewritten_arg_list(context: Context, key: str, count: int) -> None:
value = context.sandbox_rewritten_args.get(key)
assert isinstance(value, list), f"Expected list, got {type(value)}"
assert len(value) == count, f"Expected {count} items, got {len(value)}"
for item in value:
assert item.startswith("/workspace"), f"Path not rewritten: {item}"
@then('the rewritten response should have "{key}" equal to "{expected}"')
def step_rewritten_resp_equal(context: Context, key: str, expected: str) -> None:
actual = context.sandbox_rewritten_resp.get(key)
assert actual == expected, f"Expected '{expected}', got '{actual}'"
@then('the rewritten nested response "{path}" should equal "{expected}"')
def step_rewritten_nested_resp(context: Context, path: str, expected: str) -> None:
parts = path.split(".")
value: Any = context.sandbox_rewritten_resp
for part in parts:
value = value[part]
assert value == expected, f"Expected '{expected}', got '{value}'"
@then('the roundtrip path should equal "{expected}"')
def step_roundtrip_equal(context: Context, expected: str) -> None:
result = context.sandbox_roundtrip_result
for v in result.values():
if isinstance(v, str):
assert v == expected, f"Expected '{expected}', got '{v}'"
return
msg = f"No string value found in roundtrip result: {result}"
raise AssertionError(msg)
src/cleveragents/mcp/__init__.py
+15
View File
@@ -3,20 +3,35 @@
Bridges external MCP tool servers into the CleverAgents ToolRegistry
via ``MCPToolAdapter``. Includes ``MCPRefreshHook`` for wiring
``notifications/tools/list_changed`` events to ``SkillRegistry.refresh_all()``.
Lifecycle management is provided by ``McpClient`` (lazy start, auto-stop,
health monitoring) and ``McpRegistry`` (namespace-isolated server tracking).
Sandbox path rewriting is provided by ``SandboxPathRewriter``.
"""
from cleveragents.mcp.adapter import (
MCPCapabilityMetadata,
MCPServerConfig,
MCPToolAdapter,
MCPToolDescriptor,
MCPToolResult,
)
from cleveragents.mcp.client import McpClient, McpClientConfig, McpClientState
from cleveragents.mcp.refresh_hook import MCPRefreshHook
from cleveragents.mcp.registry import McpRegistry
from cleveragents.mcp.sandbox import SandboxPathRewriter, SandboxPathRewriterConfig
__all__ = [
"MCPCapabilityMetadata",
"MCPRefreshHook",
"MCPServerConfig",
"MCPToolAdapter",
"MCPToolDescriptor",
"MCPToolResult",
"McpClient",
"McpClientConfig",
"McpClientState",
"McpRegistry",
"SandboxPathRewriter",
"SandboxPathRewriterConfig",
]
src/cleveragents/mcp/adapter.py
+58
View File
@@ -38,6 +38,7 @@ from cleveragents.tool.runtime import ToolSpec
logger = logging.getLogger(__name__)
__all__ = [
"MCPCapabilityMetadata",
"MCPServerConfig",
"MCPToolAdapter",
"MCPToolDescriptor",
@@ -71,6 +72,30 @@ class MCPServerConfig(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True)
class MCPCapabilityMetadata(BaseModel):
"""Full MCP server capability metadata.
Exposes the complete capability information returned by the MCP
server during the initialize handshake, including supported
resources, prompts, and tool schemas.
Attributes:
tools: Whether the server supports tool invocation.
resources: Whether the server supports resource access.
prompts: Whether the server supports prompt templates.
raw_capabilities: Full raw capabilities dict from the server.
"""
tools: bool = Field(default=False, description="Supports tool invocation")
resources: bool = Field(default=False, description="Supports resource access")
prompts: bool = Field(default=False, description="Supports prompt templates")
raw_capabilities: dict[str, Any] = Field(
default_factory=dict, description="Raw capabilities from server"
)
model_config = ConfigDict(str_strip_whitespace=True)
class MCPToolDescriptor(BaseModel):
"""Descriptor for a tool discovered from an MCP server.
@@ -78,6 +103,7 @@ class MCPToolDescriptor(BaseModel):
name: Tool name as exposed by the MCP server.
description: Human-readable description of the tool.
input_schema: JSON Schema for tool inputs.
annotations: Optional tool annotations from the MCP server.
"""
name: str = Field(..., description="Tool name")
@@ -85,6 +111,9 @@ class MCPToolDescriptor(BaseModel):
input_schema: dict[str, Any] = Field(
default_factory=dict, description="JSON Schema for inputs"
)
annotations: dict[str, Any] = Field(
default_factory=dict, description="MCP tool annotations"
)
model_config = ConfigDict(str_strip_whitespace=True)
@@ -216,6 +245,27 @@ class MCPToolAdapter:
with self._lock:
return list(self._tools.values())
@property
def capability_metadata(self) -> MCPCapabilityMetadata:
"""Structured capability metadata from the MCP server.
Returns full capability information parsed from the server's
initialize handshake response. Handles both flat capability
dicts (``{"tools": True}``) and nested ones
(``{"capabilities": {"tools": True}}``).
"""
with self._lock:
raw = dict(self._capabilities)
# The transport may return capabilities at the top level or nested
# under a ``capabilities`` key depending on the MCP server.
caps = raw.get("capabilities", raw)
return MCPCapabilityMetadata(
tools=bool(caps.get("tools")),
resources=bool(caps.get("resources")),
prompts=bool(caps.get("prompts")),
raw_capabilities=raw,
)
def connect(self, timeout: float = 30.0) -> None:
"""Connect to the MCP server and perform the initialize handshake.
@@ -385,6 +435,7 @@ class MCPToolAdapter:
name=raw.get("name", ""),
description=raw.get("description", ""),
input_schema=raw.get("inputSchema", {}),
annotations=raw.get("annotations", {}),
)
descriptors.append(desc)
@@ -543,6 +594,7 @@ class MCPToolAdapter:
# 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.
cap_meta = self.capability_metadata
spec = ToolSpec(
name=tool_name,
description=desc.description or f"MCP tool: {desc.name}",
@@ -553,6 +605,12 @@ class MCPToolAdapter:
source_metadata={
"server": self._config.name,
"resource_slots": slot_dicts,
"annotations": desc.annotations,
"server_capabilities": {
"tools": cap_meta.tools,
"resources": cap_meta.resources,
"prompts": cap_meta.prompts,
},
},
)
registry.register(spec)
src/cleveragents/mcp/client.py
+468
View File
@@ -0,0 +1,468 @@
"""MCP Client with lazy start, auto-stop, and health monitoring.
Provides :class:`McpClient`, a high-level wrapper around
:class:`~cleveragents.mcp.adapter.MCPToolAdapter` that adds lifecycle
management features required by the specification:
- **Lazy start**: The MCP server is started on first ``call_tool()``
invocation rather than requiring an explicit ``start()`` call.
- **Auto-stop**: Idle servers are stopped after a configurable timeout
(default: 5 minutes).
- **Health monitoring**: Periodic health checks with automatic restart
on server crash.
Usage::
from cleveragents.mcp.client import McpClient, McpClientConfig
client = McpClient(McpClientConfig(
server=MCPServerConfig(name="gh", transport="stdio", command="gh-mcp"),
idle_timeout_seconds=300,
health_check_interval_seconds=30,
))
# Server starts automatically on first call:
result = client.call_tool("list_repos", {"org": "acme"})
# Shutdown when done:
client.shutdown()
"""
from __future__ import annotations
import logging
import threading
import time
from typing import TYPE_CHECKING, Any
from pydantic import BaseModel, ConfigDict, Field
from cleveragents.mcp.adapter import (
MCPServerConfig,
MCPToolAdapter,
MCPToolFilter,
MCPToolResult,
MCPTransport,
)
if TYPE_CHECKING:
pass
logger = logging.getLogger(__name__)
__all__ = [
"McpClient",
"McpClientConfig",
"McpClientState",
]
# Default idle timeout: 5 minutes.
_DEFAULT_IDLE_TIMEOUT_SECONDS: float = 300.0
# Default health-check interval: 30 seconds.
_DEFAULT_HEALTH_CHECK_INTERVAL_SECONDS: float = 30.0
class McpClientConfig(BaseModel):
"""Configuration for an :class:`McpClient` instance.
Attributes:
server: MCP server connection configuration.
idle_timeout_seconds: Seconds of inactivity before auto-stop.
health_check_interval_seconds: Seconds between health probes.
auto_restart: Whether to restart the server on health-check failure.
lazy_start: Whether to defer connection until first tool call.
"""
server: MCPServerConfig
idle_timeout_seconds: float = Field(
default=_DEFAULT_IDLE_TIMEOUT_SECONDS,
ge=0,
description="Idle timeout before auto-stop (0 = disabled)",
)
health_check_interval_seconds: float = Field(
default=_DEFAULT_HEALTH_CHECK_INTERVAL_SECONDS,
ge=0,
description="Health-check interval (0 = disabled)",
)
auto_restart: bool = Field(
default=True,
description="Auto-restart on health-check failure",
)
lazy_start: bool = Field(
default=True,
description="Defer connection until first tool call",
)
model_config = ConfigDict(str_strip_whitespace=True)
class McpClientState:
"""Observable state of an :class:`McpClient`."""
IDLE = "idle"
STARTING = "starting"
RUNNING = "running"
STOPPING = "stopping"
STOPPED = "stopped"
ERROR = "error"
class McpClient:
"""High-level MCP client with lifecycle management.
Wraps an :class:`MCPToolAdapter` and adds lazy start, idle auto-stop,
and health monitoring with automatic restart.
Parameters
----------
config:
Client configuration including server config and lifecycle params.
transport:
Optional transport override (for testing).
"""
def __init__(
self,
config: McpClientConfig,
transport: MCPTransport | None = None,
) -> None:
self._config = config
self._adapter = MCPToolAdapter(
config=config.server,
transport=transport,
)
self._lock = threading.RLock()
self._state = McpClientState.IDLE
self._last_activity: float = 0.0
self._idle_timer: threading.Timer | None = None
self._health_timer: threading.Timer | None = None
self._health_check_failures: int = 0
self._restart_count: int = 0
self._started = False
self._shutting_down = False
self._transport_override = transport
# -- Properties -----------------------------------------------------------
@property
def state(self) -> str:
"""Current lifecycle state."""
with self._lock:
return self._state
@property
def server_name(self) -> str:
"""Name of the underlying MCP server."""
return self._config.server.name
@property
def is_connected(self) -> bool:
"""Whether the adapter is currently connected."""
return self._adapter.is_connected
@property
def adapter(self) -> MCPToolAdapter:
"""Underlying MCP tool adapter."""
return self._adapter
@property
def last_activity(self) -> float:
"""Monotonic timestamp of last tool invocation."""
with self._lock:
return self._last_activity
@property
def idle_seconds(self) -> float:
"""Seconds since last tool invocation."""
with self._lock:
if self._last_activity == 0.0:
return 0.0
return time.monotonic() - self._last_activity
@property
def health_check_failures(self) -> int:
"""Number of consecutive health-check failures."""
with self._lock:
return self._health_check_failures
@property
def restart_count(self) -> int:
"""Number of automatic restarts triggered by health monitoring."""
with self._lock:
return self._restart_count
# -- Lifecycle API --------------------------------------------------------
def start(self) -> None:
"""Explicitly start the MCP server connection.
If ``lazy_start`` is enabled, this is called automatically on first
``call_tool()`` invocation. Calling ``start()`` explicitly is
always safe and idempotent.
Raises
------
ConnectionError
If the MCP server cannot be reached.
"""
with self._lock:
if self._started:
return
self._state = McpClientState.STARTING
try:
self._adapter.connect()
self._adapter.discover_tools()
except Exception:
with self._lock:
self._state = McpClientState.ERROR
raise
with self._lock:
self._started = True
self._state = McpClientState.RUNNING
self._last_activity = time.monotonic()
self._health_check_failures = 0
self._schedule_idle_timer()
self._schedule_health_check()
logger.info("McpClient '%s' started", self.server_name)
def shutdown(self) -> None:
"""Stop the MCP server connection and cancel all timers.
Safe to call multiple times and from any thread.
"""
with self._lock:
if self._shutting_down:
return
self._shutting_down = True
self._state = McpClientState.STOPPING
self._cancel_idle_timer()
self._cancel_health_check()
self._adapter.disconnect()
with self._lock:
self._started = False
self._state = McpClientState.STOPPED
self._shutting_down = False
logger.info("McpClient '%s' shut down", self.server_name)
def _ensure_started(self) -> None:
"""Ensure the client is started (lazy start support).
If lazy_start is enabled and the client hasn't started yet,
triggers ``start()``.
Raises
------
RuntimeError
If the client is not started and lazy_start is disabled.
"""
with self._lock:
if self._started:
return
if not self._config.lazy_start:
msg = (
f"McpClient '{self.server_name}' is not started "
f"and lazy_start is disabled"
)
raise RuntimeError(msg)
self.start()
# -- Tool Invocation ------------------------------------------------------
def call_tool(
self,
tool_name: str,
arguments: dict[str, Any],
) -> MCPToolResult:
"""Invoke an MCP tool, starting the server if needed.
This is the primary entry point for tool invocation. When
``lazy_start`` is enabled, the server connection is established
on the first call.
Parameters
----------
tool_name:
Name of the tool as discovered from the server.
arguments:
Tool input arguments.
Returns
-------
MCPToolResult
Invocation outcome.
"""
self._ensure_started()
self._touch_activity()
result = self._adapter.invoke(tool_name, arguments)
self._touch_activity()
return result
def discover_tools(
self,
tool_filter: MCPToolFilter | None = None,
) -> list[Any]:
"""Discover tools from the MCP server.
Triggers lazy start if needed.
"""
self._ensure_started()
self._touch_activity()
return self._adapter.discover_tools(tool_filter)
# -- Idle Timeout ---------------------------------------------------------
def _touch_activity(self) -> None:
"""Record current time as the last activity timestamp."""
with self._lock:
self._last_activity = time.monotonic()
self._schedule_idle_timer()
def _schedule_idle_timer(self) -> None:
"""Schedule the idle auto-stop timer."""
self._cancel_idle_timer()
timeout = self._config.idle_timeout_seconds
if timeout <= 0:
return
with self._lock:
timer = threading.Timer(timeout, self._check_idle)
timer.daemon = True
self._idle_timer = timer
timer.start()
def _cancel_idle_timer(self) -> None:
"""Cancel any pending idle timer."""
with self._lock:
if self._idle_timer is not None:
self._idle_timer.cancel()
self._idle_timer = None
def _check_idle(self) -> None:
"""Check if the client has been idle beyond the timeout."""
with self._lock:
if self._shutting_down or not self._started:
return
elapsed = time.monotonic() - self._last_activity
if elapsed < self._config.idle_timeout_seconds:
# Activity happened while timer was pending; reschedule.
remaining = self._config.idle_timeout_seconds - elapsed
timer = threading.Timer(remaining, self._check_idle)
timer.daemon = True
self._idle_timer = timer
timer.start()
return
logger.info(
"McpClient '%s' idle for %.1fs — auto-stopping",
self.server_name,
elapsed,
)
self.shutdown()
# -- Health Monitoring ----------------------------------------------------
def _schedule_health_check(self) -> None:
"""Schedule the next health-check timer."""
self._cancel_health_check()
interval = self._config.health_check_interval_seconds
if interval <= 0:
return
with self._lock:
timer = threading.Timer(interval, self._perform_health_check)
timer.daemon = True
self._health_timer = timer
timer.start()
def _cancel_health_check(self) -> None:
"""Cancel any pending health-check timer."""
with self._lock:
if self._health_timer is not None:
self._health_timer.cancel()
self._health_timer = None
def _perform_health_check(self) -> None:
"""Perform a health check by pinging the MCP server.
If the check fails and ``auto_restart`` is enabled, the client
is restarted. After a successful restart, the failure counter
resets.
"""
with self._lock:
if self._shutting_down or not self._started:
return
healthy = self._check_health()
if healthy:
with self._lock:
self._health_check_failures = 0
self._schedule_health_check()
return
with self._lock:
self._health_check_failures += 1
failures = self._health_check_failures
logger.warning(
"McpClient '%s' health check failed (%d consecutive)",
self.server_name,
failures,
)
if self._config.auto_restart:
self._do_restart()
else:
self._schedule_health_check()
def _check_health(self) -> bool:
"""Probe the MCP server by attempting a tools/list call.
Returns ``True`` if the server responds, ``False`` otherwise.
"""
try:
self._adapter.discover_tools()
return True
except Exception:
logger.debug(
"Health check probe failed for '%s'",
self.server_name,
exc_info=True,
)
return False
def _do_restart(self) -> None:
"""Restart the MCP server connection."""
logger.info(
"McpClient '%s' restarting after health-check failure", self.server_name
)
try:
self._adapter.reconnect()
self._adapter.discover_tools()
except Exception:
logger.error(
"McpClient '%s' restart failed",
self.server_name,
exc_info=True,
)
with self._lock:
self._state = McpClientState.ERROR
self._schedule_health_check()
return
with self._lock:
self._restart_count += 1
self._health_check_failures = 0
self._state = McpClientState.RUNNING
self._schedule_health_check()
logger.info("McpClient '%s' restarted successfully", self.server_name)
src/cleveragents/mcp/registry.py
+186
View File
@@ -0,0 +1,186 @@
"""MCP Server Registry — namespace-isolated lifecycle tracking.
Manages multiple :class:`~cleveragents.mcp.client.McpClient` instances
with independent lifecycles. Each server is identified by a unique
namespace and can be started, stopped, and health-checked independently.
Usage::
from cleveragents.mcp.registry import McpRegistry
registry = McpRegistry()
registry.register(client_config, namespace="github")
registry.register(client_config_2, namespace="jira")
# Clients start lazily on first tool call.
result = registry.call_tool("github", "list_repos", {"org": "acme"})
# Shut down all servers:
registry.shutdown_all()
"""
from __future__ import annotations
import logging
import threading
from typing import Any
from cleveragents.mcp.adapter import MCPToolResult, MCPTransport
from cleveragents.mcp.client import McpClient, McpClientConfig
logger = logging.getLogger(__name__)
__all__ = [
"McpRegistry",
]
class McpRegistry:
"""Registry of MCP clients indexed by namespace.
Provides a unified interface for managing multiple MCP server
connections with namespace isolation.
"""
def __init__(self) -> None:
self._clients: dict[str, McpClient] = {}
self._transports: dict[str, MCPTransport | None] = {}
self._lock = threading.RLock()
def register(
self,
config: McpClientConfig,
namespace: str,
transport: MCPTransport | None = None,
) -> McpClient:
"""Register an MCP server configuration under a namespace.
If a client already exists for the namespace, it is shut down
and replaced.
Parameters
----------
config:
Client configuration for the MCP server.
namespace:
Unique namespace identifier for the server.
transport:
Optional transport override (for testing).
Returns
-------
McpClient
The newly created client.
"""
with self._lock:
existing = self._clients.get(namespace)
if existing is not None:
existing.shutdown()
client = McpClient(config=config, transport=transport)
self._clients[namespace] = client
self._transports[namespace] = transport
logger.info(
"Registered MCP server '%s' in namespace '%s'",
config.server.name,
namespace,
)
return client
def unregister(self, namespace: str) -> None:
"""Unregister and shut down the client for a namespace.
Parameters
----------
namespace:
Namespace to remove.
Raises
------
KeyError
If no client is registered under the namespace.
"""
with self._lock:
client = self._clients.pop(namespace, None)
self._transports.pop(namespace, None)
if client is None:
msg = f"No MCP client registered for namespace '{namespace}'"
raise KeyError(msg)
client.shutdown()
logger.info("Unregistered MCP namespace '%s'", namespace)
def get(self, namespace: str) -> McpClient | None:
"""Retrieve the client for a namespace, or ``None``."""
with self._lock:
return self._clients.get(namespace)
def list_namespaces(self) -> list[str]:
"""Return all registered namespace names."""
with self._lock:
return list(self._clients.keys())
def call_tool(
self,
namespace: str,
tool_name: str,
arguments: dict[str, Any],
) -> MCPToolResult:
"""Invoke a tool on a specific namespace's MCP server.
The server is started lazily if not already running.
Parameters
----------
namespace:
Namespace of the MCP server.
tool_name:
Name of the tool to invoke.
arguments:
Tool input arguments.
Returns
-------
MCPToolResult
Invocation outcome.
Raises
------
KeyError
If no client is registered under the namespace.
"""
with self._lock:
client = self._clients.get(namespace)
if client is None:
msg = f"No MCP client registered for namespace '{namespace}'"
raise KeyError(msg)
return client.call_tool(tool_name, arguments)
def shutdown_all(self) -> None:
"""Shut down all registered MCP clients."""
with self._lock:
clients = list(self._clients.values())
for client in clients:
try:
client.shutdown()
except Exception:
logger.warning(
"Error shutting down MCP client '%s'",
client.server_name,
exc_info=True,
)
logger.info("All MCP clients shut down")
def __len__(self) -> int:
with self._lock:
return len(self._clients)
def __contains__(self, namespace: str) -> bool:
with self._lock:
return namespace in self._clients
src/cleveragents/mcp/sandbox.py
+179
View File
@@ -0,0 +1,179 @@
"""Sandbox path rewriting layer for MCP tool arguments and responses.
Provides :class:`SandboxPathRewriter` which rewrites file paths in MCP
tool call arguments (host -> sandbox) and in tool call responses
(sandbox -> host). This enables MCP tools to operate inside a
sandboxed workspace while the host sees paths in its own namespace.
The rewriter uses :class:`~cleveragents.tool.path_mapper.PathMapper`
for the actual bi-directional path translation and applies rewriting
to all string values in nested dicts/lists that look like absolute
file paths under the configured roots.
Usage::
from cleveragents.mcp.sandbox import SandboxPathRewriter
rewriter = SandboxPathRewriter(
host_root="/home/user/project",
sandbox_root="/workspace",
)
# Rewrite arguments before sending to MCP server:
rewritten_args = rewriter.rewrite_arguments(
{"file_path": "/home/user/project/src/main.py"}
)
# -> {"file_path": "/workspace/src/main.py"}
# Rewrite response paths from sandbox back to host:
rewritten_resp = rewriter.rewrite_response(
{"output_file": "/workspace/build/out.txt"}
)
# -> {"output_file": "/home/user/project/build/out.txt"}
"""
from __future__ import annotations
import logging
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from cleveragents.tool.path_mapper import PathMapper
logger = logging.getLogger(__name__)
__all__ = [
"SandboxPathRewriter",
"SandboxPathRewriterConfig",
]
class SandboxPathRewriterConfig(BaseModel):
"""Configuration for sandbox path rewriting.
Attributes:
host_root: Absolute path to the host-side workspace root.
sandbox_root: Absolute path to the sandbox-side workspace root.
enabled: Whether path rewriting is active.
"""
host_root: str = Field(..., min_length=1, description="Host workspace root")
sandbox_root: str = Field(..., min_length=1, description="Sandbox workspace root")
enabled: bool = Field(default=True, description="Enable path rewriting")
model_config = ConfigDict(str_strip_whitespace=True)
class SandboxPathRewriter:
"""Bi-directional path rewriter for MCP sandbox isolation.
Rewrites absolute file paths in tool arguments and responses between
host and sandbox namespaces using :class:`PathMapper`.
Parameters
----------
host_root:
Absolute path to the host-side workspace root.
sandbox_root:
Absolute path to the sandbox-side workspace root.
"""
def __init__(self, host_root: str, sandbox_root: str) -> None:
self._mapper = PathMapper(host_root=host_root, container_root=sandbox_root)
self._host_root = host_root
self._sandbox_root = sandbox_root
@classmethod
def from_config(cls, config: SandboxPathRewriterConfig) -> SandboxPathRewriter:
"""Create a rewriter from a configuration object."""
return cls(host_root=config.host_root, sandbox_root=config.sandbox_root)
@property
def host_root(self) -> str:
"""Host workspace root path."""
return self._host_root
@property
def sandbox_root(self) -> str:
"""Sandbox workspace root path."""
return self._sandbox_root
def rewrite_arguments(self, arguments: dict[str, Any]) -> dict[str, Any]:
"""Rewrite host paths to sandbox paths in tool arguments.
Parameters
----------
arguments:
Tool call arguments dict.
Returns
-------
dict[str, Any]
Arguments with host paths replaced by sandbox paths.
"""
return self._rewrite_value(arguments, self._host_to_sandbox)
def rewrite_response(self, response: dict[str, Any]) -> dict[str, Any]:
"""Rewrite sandbox paths to host paths in tool responses.
Parameters
----------
response:
Tool call response dict.
Returns
-------
dict[str, Any]
Response with sandbox paths replaced by host paths.
"""
return self._rewrite_value(response, self._sandbox_to_host)
def _host_to_sandbox(self, path: str) -> str:
"""Map a host path to a sandbox path."""
return self._mapper.host_to_container(path)
def _sandbox_to_host(self, path: str) -> str:
"""Map a sandbox path to a host path."""
return self._mapper.container_to_host(path)
def _rewrite_value(
self,
value: Any,
rewrite_fn: Any,
) -> Any:
"""Recursively rewrite string values that look like paths.
Parameters
----------
value:
Any JSON-compatible value (str, dict, list, number, bool, None).
rewrite_fn:
Function that maps a single path string.
Returns
-------
Any
The value with path strings rewritten.
"""
if isinstance(value, str):
return self._maybe_rewrite_path(value, rewrite_fn)
if isinstance(value, dict):
return {k: self._rewrite_value(v, rewrite_fn) for k, v in value.items()}
if isinstance(value, list):
return [self._rewrite_value(item, rewrite_fn) for item in value]
return value
@staticmethod
def _maybe_rewrite_path(
value: str,
rewrite_fn: Any,
) -> str:
"""Rewrite a string if it looks like an absolute path.
Only strings starting with ``/`` are considered candidates for
path rewriting to avoid mutating non-path strings.
"""
if not value.startswith("/"):
return value
return rewrite_fn(value)