ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity
Files
bd35da11b6d8a4c8d90edba4873433732448fe85
manual_slop/docs/guide_mcp_client.md
T

22 KiB
Raw Blame History

src/mcp_client.py — Agent Tools (45 MCP + 1 shell = 46 total, 3-layer security)

Top | Architecture | Tools & IPC | Testing

Overview

src/mcp_client.py (~81KB) is the MCP (Model Context Protocol) tool implementation for Manual Slop. It provides 45 tools that the AI can invoke to read/write files, analyze code structure, search symbols, and more. The canonical 46-tool list (which includes run_powershell from src/shell_runner.py) lives in models.AGENT_TOOL_NAMES (src/models.py:58). This guide covers the 45 MCP tools; run_powershell is documented in guide_tools.md.

The module implements the client side of MCP — it provides the tools that an AI model can call during a conversation. It also implements the project's strict filesystem security model.

Architecture

┌─────────────────────────────────────────────────┐
│  ai_client.send(...)                              │
│  AI returns: { "name": "read_file", "args": {...} } │
└─────────────────┬───────────────────────────────┘
                  │ calls
                  ▼
┌─────────────────────────────────────────────────┐
│  mcp_client.dispatch(tool_name, tool_input)      │
│  Routes to the registered tool function           │
│  Returns tool result as string                    │
└─────────────────┬───────────────────────────────┘
                  │ calls (with security check)
                  ▼
┌─────────────────────────────────────────────────┐
│  3-Layer Security: Allowlist → Validate → Resolve │
└─────────────────┬───────────────────────────────┘
                  │ passes
                  ▼
┌─────────────────────────────────────────────────┐
│  The actual tool function (45 of them)            │
└─────────────────────────────────────────────────┘

The 3-Layer Security Model

Every filesystem access passes through 3 layers:

Layer 1: Allowlist Construction (configure)

Called by ai_client before each send cycle (and on project load):

def configure(file_items: list[dict], base_dirs: list[str]) -> None:
    """Build the allowlist from the project's tracked files and base dirs."""
    _allowed_paths.clear()
    _base_dirs.clear()
    if base_dirs:
        _primary_base_dir = Path(base_dirs[0]).resolve()
    for f in file_items:
        if isinstance(f, dict) and "path" in f:
            _allowed_paths.add(Path(f["path"]).resolve())
    # Blacklist: history.toml and config.toml are NEVER allowed

After this call, _allowed_paths contains every file the AI is allowed to touch.

Layer 2: Path Validation (_is_allowed)

Called on every path before any I/O:

def _is_allowed(path: Path) -> bool:
    """Return True if `path` is within the allowlist."""
    if path.name in {"history.toml", "config.toml", "credentials.toml"}:
        return False
    if "*_history.toml" in path.name:
        return False
    # ... checks against _allowed_paths and _base_dirs

Returns False for any path the AI is not allowed to touch.

Layer 3: Resolution Gate (_resolve_and_check)

The final gate. Resolves the path (handling symlinks, relative paths) and re-checks.

As of 2026-06-11: This section documents the post-refactor Result[Path] signature, applied by the data_oriented_error_handling_20260606 track. The pre-refactor (Path | None, str) tuple and the 30+ assert p is not None chain in tool bodies (lines 304-794) are replaced. See the new Data-Oriented Error Handling (Fleury Pattern) section below for the full convention.

def _resolve_and_check(raw_path: str) -> Result[Path]:
    """Resolve raw_path and verify it passes the allowlist check.

    On success: result.data is the real pathlib.Path; result.errors is [].
    On failure: result.data is NIL_PATH; result.errors has 1 ErrorInfo
      with kind=ErrorKind.PERMISSION (or NOT_FOUND / INVALID_INPUT).
    """
    p = Path(raw_path).resolve()
    if not _is_allowed(p):
        return Result(
            data=NIL_PATH,
            errors=[ErrorInfo(
                kind=ErrorKind.PERMISSION,
                message=f"path not in allowlist: {raw_path}",
                source="mcp._resolve_and_check",
            )],
        )
    return Result(data=p)

Every tool function calls this first. If result.errors is non-empty, the tool returns its own Result[data="", errors=resolved.errors] to propagate the gate's error to the AI. The 3-layer security model is preserved unchanged — only the return-type contract evolves.

The 45 Tools

