diff --git a/CHANGES b/CHANGES index 436ff834..ad8aaa66 100644 --- a/CHANGES +++ b/CHANGES @@ -6,6 +6,12 @@ _Notes on upcoming releases will be added here_ +### What's new + +**Tier-aware tool batching** + +{tooliconl}`call-readonly-tools-batch`, {tooliconl}`call-mutating-tools-batch`, and {tooliconl}`call-destructive-tools-batch` run an ordered list of existing MCP tools in a single call and return a per-operation result for each, preserving every nested tool's own structured output. Each wrapper caps the safety tier of the calls it will make — the readonly wrapper refuses mutating or destructive operations, and the mutating wrapper refuses destructive ones — regardless of the server's `LIBTMUX_SAFETY` tier. Nested calls keep their normal schema validation, middleware, and safety checks, and `on_error` selects stop-at-first-failure or continue-and-report handling. Large aggregate results stay within the server's response limit — oversized nested payloads are dropped (with the truncation flagged in the result), and very large operation lists are rejected rather than allowed to overflow it. (#79) + ## libtmux-mcp 0.1.0a13 (2026-06-13) libtmux-mcp 0.1.0a13 adds {tooliconl}`send-keys-batch` for sending an ordered batch of raw key/text operations to tmux panes in a single call, with per-operation results, stop-or-continue error handling, and an optional timeout that bounds both the batch and each send. Argument-validation failures also stop echoing the rejected input into the server's logs and tool error results, so a secret-bearing argument can no longer surface there. diff --git a/README.md b/README.md index 27435cc6..b442d375 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ Give your AI agent hands inside the terminal — create sessions, run commands, | Module | Tools | |--------|-------| | **Server** | `list_servers`, `list_sessions`, `create_session`, `kill_server`, `get_server_info` | +| **Batch** | `call_readonly_tools_batch`, `call_mutating_tools_batch`, `call_destructive_tools_batch` | | **Session** | `list_windows`, `get_session_info`, `create_window`, `rename_session`, `select_window`, `kill_session` | | **Window** | `list_panes`, `get_window_info`, `split_window`, `rename_window`, `select_layout`, `resize_window`, `move_window`, `kill_window` | | **Pane** | `run_command`, `send_keys`, `send_keys_batch`, `paste_text`, `capture_pane`, `capture_since`, `snapshot_pane`, `search_panes`, `find_pane_by_position`, `get_pane_info`, `wait_for_text`, `wait_for_content_change`, `wait_for_channel`, `signal_channel`, `display_message`, `select_pane`, `swap_pane`, `resize_pane`, `set_pane_title`, `clear_pane`, `pipe_pane`, `enter_copy_mode`, `exit_copy_mode`, `respawn_pane`, `kill_pane` | diff --git a/docs/conf.py b/docs/conf.py index e0fd0530..af8bbd12 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -118,6 +118,7 @@ def _patched_tool_collector_tool(self: ToolCollector, **kwargs: t.Any) -> t.Any: conf["myst_enable_extensions"] = [*conf["myst_enable_extensions"], "attrs_inline"] conf["fastmcp_tool_modules"] = [ + "libtmux_mcp.tools.batch_tools", "libtmux_mcp.tools.server_tools", "libtmux_mcp.tools.session_tools", "libtmux_mcp.tools.window_tools", @@ -129,6 +130,7 @@ def _patched_tool_collector_tool(self: ToolCollector, **kwargs: t.Any) -> t.Any: "libtmux_mcp.tools.hook_tools", ] conf["fastmcp_area_map"] = { + "batch_tools": "batch/index", "server_tools": "server/index", "session_tools": "session/index", "window_tools": "window/index", @@ -159,6 +161,9 @@ def _patched_tool_collector_tool(self: ToolCollector, **kwargs: t.Any) -> t.Any: "SendKeysOperation", "SendKeysOperationResult", "SendKeysBatchResult", + "ToolCallOperation", + "ToolCallOperationResult", + "ToolCallBatchResult", "HookEntry", "HookListResult", "BufferRef", diff --git a/docs/reference/api/tools.md b/docs/reference/api/tools.md index 0192abbb..fddc84e3 100644 --- a/docs/reference/api/tools.md +++ b/docs/reference/api/tools.md @@ -1,5 +1,14 @@ # Tools +## Batch tools + +```{eval-rst} +.. automodule:: libtmux_mcp.tools.batch_tools + :members: + :undoc-members: + :show-inheritance: +``` + ## Server tools ```{eval-rst} diff --git a/docs/tools/batch/call-destructive-tools-batch.md b/docs/tools/batch/call-destructive-tools-batch.md new file mode 100644 index 00000000..aad382c8 --- /dev/null +++ b/docs/tools/batch/call-destructive-tools-batch.md @@ -0,0 +1,32 @@ +# Call destructive tools batch + +```{fastmcp-tool} batch_tools.call_destructive_tools_batch +``` + +**Use when** a reviewed workflow intentionally includes destructive +tools and should still return one per-operation result envelope. + +**Avoid when** the workflow can fit inside +{tooliconl}`call-mutating-tools-batch`. This wrapper can invoke +destructive nested tools when the server safety tier permits them. + +**Side effects:** Runs readonly, mutating, and destructive nested tools +in order. Recursive batch calls are rejected. + +**Example:** + +```json +{ + "tool": "call_destructive_tools_batch", + "arguments": { + "operations": [ + {"tool": "kill_pane", "arguments": {"pane_id": "%7"}}, + {"tool": "list_panes", "arguments": {"window_id": "@3"}} + ], + "on_error": "stop" + } +} +``` + +```{fastmcp-tool-input} batch_tools.call_destructive_tools_batch +``` diff --git a/docs/tools/batch/call-mutating-tools-batch.md b/docs/tools/batch/call-mutating-tools-batch.md new file mode 100644 index 00000000..ceb34bc5 --- /dev/null +++ b/docs/tools/batch/call-mutating-tools-batch.md @@ -0,0 +1,41 @@ +# Call mutating tools batch + +```{fastmcp-tool} batch_tools.call_mutating_tools_batch +``` + +**Use when** you need an ordered workflow made from existing typed MCP +tools, such as renaming and splitting a known window, while preserving +each tool's own schema and safety checks. + +**Avoid when** you need tmux's native semicolon command parsing. This +tool batches MCP tools; it does not create one tmux command sequence. +For shell commands with completion and output, prefer +{tooliconl}`run-command`. + +**Side effects:** Runs readonly and mutating nested tools in order. +Destructive nested tools are rejected even when the server process is +running with `LIBTMUX_SAFETY=destructive`. + +**Example:** + +```json +{ + "tool": "call_mutating_tools_batch", + "arguments": { + "operations": [ + { + "tool": "rename_window", + "arguments": {"window_id": "@2", "new_name": "logs"} + }, + { + "tool": "split_window", + "arguments": {"window_id": "@2", "direction": "right"} + } + ], + "on_error": "stop" + } +} +``` + +```{fastmcp-tool-input} batch_tools.call_mutating_tools_batch +``` diff --git a/docs/tools/batch/call-readonly-tools-batch.md b/docs/tools/batch/call-readonly-tools-batch.md new file mode 100644 index 00000000..8d32190c --- /dev/null +++ b/docs/tools/batch/call-readonly-tools-batch.md @@ -0,0 +1,34 @@ +# Call readonly tools batch + +```{fastmcp-tool} batch_tools.call_readonly_tools_batch +``` + +**Use when** you need several read-only observations in one ordered +MCP turn, such as listing sessions and then reading server metadata. + +**Avoid when** any nested operation changes tmux state — use +{tooliconl}`call-mutating-tools-batch` for readonly + mutating +workflows, or call the individual tools when each result should be +reviewed before choosing the next action. + +**Side effects:** None beyond the nested readonly tools. Mutating and +destructive nested tools are rejected even when the server process is +running with a higher safety tier. + +**Example:** + +```json +{ + "tool": "call_readonly_tools_batch", + "arguments": { + "operations": [ + {"tool": "list_sessions", "arguments": {}}, + {"tool": "get_server_info", "arguments": {}} + ], + "on_error": "stop" + } +} +``` + +```{fastmcp-tool-input} batch_tools.call_readonly_tools_batch +``` diff --git a/docs/tools/batch/index.md b/docs/tools/batch/index.md new file mode 100644 index 00000000..be5684cf --- /dev/null +++ b/docs/tools/batch/index.md @@ -0,0 +1,31 @@ +# Batch tools + +Batch tools coordinate existing MCP tool calls. They do not replace tmux +targeting: each nested tool call still supplies its own arguments, +including `socket_name` when needed. + +::::{grid} 1 1 2 3 +:gutter: 2 2 3 3 + +:::{grid-item-card} {tooliconl}`call-readonly-tools-batch` +Call readonly tools in order. +::: + +:::{grid-item-card} {tooliconl}`call-mutating-tools-batch` +Call readonly or mutating tools in order. +::: + +:::{grid-item-card} {tooliconl}`call-destructive-tools-batch` +Call readonly, mutating, or destructive tools in order. +::: + +:::: + +```{toctree} +:hidden: +:maxdepth: 1 + +call-readonly-tools-batch +call-mutating-tools-batch +call-destructive-tools-batch +``` diff --git a/docs/tools/index.md b/docs/tools/index.md index 141d4d8d..7050fd8b 100644 --- a/docs/tools/index.md +++ b/docs/tools/index.md @@ -2,7 +2,11 @@ # Tools -All tools accept an optional `socket_name` parameter for multi-server support. It defaults to the {envvar}`LIBTMUX_SOCKET` env var. See {ref}`configuration`. +Targeted tmux tools accept an optional `socket_name` parameter for +multi-server support. It defaults to the {envvar}`LIBTMUX_SOCKET` env +var. {toolref}`list-servers` discovers sockets itself, and batch tools +leave socket selection inside each nested tool's arguments. See +{ref}`configuration`. ## Which tool do I want? @@ -50,6 +54,11 @@ All tools accept an optional `socket_name` parameter for multi-server support. I - Block until signalled → {tool}`wait-for-channel` - Signal a waiter → {tool}`signal-channel` +**Batching typed tool calls?** +- Read-only observations → {tool}`call-readonly-tools-batch` +- Ordered readonly + mutating workflows → {tool}`call-mutating-tools-batch` +- Reviewed workflows that include destructive steps → {tool}`call-destructive-tools-batch` + **Staging multi-line input?** - Stage content → {tool}`load-buffer` - Push into pane → {tool}`paste-buffer` @@ -138,6 +147,12 @@ Wait for text to appear in a pane. Get tmux server info. ::: +:::{grid-item-card} call_readonly_tools_batch +:link: call-readonly-tools-batch +:link-type: ref +Call typed readonly tools in order. +::: + :::{grid-item-card} list_servers :link: list-servers :link-type: ref @@ -237,6 +252,12 @@ Send several ordered raw-input operations. Run a shell command and report exit status. ::: +:::{grid-item-card} call_mutating_tools_batch +:link: call-mutating-tools-batch +:link-type: ref +Call typed readonly or mutating tools in order. +::: + :::{grid-item-card} rename_session :link: rename-session :link-type: ref @@ -396,6 +417,12 @@ Destroy a pane. Kill the entire tmux server. ::: +:::{grid-item-card} call_destructive_tools_batch +:link: call-destructive-tools-batch +:link-type: ref +Call typed tools including destructive steps. +::: + :::{grid-item-card} delete_buffer :link: delete-buffer :link-type: ref @@ -409,6 +436,7 @@ Delete an MCP-staged tmux paste buffer. :caption: Tools by tmux scope server/index +batch/index session/index window/index pane/index diff --git a/docs/topics/architecture.md b/docs/topics/architecture.md index 2ac8f7ee..a630b00b 100644 --- a/docs/topics/architecture.md +++ b/docs/topics/architecture.md @@ -15,6 +15,7 @@ src/libtmux_mcp/ models.py # Pydantic output models middleware.py # Safety, audit, retry, and error-result middleware tools/ + batch_tools.py # call_readonly_tools_batch, call_mutating_tools_batch, call_destructive_tools_batch server_tools.py # list_servers, list_sessions, create_session, kill_server, get_server_info session_tools.py # list_windows, create_window, rename_session, kill_session window_tools.py # list_panes, split_window, rename_window, kill_window, select_layout, resize_window diff --git a/docs/topics/safety.md b/docs/topics/safety.md index 8c191379..22052810 100644 --- a/docs/topics/safety.md +++ b/docs/topics/safety.md @@ -8,9 +8,9 @@ libtmux-mcp uses a three-tier safety system to control which tools are available | Tier | Label | Access | Use case | |------|-------|--------|----------| -| `readonly` | {badge}`readonly` | List, capture, search, info | Monitoring, browsing | -| `mutating` (default) | {badge}`mutating` | + create, send_keys, send_keys_batch, rename, resize | Normal agent workflow | -| `destructive` | {badge}`destructive` | + kill_server, kill_session, kill_window, kill_pane | Full control | +| `readonly` | {badge}`readonly` | List, capture, search, info, readonly batches | Monitoring, browsing | +| `mutating` (default) | {badge}`mutating` | + create, send_keys, send_keys_batch, mutating batches, rename, resize | Normal agent workflow | +| `destructive` | {badge}`destructive` | + destructive batches, kill_server, kill_session, kill_window, kill_pane | Full control | ## Configuration diff --git a/src/libtmux_mcp/middleware.py b/src/libtmux_mcp/middleware.py index c5635abf..ddb99b31 100644 --- a/src/libtmux_mcp/middleware.py +++ b/src/libtmux_mcp/middleware.py @@ -456,9 +456,9 @@ async def on_call_tool( ) #: Nested argument containers that may contain sensitive argument names. -#: ``operations`` is used by ``send_keys_batch``; preserving pane ids and -#: booleans is useful for audit trails, but each nested ``keys`` payload -#: must be digested the same way top-level ``send_keys(keys=...)`` is. +#: ``operations`` is used by ``send_keys_batch`` and the generic tool-batch +#: wrappers. Preserving routing metadata is useful for audit trails, but +#: nested payloads must be digested the same way top-level tool calls are. _NESTED_ARG_LIST_NAMES: frozenset[str] = frozenset({"operations"}) _NONE_TYPE = type(None) @@ -517,6 +517,26 @@ def _summarize_send_keys_operation_args(args: dict[str, t.Any]) -> dict[str, t.A return summary +def _summarize_tool_batch_operation_args(args: dict[str, t.Any]) -> dict[str, t.Any]: + """Summarize one generic tool-batch operation for audit logging.""" + summary: dict[str, t.Any] = {} + for key, value in args.items(): + if key == "tool" and isinstance(value, str): + summary[key] = value + elif key == "arguments" and isinstance(value, dict): + summary[key] = _summarize_args(value) + else: + summary[key] = _redacted_value_shape(value) + return summary + + +def _summarize_nested_operation_args(args: dict[str, t.Any]) -> dict[str, t.Any]: + """Summarize a known nested operation shape.""" + if "tool" in args or "arguments" in args: + return _summarize_tool_batch_operation_args(args) + return _summarize_send_keys_operation_args(args) + + def _summarize_args(args: dict[str, t.Any]) -> dict[str, t.Any]: """Summarize tool arguments for audit logging. @@ -558,7 +578,7 @@ def _summarize_args(args: dict[str, t.Any]) -> dict[str, t.Any]: elif key in _NESTED_ARG_LIST_NAMES: if isinstance(value, list): summary[key] = [ - _summarize_send_keys_operation_args(item) + _summarize_nested_operation_args(item) if isinstance(item, dict) else _redacted_value_shape(item) for item in value diff --git a/src/libtmux_mcp/models.py b/src/libtmux_mcp/models.py index a568ad90..fc97f71d 100644 --- a/src/libtmux_mcp/models.py +++ b/src/libtmux_mcp/models.py @@ -375,6 +375,72 @@ class SendKeysBatchResult(BaseModel): ) +class ToolCallOperation(BaseModel): + """One nested MCP tool call for a batch wrapper.""" + + model_config = ConfigDict(extra="forbid") + + tool: str = Field(description="Registered MCP tool name to call.") + arguments: dict[str, t.Any] = Field( + default_factory=dict, + description="Arguments for the nested tool call.", + ) + + +class ToolCallOperationResult(BaseModel): + """Per-operation result from a generic MCP tool batch.""" + + index: int = Field(description="Zero-based index in the submitted operation list.") + tool: str = Field(description="Nested tool name that was attempted.") + success: bool = Field(description="True when this nested tool call succeeded.") + error: str | None = Field( + default=None, + description="Error message for this operation, if it failed.", + ) + content: list[dict[str, t.Any]] = Field( + default_factory=list, + description="MCP content blocks returned by the nested tool.", + ) + structured_content: dict[str, t.Any] | None = Field( + default=None, + description="Structured content returned by the nested tool, if any.", + ) + meta: dict[str, t.Any] | None = Field( + default=None, + description="Runtime metadata returned by the nested tool, if any.", + ) + elapsed_seconds: float = Field(description="Time spent on this operation.") + + +class ToolCallBatchResult(BaseModel): + """Structured result for a serial batch of MCP tool calls.""" + + results: list[ToolCallOperationResult] = Field( + default_factory=list, + description="Per-operation results in attempted order.", + ) + succeeded: int = Field(description="Number of nested tool calls that succeeded.") + failed: int = Field(description="Number of nested tool calls that failed.") + stopped_at: int | None = Field( + default=None, + description=( + "Index where processing stopped because on_error='stop', or None " + "when all operations were attempted." + ), + ) + response_truncated: bool = Field( + default=False, + description=( + "True when nested result payloads were elided to keep the batch " + "response under the server response cap." + ), + ) + response_truncated_bytes: int = Field( + default=0, + description="Approximate serialized bytes removed from nested result payloads.", + ) + + class PaneSnapshot(BaseModel): """Rich screen capture with metadata: content, cursor, mode, and scroll state.""" diff --git a/src/libtmux_mcp/tools/__init__.py b/src/libtmux_mcp/tools/__init__.py index a11a4e93..7a72f9ab 100644 --- a/src/libtmux_mcp/tools/__init__.py +++ b/src/libtmux_mcp/tools/__init__.py @@ -11,6 +11,7 @@ def register_tools(mcp: FastMCP) -> None: """Register all tool modules with the FastMCP instance.""" from libtmux_mcp.tools import ( + batch_tools, buffer_tools, env_tools, hook_tools, @@ -22,6 +23,7 @@ def register_tools(mcp: FastMCP) -> None: window_tools, ) + batch_tools.register(mcp) server_tools.register(mcp) session_tools.register(mcp) window_tools.register(mcp) diff --git a/src/libtmux_mcp/tools/batch_tools.py b/src/libtmux_mcp/tools/batch_tools.py new file mode 100644 index 00000000..65d760bd --- /dev/null +++ b/src/libtmux_mcp/tools/batch_tools.py @@ -0,0 +1,365 @@ +"""Generic MCP tool batching helpers.""" + +from __future__ import annotations + +import json +import time +import typing as t + +from fastmcp import Context +from fastmcp.tools.base import ToolResult +from pydantic import BaseModel + +from libtmux_mcp._utils import ( + ANNOTATIONS_RO, + TAG_DESTRUCTIVE, + TAG_MUTATING, + TAG_READONLY, + ExpectedToolError, + handle_tool_errors_async, +) +from libtmux_mcp.middleware import DEFAULT_RESPONSE_LIMIT_BYTES +from libtmux_mcp.models import ( + ToolCallBatchResult, + ToolCallOperation, + ToolCallOperationResult, +) + +if t.TYPE_CHECKING: + from fastmcp import FastMCP + +_OnError: t.TypeAlias = t.Literal["stop", "continue"] + +_TIER_LEVELS: dict[str, int] = { + TAG_READONLY: 0, + TAG_MUTATING: 1, + TAG_DESTRUCTIVE: 2, +} + +_BATCH_TOOL_NAMES: frozenset[str] = frozenset( + { + "call_readonly_tools_batch", + "call_mutating_tools_batch", + "call_destructive_tools_batch", + } +) + +MAX_BATCH_OPERATIONS = 1_000 + +_BATCH_TRUNCATED_CONTENT: list[dict[str, t.Any]] = [ + { + "type": "text", + "text": "[... batch truncated nested content ...]", + } +] + +_ANNOTATIONS_BATCH_SIDE_EFFECTS: dict[str, bool] = { + "readOnlyHint": False, + "destructiveHint": True, + "idempotentHint": False, + "openWorldHint": True, +} + + +def _content_block_to_dict(block: t.Any) -> dict[str, t.Any]: + """Return a JSON-ready representation of an MCP content block.""" + if isinstance(block, BaseModel): + return block.model_dump(mode="json", by_alias=True, exclude_none=True) + if hasattr(block, "model_dump"): + dumped = block.model_dump(mode="json", by_alias=True, exclude_none=True) + if isinstance(dumped, dict): + return t.cast("dict[str, t.Any]", dumped) + return {"type": type(block).__name__, "value": str(block)} + + +def _result_error_text(result: ToolResult) -> str | None: + """Extract a readable error string from a FastMCP ``ToolResult``.""" + text_blocks: list[str] = [] + for block in result.content: + text = getattr(block, "text", None) + if isinstance(text, str): + text_blocks.append(text) + if text_blocks: + return "\n".join(text_blocks) + if result.is_error: + return "Tool call returned an error result." + return None + + +def _tool_tier(tool_name: str, tags: set[str]) -> str: + """Return the highest recognized safety tier for a registered tool.""" + found = [tier for tier in _TIER_LEVELS if tier in tags] + if not found: + msg = f"Tool {tool_name!r} has no recognized safety tier tag." + raise ExpectedToolError(msg) + return max(found, key=lambda tier: _TIER_LEVELS[tier]) + + +def _check_operation_allowed( + *, + tool_name: str, + tool_tier: str, + max_tier: str, +) -> None: + """Raise when a nested tool exceeds this batch wrapper's tier.""" + if _TIER_LEVELS[tool_tier] <= _TIER_LEVELS[max_tier]: + return + msg = ( + f"Tool {tool_name!r} has tier {tool_tier!r}, which exceeds " + f"batch tier {max_tier}." + ) + raise ExpectedToolError(msg) + + +async def _get_allowed_tool_tier( + *, + fastmcp: FastMCP, + operation: ToolCallOperation, + max_tier: str, +) -> None: + """Validate that one nested operation targets an allowed tool.""" + if operation.tool in _BATCH_TOOL_NAMES: + msg = "Batch tools cannot call batch tools recursively." + raise ExpectedToolError(msg) + + tool = await fastmcp.get_tool(operation.tool) + if tool is None: + msg = f"Unknown tool: {operation.tool!r}" + raise ExpectedToolError(msg) + + tool_tier = _tool_tier(operation.tool, tool.tags) + _check_operation_allowed( + tool_name=operation.tool, + tool_tier=tool_tier, + max_tier=max_tier, + ) + + +def _ensure_tool_result(tool_name: str, result: t.Any) -> ToolResult: + """Return ``result`` as a ``ToolResult`` or raise a row-level error.""" + if isinstance(result, ToolResult): + return result + msg = f"Tool {tool_name!r} returned an unsupported result." + raise ExpectedToolError(msg) + + +def _batch_response_size(result: ToolCallBatchResult) -> int: + """Return the serialized byte size of FastMCP's batch response envelope.""" + result_json = result.model_dump_json(fallback=str) + envelope = { + "content": [{"type": "text", "text": result_json}], + "structuredContent": json.loads(result_json), + "isError": False, + } + serialized = json.dumps(envelope, separators=(",", ":"), sort_keys=True) + return len(serialized.encode("utf-8")) + + +def _operation_has_nested_payload(result: ToolCallOperationResult) -> bool: + """Return True when a row still carries payload fields that can be elided.""" + return bool(result.content) or result.structured_content is not None + + +def _limit_batch_result( + result: ToolCallBatchResult, + *, + max_bytes: int = DEFAULT_RESPONSE_LIMIT_BYTES, +) -> ToolCallBatchResult: + """Elide nested result payloads until the batch envelope fits.""" + if _batch_response_size(result) <= max_bytes: + return result + + limited = result.model_copy( + deep=True, + update={"response_truncated": True}, + ) + dropped_bytes = 0 + for operation in limited.results: + if not _operation_has_nested_payload(operation): + continue + + before = _batch_response_size(limited) + operation.content = [item.copy() for item in _BATCH_TRUNCATED_CONTENT] + operation.structured_content = None + after = _batch_response_size(limited) + dropped_bytes += max(before - after, 0) + limited.response_truncated_bytes = dropped_bytes + + if _batch_response_size(limited) <= max_bytes: + break + + return limited + + +async def _call_one_tool( + *, + fastmcp: FastMCP, + operation: ToolCallOperation, + index: int, + max_tier: str, +) -> ToolCallOperationResult: + """Call one nested tool and convert its outcome to a batch result row.""" + start = time.monotonic() + try: + await _get_allowed_tool_tier( + fastmcp=fastmcp, + operation=operation, + max_tier=max_tier, + ) + + result = _ensure_tool_result( + operation.tool, + await fastmcp.call_tool( + operation.tool, + operation.arguments, + run_middleware=True, + ), + ) + + error = _result_error_text(result) + return ToolCallOperationResult( + index=index, + tool=operation.tool, + success=not result.is_error, + error=error if result.is_error else None, + content=[_content_block_to_dict(block) for block in result.content], + structured_content=result.structured_content, + meta=result.meta, + elapsed_seconds=time.monotonic() - start, + ) + except Exception as exc: + return ToolCallOperationResult( + index=index, + tool=operation.tool, + success=False, + error=str(exc), + elapsed_seconds=time.monotonic() - start, + ) + + +async def _call_tools_batch( + *, + operations: list[ToolCallOperation], + on_error: _OnError, + max_tier: str, + ctx: Context | None, +) -> ToolCallBatchResult: + """Execute nested MCP tool calls serially through FastMCP.""" + if not operations: + msg = "operations must contain at least one tool call" + raise ExpectedToolError(msg) + if len(operations) > MAX_BATCH_OPERATIONS: + msg = f"operations must contain at most {MAX_BATCH_OPERATIONS} tool calls" + raise ExpectedToolError(msg) + if on_error not in {"stop", "continue"}: + msg = "on_error must be 'stop' or 'continue'" + raise ExpectedToolError(msg) + if ctx is None: + msg = "FastMCP context is required; call this tool through MCP." + raise ExpectedToolError(msg) + + results: list[ToolCallOperationResult] = [] + stopped_at: int | None = None + for index, operation in enumerate(operations): + result = await _call_one_tool( + fastmcp=ctx.fastmcp, + operation=operation, + index=index, + max_tier=max_tier, + ) + results.append(result) + if not result.success and on_error == "stop": + stopped_at = index + break + + succeeded = sum(1 for result in results if result.success) + failed = len(results) - succeeded + return _limit_batch_result( + ToolCallBatchResult( + results=results, + succeeded=succeeded, + failed=failed, + stopped_at=stopped_at, + ) + ) + + +@handle_tool_errors_async +async def call_readonly_tools_batch( + operations: list[ToolCallOperation], + on_error: _OnError = "stop", + ctx: Context | None = None, +) -> ToolCallBatchResult: + """Call readonly MCP tools serially and return per-tool results. + + Use when several read-only observations should be made in one agent + turn. Each nested call still goes through FastMCP validation, + middleware, and safety checks. Mutating and destructive tools are + rejected even if the server process itself is running at a higher + safety tier. + """ + return await _call_tools_batch( + operations=operations, + on_error=on_error, + max_tier=TAG_READONLY, + ctx=ctx, + ) + + +@handle_tool_errors_async +async def call_mutating_tools_batch( + operations: list[ToolCallOperation], + on_error: _OnError = "stop", + ctx: Context | None = None, +) -> ToolCallBatchResult: + """Call readonly or mutating MCP tools serially and return per-tool results. + + Use for ordered tmux workflows where every step is still an existing + typed MCP tool. Destructive tools are rejected regardless of the + process-wide safety tier. + """ + return await _call_tools_batch( + operations=operations, + on_error=on_error, + max_tier=TAG_MUTATING, + ctx=ctx, + ) + + +@handle_tool_errors_async +async def call_destructive_tools_batch( + operations: list[ToolCallOperation], + on_error: _OnError = "stop", + ctx: Context | None = None, +) -> ToolCallBatchResult: + """Call readonly, mutating, or destructive MCP tools serially. + + This wrapper preserves the normal per-tool schemas and middleware + but its tier permits destructive nested operations. Prefer the + narrower readonly or mutating wrappers whenever possible. + """ + return await _call_tools_batch( + operations=operations, + on_error=on_error, + max_tier=TAG_DESTRUCTIVE, + ctx=ctx, + ) + + +def register(mcp: FastMCP) -> None: + """Register generic MCP batch tools.""" + mcp.tool( + title="Call Readonly Tools Batch", + annotations=ANNOTATIONS_RO, + tags={TAG_READONLY}, + )(call_readonly_tools_batch) + mcp.tool( + title="Call Mutating Tools Batch", + annotations=_ANNOTATIONS_BATCH_SIDE_EFFECTS, + tags={TAG_MUTATING}, + )(call_mutating_tools_batch) + mcp.tool( + title="Call Destructive Tools Batch", + annotations=_ANNOTATIONS_BATCH_SIDE_EFFECTS, + tags={TAG_DESTRUCTIVE}, + )(call_destructive_tools_batch) diff --git a/src/libtmux_mcp/tools/server_tools.py b/src/libtmux_mcp/tools/server_tools.py index 55a5e772..9b0b9a2f 100644 --- a/src/libtmux_mcp/tools/server_tools.py +++ b/src/libtmux_mcp/tools/server_tools.py @@ -263,13 +263,21 @@ def _probe_server_by_path(socket_path: pathlib.Path) -> ServerInfo | None: #: Tools that intentionally do NOT accept ``socket_name`` because they -#: discover or enumerate sockets themselves rather than connecting to a -#: known one. Read by ``test_registered_tools_accept_socket_name`` to -#: enforce the agent-facing contract advertised in +#: either discover sockets themselves or coordinate nested tools whose +#: arguments carry their own targeting. Read by +#: ``test_registered_tools_accept_socket_name`` to enforce the +#: agent-facing contract advertised in #: :data:`libtmux_mcp.server._BASE_INSTRUCTIONS`. When you add a new #: discovery-style tool, append it here AND update the prose in #: ``_BASE_INSTRUCTIONS`` so the two stay in lockstep. -SOCKET_NAME_EXEMPT: frozenset[str] = frozenset({"list_servers"}) +SOCKET_NAME_EXEMPT: frozenset[str] = frozenset( + { + "call_destructive_tools_batch", + "call_mutating_tools_batch", + "call_readonly_tools_batch", + "list_servers", + } +) @handle_tool_errors diff --git a/tests/test_batch_tools.py b/tests/test_batch_tools.py new file mode 100644 index 00000000..96d31154 --- /dev/null +++ b/tests/test_batch_tools.py @@ -0,0 +1,457 @@ +"""Tests for generic MCP tool batching.""" + +from __future__ import annotations + +import asyncio +import json +import typing as t + +import pytest + +from libtmux_mcp._utils import ( + ANNOTATIONS_DESTRUCTIVE, + ANNOTATIONS_MUTATING, + ANNOTATIONS_RO, + TAG_DESTRUCTIVE, + TAG_MUTATING, + TAG_READONLY, +) + +if t.TYPE_CHECKING: + from fastmcp import FastMCP + + +class BatchResponseLimitFixture(t.NamedTuple): + """Test fixture for aggregate batch response limiting.""" + + test_id: str + payload_size: int + + +BATCH_RESPONSE_LIMIT_FIXTURES: list[BatchResponseLimitFixture] = [ + BatchResponseLimitFixture( + test_id="two_large_readonly_results", + payload_size=300_000, + ), +] + + +class BatchOperationLimitFixture(t.NamedTuple): + """Test fixture for operation-count batch limiting.""" + + test_id: str + operation_count: int + + +BATCH_OPERATION_LIMIT_FIXTURES: list[BatchOperationLimitFixture] = [ + BatchOperationLimitFixture( + test_id="many_missing_tools", + operation_count=6_000, + ), +] + + +class BatchAnnotationFixture(t.NamedTuple): + """Test fixture for generic batch wrapper annotations.""" + + test_id: str + tool_name: str + read_only_hint: bool + destructive_hint: bool + idempotent_hint: bool + open_world_hint: bool + + +BATCH_ANNOTATION_FIXTURES: list[BatchAnnotationFixture] = [ + BatchAnnotationFixture( + test_id="mutating_batch_warns_destructive_open_world", + tool_name="call_mutating_tools_batch", + read_only_hint=False, + destructive_hint=True, + idempotent_hint=False, + open_world_hint=True, + ), + BatchAnnotationFixture( + test_id="destructive_batch_warns_destructive_open_world", + tool_name="call_destructive_tools_batch", + read_only_hint=False, + destructive_hint=True, + idempotent_hint=False, + open_world_hint=True, + ), +] + + +def _content_block_to_wire(block: t.Any) -> dict[str, t.Any]: + if hasattr(block, "model_dump"): + dumped = block.model_dump(mode="json", by_alias=True, exclude_none=True) + if isinstance(dumped, dict): + return t.cast("dict[str, t.Any]", dumped) + return {"type": type(block).__name__, "value": str(block)} + + +def _call_tool_result_wire(result: t.Any) -> dict[str, t.Any]: + return { + "content": [_content_block_to_wire(block) for block in result.content], + "structuredContent": result.structured_content, + "isError": result.is_error, + } + + +def _batch_probe_server() -> FastMCP: + """Build a small FastMCP server with batch tools and tiered probes.""" + from fastmcp import FastMCP + + from libtmux_mcp.middleware import SafetyMiddleware, ToolErrorResultMiddleware + from libtmux_mcp.tools.batch_tools import register as register_batch_tools + + mcp = FastMCP( + name="batch-probe", + middleware=[ + ToolErrorResultMiddleware(transform_errors=True), + SafetyMiddleware(max_tier=TAG_DESTRUCTIVE), + ], + ) + register_batch_tools(mcp) + + @mcp.tool(title="Readonly Probe", annotations=ANNOTATIONS_RO, tags={TAG_READONLY}) + def readonly_probe(value: str) -> dict[str, str]: + return {"value": value} + + @mcp.tool( + title="Mutating Probe", + annotations=ANNOTATIONS_MUTATING, + tags={TAG_MUTATING}, + ) + def mutating_probe(value: str) -> dict[str, str]: + return {"value": value} + + @mcp.tool( + title="Destructive Probe", + annotations=ANNOTATIONS_DESTRUCTIVE, + tags={TAG_DESTRUCTIVE}, + ) + def destructive_probe(value: str) -> dict[str, str]: + return {"value": value} + + return mcp + + +def test_call_readonly_tools_batch_preserves_structured_results() -> None: + """The readonly batch wrapper returns per-tool structured content.""" + from fastmcp import Client + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_readonly_tools_batch", + { + "operations": [ + { + "tool": "readonly_probe", + "arguments": {"value": "alpha"}, + }, + { + "tool": "readonly_probe", + "arguments": {"value": "beta"}, + }, + ], + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + + assert result.is_error is False + assert result.structured_content["succeeded"] == 2 + assert result.structured_content["failed"] == 0 + assert result.structured_content["stopped_at"] is None + first, second = result.structured_content["results"] + assert first == { + "index": 0, + "tool": "readonly_probe", + "success": True, + "error": None, + "content": [{"type": "text", "text": '{"value":"alpha"}'}], + "structured_content": {"value": "alpha"}, + "meta": None, + "elapsed_seconds": first["elapsed_seconds"], + } + assert second == { + "index": 1, + "tool": "readonly_probe", + "success": True, + "error": None, + "content": [{"type": "text", "text": '{"value":"beta"}'}], + "structured_content": {"value": "beta"}, + "meta": None, + "elapsed_seconds": second["elapsed_seconds"], + } + assert first["elapsed_seconds"] >= 0.0 + assert second["elapsed_seconds"] >= 0.0 + + +@pytest.mark.parametrize( + BatchResponseLimitFixture._fields, + BATCH_RESPONSE_LIMIT_FIXTURES, + ids=[fixture.test_id for fixture in BATCH_RESPONSE_LIMIT_FIXTURES], +) +def test_call_readonly_tools_batch_caps_aggregate_response( + test_id: str, + payload_size: int, +) -> None: + """The batch envelope survives when nested result payloads are capped.""" + from fastmcp import Client + + from libtmux_mcp.middleware import DEFAULT_RESPONSE_LIMIT_BYTES + + first_payload = "first-" + ("a" * payload_size) + second_payload = "second-" + ("b" * payload_size) + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_readonly_tools_batch", + { + "operations": [ + { + "tool": "readonly_probe", + "arguments": {"value": first_payload}, + }, + { + "tool": "readonly_probe", + "arguments": {"value": second_payload}, + }, + ], + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + + assert result.is_error is False + structured = result.structured_content + assert structured["response_truncated"] is True + assert structured["response_truncated_bytes"] > 0 + assert structured["succeeded"] == 2 + assert structured["failed"] == 0 + assert structured["stopped_at"] is None + + serialized = json.dumps( + _call_tool_result_wire(result), + separators=(",", ":"), + sort_keys=True, + ) + assert len(serialized.encode("utf-8")) <= DEFAULT_RESPONSE_LIMIT_BYTES + assert first_payload not in serialized + assert second_payload not in serialized + + first, second = structured["results"] + assert first["index"] == 0 + assert first["tool"] == "readonly_probe" + assert first["success"] is True + assert first["structured_content"] is None + assert first["content"] == [ + { + "type": "text", + "text": "[... batch truncated nested content ...]", + } + ] + assert second["structured_content"] is None + assert second["content"] == [ + { + "type": "text", + "text": "[... batch truncated nested content ...]", + } + ] + + +@pytest.mark.parametrize( + BatchOperationLimitFixture._fields, + BATCH_OPERATION_LIMIT_FIXTURES, + ids=[fixture.test_id for fixture in BATCH_OPERATION_LIMIT_FIXTURES], +) +def test_call_readonly_tools_batch_rejects_oversized_operation_count( + test_id: str, + operation_count: int, +) -> None: + """The batch wrapper rejects requests whose rows alone can exceed the cap.""" + from fastmcp import Client + + from libtmux_mcp.middleware import DEFAULT_RESPONSE_LIMIT_BYTES + + assert test_id + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_readonly_tools_batch", + { + "operations": [ + { + "tool": "missing_probe", + "arguments": {}, + } + for _ in range(operation_count) + ], + "on_error": "continue", + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + serialized = json.dumps( + _call_tool_result_wire(result), + separators=(",", ":"), + sort_keys=True, + ) + + assert len(serialized.encode("utf-8")) <= DEFAULT_RESPONSE_LIMIT_BYTES + assert result.is_error is True + assert result.structured_content is None + assert "operations must contain at most" in serialized + + +def test_call_readonly_tools_batch_rejects_mutating_inner_tool() -> None: + """Readonly batching does not tunnel a mutating tool call.""" + from fastmcp import Client + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_readonly_tools_batch", + { + "operations": [ + { + "tool": "mutating_probe", + "arguments": {"value": "changed"}, + } + ], + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + + assert result.is_error is False + assert result.structured_content["succeeded"] == 0 + assert result.structured_content["failed"] == 1 + assert result.structured_content["stopped_at"] == 0 + [operation] = result.structured_content["results"] + assert operation["success"] is False + assert "exceeds batch tier readonly" in operation["error"] + + +def test_call_mutating_tools_batch_rejects_destructive_inner_tool() -> None: + """Mutating batching does not tunnel a destructive tool call.""" + from fastmcp import Client + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_mutating_tools_batch", + { + "operations": [ + { + "tool": "destructive_probe", + "arguments": {"value": "destroy"}, + } + ], + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + + assert result.is_error is False + [operation] = result.structured_content["results"] + assert operation["success"] is False + assert "exceeds batch tier mutating" in operation["error"] + + +def test_call_mutating_tools_batch_continues_after_error() -> None: + """Continue mode attempts later operations after a failed tool call.""" + from fastmcp import Client + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_mutating_tools_batch", + { + "on_error": "continue", + "operations": [ + { + "tool": "missing_probe", + "arguments": {}, + }, + { + "tool": "mutating_probe", + "arguments": {"value": "kept-going"}, + }, + ], + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + + assert result.is_error is False + assert result.structured_content["succeeded"] == 1 + assert result.structured_content["failed"] == 1 + assert result.structured_content["stopped_at"] is None + first, second = result.structured_content["results"] + assert first["success"] is False + assert second["success"] is True + assert second["structured_content"] == {"value": "kept-going"} + + +def test_call_tools_batch_rejects_self_invocation() -> None: + """Batch wrappers cannot recursively call batch wrappers.""" + from fastmcp import Client + + async def _call() -> t.Any: + async with Client(_batch_probe_server()) as client: + return await client.call_tool( + "call_destructive_tools_batch", + { + "operations": [ + { + "tool": "call_destructive_tools_batch", + "arguments": {"operations": []}, + } + ], + }, + raise_on_error=False, + ) + + result = asyncio.run(_call()) + + assert result.is_error is False + [operation] = result.structured_content["results"] + assert operation["success"] is False + assert "cannot call batch tools recursively" in operation["error"] + + +@pytest.mark.parametrize( + BatchAnnotationFixture._fields, + BATCH_ANNOTATION_FIXTURES, + ids=[fixture.test_id for fixture in BATCH_ANNOTATION_FIXTURES], +) +def test_batch_wrappers_advertise_worst_case_annotations( + test_id: str, + tool_name: str, + read_only_hint: bool, + destructive_hint: bool, + idempotent_hint: bool, + open_world_hint: bool, +) -> None: + """Batch wrappers advertise the strongest hint from their allowed tools.""" + mcp = _batch_probe_server() + + tool = asyncio.run(mcp.get_tool(tool_name)) + assert tool is not None, f"{tool_name} should be registered" + assert tool.annotations is not None, f"{tool_name} should carry annotations" + assert tool.annotations.readOnlyHint is read_only_hint + assert tool.annotations.destructiveHint is destructive_hint + assert tool.annotations.idempotentHint is idempotent_hint + assert tool.annotations.openWorldHint is open_world_hint diff --git a/tests/test_middleware.py b/tests/test_middleware.py index f033851a..698db81d 100644 --- a/tests/test_middleware.py +++ b/tests/test_middleware.py @@ -343,6 +343,41 @@ def test_summarize_args_redacts_send_keys_batch_operations() -> None: assert "sha256_prefix" in operation["keys"] +def test_summarize_args_redacts_nested_tool_batch_arguments() -> None: + """Generic batch operations preserve tool names while digesting payloads.""" + args: dict[str, t.Any] = { + "operations": [ + { + "tool": "send_keys", + "arguments": { + "keys": "psql -U admin -W supersecret mydb", + "pane_id": "%1", + }, + }, + { + "tool": "set_environment", + "arguments": { + "name": "DATABASE_URL", + "value": "postgres://admin:topsecret@db/prod", + }, + }, + ], + } + + summary = _summarize_args(args) + rendered = str(summary) + + assert "supersecret" not in rendered + assert "topsecret" not in rendered + first, second = summary["operations"] + assert first["tool"] == "send_keys" + assert first["arguments"]["pane_id"] == "%1" + assert isinstance(first["arguments"]["keys"], dict) + assert second["tool"] == "set_environment" + assert second["arguments"]["name"] == "DATABASE_URL" + assert isinstance(second["arguments"]["value"], dict) + + @pytest.mark.parametrize( MalformedOperationAuditFixture._fields, MALFORMED_OPERATION_AUDIT_FIXTURES,