The tools are organized by category. The full registered count is 45.

File I/O Tools (4)

Tool Parameters Description
read_file path UTF-8 file content extraction
list_directory path Compact table: [file/dir] name size. Applies blacklist filter.
search_files path, pattern Glob pattern matching within an allowed directory
get_file_summary path Heuristic summary via summarize.py (imports, classes, etc.)

File Edit Tools (3)

Tool Parameters Description
get_file_slice path, start_line, end_line Returns specific line range (1-based, inclusive)
set_file_slice path, start_line, end_line, new_content Replaces a line range with new content
edit_file path, old_string, new_string, replace_all Exact string match replace. Preserves indentation.

Python AST Tools (18)

Tool Parameters Description
py_get_skeleton path Skeleton: signatures + docstrings, bodies replaced with ...
py_get_code_outline path Hierarchical outline: classes, functions, methods with line ranges
py_get_definition path, name Full source for a class/function/method
py_update_definition path, name, new_content Surgical replacement (locates via ast, delegates to set_file_slice)
py_get_signature path, name Just the def line through the colon
py_set_signature path, name, new_signature Replaces only the signature, preserving body
py_get_class_summary path, name Class docstring + method signatures
py_get_var_declaration path, name Module/class-level variable assignment line(s)
py_set_var_declaration path, name, new_declaration Surgical variable replacement
py_get_docstring path, name Docstring for module/class/function
py_find_usages path, name All references to a symbol in a file/dir
py_get_imports path AST-derived dependency list (modules, froms, names)
py_check_syntax path Quick syntax check via ast.parse()
py_get_hierarchy path, class_name Subclasses of a given class
py_remove_def path, name AST-derived line-range excision of a class/function
py_add_def path, name, new_content, anchor_type, anchor_symbol Inserts a new definition at a specific anchor
py_move_def src_path, dest_path, name, dest_name, anchor_type, anchor_symbol Relocates a definition within/across files
py_region_wrap path, start_line, end_line, region_name Wraps a block in #region: Name / #endregion: Name
py_get_imports path AST-parsed dependency list
py_find_usages path, name Exact string match search
py_check_syntax path Syntax validation via ast.parse()
py_remove_def path, name Excises a definition
py_add_def path, name, new_content, anchor_type, anchor_symbol Inserts with 1-space indent normalization
py_move_def src_path, dest_path, name, dest_name, anchor_type, anchor_symbol Relocates code
py_region_wrap path, start_line, end_line, region_name Wraps line range in #region / #endregion

C/C++ AST Tools (10)

Tool Parameters Description
ts_c_get_skeleton path C function signatures and struct definitions, bodies replaced
ts_cpp_get_skeleton path C++ class/struct/method signatures, inheritance
ts_c_get_code_outline path C outline
ts_cpp_get_code_outline path C++ outline with classes and inheritance
ts_c_get_definition path, name C struct or function source
ts_cpp_get_definition path, name C++ class, struct, or method source (supports Class::method)
ts_c_update_definition path, name, new_content Surgical C replacement
ts_cpp_update_definition path, name, new_content Surgical C++ replacement
ts_c_get_signature path, name C function/struct declaration
ts_cpp_get_signature path, name C++ method/function declaration

All C/C++ tools use tree-sitter (via src/file_cache.py's ASTParser).

Analysis Tools (3)

Tool Parameters Description
derive_code_path target, max_depth Traces execution path of a function across multiple files
py_get_imports path AST-parsed dependency list
py_find_usages path, name String match search

Network Tools (2)

Tool Parameters Description
web_search query DuckDuckGo HTML scrape via _DDGParser (HTMLParser subclass). Returns top 5 results.
fetch_url url Fetches URL, strips HTML via _TextExtractor

Runtime Tools (1)

Tool Parameters Description
get_ui_performance (none) FPS, Frame Time, CPU, Input Lag via injected perf_monitor_callback

Beads Tools (4)

Tool Parameters Description
bd_list (none) Lists all beads in active .beads/ repo
bd_create title, description Creates a new bead
bd_update bead_id, status Updates bead status
bd_ready (none) Lists beads with no unresolved dependencies

The dispatch Function

The single entry point for all tool calls.

def dispatch(tool_name: str, tool_input: dict[str, Any]) -> str:
    """Dispatch an MCP tool call by name. Returns the result as a string."""

Returns the result as a string (errors included). The AI client receives this string and appends it to the conversation history.

Dispatch Flow

def dispatch(tool_name, tool_input):
    if tool_name.startswith("bd_"):
        return _dispatch_beads(tool_name, tool_input)
    if tool_name == "read_file":
        return _read_file(tool_input["path"])
    if tool_name == "py_get_skeleton":
        return _py_get_skeleton(tool_input["path"])
    # ... etc, one branch per tool ...
    return f"ERROR: unknown tool: {tool_name}"

The bd_* tools are dispatched separately because they require an active .beads/ repository.

Async Dispatch

async def async_dispatch(tool_name, tool_input) -> str:
    """Async version of dispatch. Uses asyncio.to_thread for blocking I/O."""
    return await asyncio.to_thread(dispatch, tool_name, tool_input)

For concurrent tool execution (when the AI emits multiple calls in one turn), the AI client uses asyncio.gather over async_dispatch.

The 3-Layer Security Details

Blacklist

Always blocked, regardless of allowlist:

  • history.toml
  • *_history.toml
  • config.toml
  • credentials.toml

These are matched by exact filename (no path component) in _is_allowed.

Allowlist Construction Order

def configure(file_items, base_dirs):
    # Reset
    _allowed_paths.clear()
    _base_dirs.clear()

    # Primary base dir from first entry
    if base_dirs:
        _primary_base_dir = Path(base_dirs[0]).resolve()
        _base_dirs.add(_primary_base_dir)

    # Add all file item paths
    for f in file_items:
        if isinstance(f, dict) and "path" in f:
            try:
                _allowed_paths.add(Path(f["path"]).resolve())
            except (OSError, ValueError):
                pass

_primary_base_dir is the first base_dir; it's used for relative-path resolution.

Resolution Edge Cases

_resolve_and_check handles:

  • Absolute paths (used as-is)
  • Relative paths (resolved against _primary_base_dir)
  • Symlinks (resolved via Path.resolve())
  • Windows path separators
  • UNC paths

If resolution fails (e.g., path doesn't exist), returns the error to the AI.

External MCP Servers

In addition to the 45 native tools, mcp_client.py manages external MCP servers via ExternalMCPManager:

StdioMCPServer (in src/mcp_client.py)

Manages a local MCP server via subprocess (stdin/stdout).

server = StdioMCPServer(
    name="my_server",
    command=["python", "-m", "my_mcp_server"],
    cwd="/path/to/server",
)
server.start()
tools = server.list_tools()  # Get the server's tool schemas
result = server.call_tool("tool_name", {"arg": "value"})
server.stop()

RemoteMCPServer (in src/mcp_client.py)

SSE-based remote MCP server integration. Foundation for connecting to remote MCP services.

ExternalMCPManager

Manages multiple server lifecycles:

manager = ExternalMCPManager()
manager.add_server(server_config)  # Stdio or Remote
tools = manager.get_all_tools()  # All tools from all servers
manager.stop_all()

The dispatch function transparently routes calls to external server tools as well as native ones.

JSON-RPC 2.0 Engine

External MCP servers use JSON-RPC 2.0 over their respective transports (stdio or SSE). The MCP client implements:

  • Request ID generation
  • Async request/response matching
  • Timeout handling
  • Error code mapping (JSON-RPC error codes → string error messages)

Public API

Function Purpose
configure(file_items, base_dirs) Build the allowlist
dispatch(tool_name, tool_input) Call a tool by name
async_dispatch(tool_name, tool_input) Async version
get_tool_schemas() -> list[dict] All tool schemas (for AI capability declaration)
is_allowed(path: str) -> bool Check if a path is allowed (for testing)
get_external_mcp_manager() -> ExternalMCPManager Get the singleton manager

Configuration

# config.toml
[mcp]
enabled = true
blacklist = ["*.pem", "*.key", ".env"]  # Additional patterns to always block
allow_symlinks = false

External MCP server config (mcp_config.json, standard format):

{
    "servers": [
        {
            "name": "filesystem",
            "command": ["python", "-m", "filesystem_mcp"],
            "env": {}
        }
    ]
}

Located at <user_config>/mcp_config.json or <project_root>/mcp_config.json.

Testing

Unit Tests

tests/test_arch_boundary_phase1.py::test_mcp_client_whitelist_enforcement — tests the security model.

tests/test_mcp_ts_integration.py — tests tree-sitter integration.

Mocking MCP

Tests that don't want real filesystem access can mock the dispatch:

from unittest.mock import patch

def test_my_code(monkeypatch):
    def fake_dispatch(name, args):
        if name == "read_file":
            return "mocked content"
        return ""
    monkeypatch.setattr("src.mcp_client.dispatch", fake_dispatch)
    # ... test code ...

Performance

  • File I/O: synchronous (blocks the calling thread). Use async_dispatch for parallel calls.
  • Tree-sitter parsing: ~10-50ms per file for typical Python files. Cached in _ast_cache (mtime-based).
  • Network tools (web_search, fetch_url): 100ms-2s depending on the network.

Data-Oriented Error Handling (Fleury Pattern)

The MCP tool layer follows the "errors are just cases" framework (Ryan Fleury). The canonical reference is conductor/code_styleguides/error_handling.md.

Result-Based Returns

The 9 tool functions that previously returned (Path | None, str) tuples or raised exceptions now return Result[str] for content and Result[Path] for the resolution gate:

Function Old signature New signature
_resolve_and_check(raw_path) tuple[Path | None, str] Result[Path] (data is real Path or NIL_PATH)
read_file(path) str (error prefix) Result[str] (data is "" on failure)
list_directory(path) str (error prefix) Result[str] (data is "" on failure)
search_files(...) str (error prefix) Result[str] (data is "" on failure)
get_file_summary(path) str (error prefix) Result[str] (data is "" on failure)
py_get_skeleton(path) str (error prefix) Result[str] (data is "" on failure)
py_get_code_outline(path) str (error prefix) Result[str] (data is "" on failure)
py_get_definition(path, name) str (error prefix) Result[str] (data is "" on failure)
py_get_imports(path) str (error prefix) Result[str] (data is "" on failure)
(and 35 more — all 45 tools) str (error prefix) Result[str] (data is "" on failure)

Nil-Sentinel Pattern

The NIL_PATH dataclass is the "empty path" — it has all default values (exists=False, read_text="", errors=[]) and is safe to read from:

@dataclass(frozen=True)
class NilPath:
    exists: bool = False
    read_text: str = ""
    errors: list[ErrorInfo] = field(default_factory=list)

NIL_PATH = NilPath()  # module-level singleton

Callers that need a real pathlib.Path for filesystem operations check if isinstance(result.data, NilPath): handle() — but most callers just need the read text, and NIL_PATH.read_text == "" is fine for the AI model's purposes. This eliminates the 30+ assert p is not None chain in tool bodies (lines 304-794 pre-refactor) and the if err or p is None: return err patterns at the top of every tool function.

Dispatch Internals

The dispatch and async_dispatch functions unwrap the Result before returning to the AI model (so the model's view of MCP errors is unchanged — it still sees error messages as plain strings):

def dispatch(tool_name: str, tool_input: dict) -> str:
    result = _DISPATCH_TABLE[tool_name](tool_input)
    if not result.ok:
        for err in result.errors:
            _append_comms("WARN", "mcp_tool_error", [err.ui_message()])
    return result.data or "".join(e.message for e in result.errors)

The async_dispatch path handles the case where mcp_client has no comms log: it just returns result.data (the empty success value) and the errors are silently dropped. The Result's data field is always readable (zero-initialized) so callers don't need defensive is None checks.

Example

from src import mcp_client
from src.result_types import ErrorKind

r = mcp_client.read_file("/path/to/file.py")
if r.errors:
    for err in r.errors:
        if err.kind == ErrorKind.PERMISSION:
            log.warning("path not in allowlist: %s", err.message)
        elif err.kind == ErrorKind.NOT_FOUND:
            log.info("file not found: %s", err.message)
        else:
            log.error(err.ui_message())
# use r.data regardless (it's the zero-initialized "" on failure)
process(r.data)

Security Invariant

The 3-layer security model (Allowlist → Validate → Resolve) is preserved unchanged by the refactor. The new Result return type only changes the signature of the tool functions; the behavior (the 3 layers must all pass) is identical. The ErrorKind.PERMISSION value is what the model sees when the allowlist rejects a path — same error condition as the pre-refactor "ERROR: path not in allowlist: ..." string, just typed data instead of stringly-typed control flow.

See Also (in-doc)

See Also

View Git Blame Copy Permalink