From bcc0b82498ea04a25dec3d2a298b6d415fbc1957 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 06:04:04 -0500 Subject: [PATCH 01/10] mcp(feat[batch_tools]): Add tier-aware MCP tool batching why: Enable ordered bulk calls while preserving each nested tool's schemas, middleware, and safety checks. what: - Add readonly, mutating, and destructive batch wrappers with per-operation results - Preserve nested FastMCP content, structured_content, and meta - Extend audit redaction for nested batch arguments - Cover tier enforcement, continuation, recursion rejection, and audit redaction --- src/libtmux_mcp/middleware.py | 28 ++- src/libtmux_mcp/models.py | 55 +++++ src/libtmux_mcp/tools/__init__.py | 2 + src/libtmux_mcp/tools/batch_tools.py | 296 ++++++++++++++++++++++++++ src/libtmux_mcp/tools/server_tools.py | 16 +- tests/test_batch_tools.py | 231 ++++++++++++++++++++ tests/test_middleware.py | 35 +++ 7 files changed, 655 insertions(+), 8 deletions(-) create mode 100644 src/libtmux_mcp/tools/batch_tools.py create mode 100644 tests/test_batch_tools.py 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..04069eb6 100644 --- a/src/libtmux_mcp/models.py +++ b/src/libtmux_mcp/models.py @@ -375,6 +375,61 @@ 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." + ), + ) + + 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..599b4919 --- /dev/null +++ b/src/libtmux_mcp/tools/batch_tools.py @@ -0,0 +1,296 @@ +"""Generic MCP tool batching helpers.""" + +from __future__ import annotations + +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_DESTRUCTIVE, + ANNOTATIONS_RO, + ANNOTATIONS_SHELL, + TAG_DESTRUCTIVE, + TAG_MUTATING, + TAG_READONLY, + ExpectedToolError, + handle_tool_errors_async, +) +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", + } +) + + +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) + + +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 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 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_SHELL, + tags={TAG_MUTATING}, + )(call_mutating_tools_batch) + mcp.tool( + title="Call Destructive Tools Batch", + annotations=ANNOTATIONS_DESTRUCTIVE, + 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..8fc61bf4 --- /dev/null +++ b/tests/test_batch_tools.py @@ -0,0 +1,231 @@ +"""Tests for generic MCP tool batching.""" + +from __future__ import annotations + +import asyncio +import typing as t + +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 + + +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 + + +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"] 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, From 5465269ce6f11f275762225fc7131adf40f70b7f Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 06:04:22 -0500 Subject: [PATCH 02/10] docs(feat[batch_tools]): Document MCP tool batching why: Keep the published tool catalog and Sphinx FastMCP collector aligned with the new batch tools. what: - Add batch tool reference pages and overview navigation - Register batch tools and models in docs configuration and API reference - Update README, architecture, and safety summaries --- README.md | 1 + docs/conf.py | 5 +++ docs/reference/api/tools.md | 9 ++++ .../batch/call-destructive-tools-batch.md | 32 +++++++++++++++ docs/tools/batch/call-mutating-tools-batch.md | 41 +++++++++++++++++++ docs/tools/batch/call-readonly-tools-batch.md | 34 +++++++++++++++ docs/tools/batch/index.md | 31 ++++++++++++++ docs/tools/index.md | 30 +++++++++++++- docs/topics/architecture.md | 1 + docs/topics/safety.md | 6 +-- 10 files changed, 186 insertions(+), 4 deletions(-) create mode 100644 docs/tools/batch/call-destructive-tools-batch.md create mode 100644 docs/tools/batch/call-mutating-tools-batch.md create mode 100644 docs/tools/batch/call-readonly-tools-batch.md create mode 100644 docs/tools/batch/index.md 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..2a89337f --- /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 creating a window and splitting it, 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": "create_window", + "arguments": {"session_name": "dev", "window_name": "logs"} + }, + { + "tool": "split_window", + "arguments": {"session_name": "dev", "window_name": "logs"} + } + ], + "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 From 6f635d1b9fefd2040a1398794f32494f4bc4a3be Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 06:18:20 -0500 Subject: [PATCH 03/10] mcp(docs[CHANGES]): Note tier-aware MCP tool batching why: The unreleased notes did not yet announce the readonly, mutating, and destructive batch wrappers, which are user-facing. what: - Add a What's new entry for the tier-aware tool batch family - Describe the per-tier safety ceiling and per-operation results --- CHANGES | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGES b/CHANGES index 436ff834..f4c44cf6 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. (#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. From 50935b86c7cb1e00a4383140b92ad2fa7d0b4695 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 06:47:00 -0500 Subject: [PATCH 04/10] mcp(fix[batch_tools]): Cap aggregate batch responses why: Batches of output-heavy read tools could multiply the normal response backstop by embedding each nested result payload in one outer batch envelope. what: - Add batch-level truncation metadata and payload elision - Limit serialized batch envelopes before returning them - Cover oversized readonly batches with a regression test --- src/libtmux_mcp/models.py | 11 ++++ src/libtmux_mcp/tools/batch_tools.py | 61 ++++++++++++++++++-- tests/test_batch_tools.py | 83 ++++++++++++++++++++++++++++ 3 files changed, 150 insertions(+), 5 deletions(-) diff --git a/src/libtmux_mcp/models.py b/src/libtmux_mcp/models.py index 04069eb6..fc97f71d 100644 --- a/src/libtmux_mcp/models.py +++ b/src/libtmux_mcp/models.py @@ -428,6 +428,17 @@ class ToolCallBatchResult(BaseModel): "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): diff --git a/src/libtmux_mcp/tools/batch_tools.py b/src/libtmux_mcp/tools/batch_tools.py index 599b4919..181aace5 100644 --- a/src/libtmux_mcp/tools/batch_tools.py +++ b/src/libtmux_mcp/tools/batch_tools.py @@ -19,6 +19,7 @@ ExpectedToolError, handle_tool_errors_async, ) +from libtmux_mcp.middleware import DEFAULT_RESPONSE_LIMIT_BYTES from libtmux_mcp.models import ( ToolCallBatchResult, ToolCallOperation, @@ -44,6 +45,13 @@ } ) +_BATCH_TRUNCATED_CONTENT: list[dict[str, t.Any]] = [ + { + "type": "text", + "text": "[... batch truncated nested content ...]", + } +] + def _content_block_to_dict(block: t.Any) -> dict[str, t.Any]: """Return a JSON-ready representation of an MCP content block.""" @@ -127,6 +135,47 @@ def _ensure_tool_result(tool_name: str, result: t.Any) -> ToolResult: raise ExpectedToolError(msg) +def _batch_result_size(result: ToolCallBatchResult) -> int: + """Return the serialized byte size of a batch result.""" + return len(result.model_dump_json(fallback=str).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_result_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_result_size(limited) + operation.content = [item.copy() for item in _BATCH_TRUNCATED_CONTENT] + operation.structured_content = None + after = _batch_result_size(limited) + dropped_bytes += max(before - after, 0) + limited.response_truncated_bytes = dropped_bytes + + if _batch_result_size(limited) <= max_bytes: + break + + return limited + + async def _call_one_tool( *, fastmcp: FastMCP, @@ -207,11 +256,13 @@ async def _call_tools_batch( succeeded = sum(1 for result in results if result.success) failed = len(results) - succeeded - return ToolCallBatchResult( - results=results, - succeeded=succeeded, - failed=failed, - stopped_at=stopped_at, + return _limit_batch_result( + ToolCallBatchResult( + results=results, + succeeded=succeeded, + failed=failed, + stopped_at=stopped_at, + ) ) diff --git a/tests/test_batch_tools.py b/tests/test_batch_tools.py index 8fc61bf4..b108a727 100644 --- a/tests/test_batch_tools.py +++ b/tests/test_batch_tools.py @@ -3,8 +3,11 @@ from __future__ import annotations import asyncio +import json import typing as t +import pytest + from libtmux_mcp._utils import ( ANNOTATIONS_DESTRUCTIVE, ANNOTATIONS_MUTATING, @@ -18,6 +21,21 @@ 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, + ), +] + + def _batch_probe_server() -> FastMCP: """Build a small FastMCP server with batch tools and tiered probes.""" from fastmcp import FastMCP @@ -111,6 +129,71 @@ async def _call() -> t.Any: 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(structured, separators=(",", ":"), sort_keys=True) + assert len(serialized.encode("utf-8")) <= DEFAULT_RESPONSE_LIMIT_BYTES + assert first_payload not in serialized + assert second_payload 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"] == {"value": second_payload} + + def test_call_readonly_tools_batch_rejects_mutating_inner_tool() -> None: """Readonly batching does not tunnel a mutating tool call.""" from fastmcp import Client From 67fd69c97660d8e7a02557052cab18c39a15be99 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 06:51:12 -0500 Subject: [PATCH 05/10] mcp(fix[batch_tools]): Advertise batch wrapper side effects why: MCP clients only see the outer batch tool annotations when building approval UI, so wrapper hints must disclose the strongest nested behavior each wrapper can invoke. what: - Mark side-effecting batch wrappers destructive and open-world - Add registration tests for mutating and destructive batch hints --- src/libtmux_mcp/tools/batch_tools.py | 13 +++++-- tests/test_batch_tools.py | 56 ++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 4 deletions(-) diff --git a/src/libtmux_mcp/tools/batch_tools.py b/src/libtmux_mcp/tools/batch_tools.py index 181aace5..7d46a566 100644 --- a/src/libtmux_mcp/tools/batch_tools.py +++ b/src/libtmux_mcp/tools/batch_tools.py @@ -10,9 +10,7 @@ from pydantic import BaseModel from libtmux_mcp._utils import ( - ANNOTATIONS_DESTRUCTIVE, ANNOTATIONS_RO, - ANNOTATIONS_SHELL, TAG_DESTRUCTIVE, TAG_MUTATING, TAG_READONLY, @@ -52,6 +50,13 @@ } ] +_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.""" @@ -337,11 +342,11 @@ def register(mcp: FastMCP) -> None: )(call_readonly_tools_batch) mcp.tool( title="Call Mutating Tools Batch", - annotations=ANNOTATIONS_SHELL, + annotations=_ANNOTATIONS_BATCH_SIDE_EFFECTS, tags={TAG_MUTATING}, )(call_mutating_tools_batch) mcp.tool( title="Call Destructive Tools Batch", - annotations=ANNOTATIONS_DESTRUCTIVE, + annotations=_ANNOTATIONS_BATCH_SIDE_EFFECTS, tags={TAG_DESTRUCTIVE}, )(call_destructive_tools_batch) diff --git a/tests/test_batch_tools.py b/tests/test_batch_tools.py index b108a727..900165ff 100644 --- a/tests/test_batch_tools.py +++ b/tests/test_batch_tools.py @@ -36,6 +36,37 @@ class BatchResponseLimitFixture(t.NamedTuple): ] +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 _batch_probe_server() -> FastMCP: """Build a small FastMCP server with batch tools and tiered probes.""" from fastmcp import FastMCP @@ -312,3 +343,28 @@ async def _call() -> t.Any: [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 From ea60d44923aad44f9fdfe8e4fb2d09a1a025c1f3 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 06:53:27 -0500 Subject: [PATCH 06/10] docs(fix[batch_tools]): Use valid mutating batch example why: The mutating batch docs passed window_name to split_window, whose schema rejects that argument. what: - Show rename_window and split_window targeting the same known window_id - Avoid implying batches feed a created window id into later operations --- docs/tools/batch/call-mutating-tools-batch.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/tools/batch/call-mutating-tools-batch.md b/docs/tools/batch/call-mutating-tools-batch.md index 2a89337f..ceb34bc5 100644 --- a/docs/tools/batch/call-mutating-tools-batch.md +++ b/docs/tools/batch/call-mutating-tools-batch.md @@ -4,7 +4,7 @@ ``` **Use when** you need an ordered workflow made from existing typed MCP -tools, such as creating a window and splitting it, while preserving +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 @@ -24,12 +24,12 @@ running with `LIBTMUX_SAFETY=destructive`. "arguments": { "operations": [ { - "tool": "create_window", - "arguments": {"session_name": "dev", "window_name": "logs"} + "tool": "rename_window", + "arguments": {"window_id": "@2", "new_name": "logs"} }, { "tool": "split_window", - "arguments": {"session_name": "dev", "window_name": "logs"} + "arguments": {"window_id": "@2", "direction": "right"} } ], "on_error": "stop" From ce9b85b42296dbf600a60e20c9c73dd5bcc0e4c6 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 08:04:05 -0500 Subject: [PATCH 07/10] mcp(fix[batch_tools]): Cap FastMCP batch response envelopes why: FastMCP serializes typed tool returns as both text content and structuredContent, so measuring only the batch model could still leave large batched responses above the server response cap. what: - Measure batch truncation against the FastMCP response envelope - Extend the oversized batch regression to cover content plus structuredContent --- src/libtmux_mcp/tools/batch_tools.py | 22 +++++++++++++------ tests/test_batch_tools.py | 32 +++++++++++++++++++++++++--- 2 files changed, 44 insertions(+), 10 deletions(-) diff --git a/src/libtmux_mcp/tools/batch_tools.py b/src/libtmux_mcp/tools/batch_tools.py index 7d46a566..f5f29bae 100644 --- a/src/libtmux_mcp/tools/batch_tools.py +++ b/src/libtmux_mcp/tools/batch_tools.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import time import typing as t @@ -140,9 +141,16 @@ def _ensure_tool_result(tool_name: str, result: t.Any) -> ToolResult: raise ExpectedToolError(msg) -def _batch_result_size(result: ToolCallBatchResult) -> int: - """Return the serialized byte size of a batch result.""" - return len(result.model_dump_json(fallback=str).encode("utf-8")) +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: @@ -156,7 +164,7 @@ def _limit_batch_result( max_bytes: int = DEFAULT_RESPONSE_LIMIT_BYTES, ) -> ToolCallBatchResult: """Elide nested result payloads until the batch envelope fits.""" - if _batch_result_size(result) <= max_bytes: + if _batch_response_size(result) <= max_bytes: return result limited = result.model_copy( @@ -168,14 +176,14 @@ def _limit_batch_result( if not _operation_has_nested_payload(operation): continue - before = _batch_result_size(limited) + before = _batch_response_size(limited) operation.content = [item.copy() for item in _BATCH_TRUNCATED_CONTENT] operation.structured_content = None - after = _batch_result_size(limited) + after = _batch_response_size(limited) dropped_bytes += max(before - after, 0) limited.response_truncated_bytes = dropped_bytes - if _batch_result_size(limited) <= max_bytes: + if _batch_response_size(limited) <= max_bytes: break return limited diff --git a/tests/test_batch_tools.py b/tests/test_batch_tools.py index 900165ff..0a4f3a44 100644 --- a/tests/test_batch_tools.py +++ b/tests/test_batch_tools.py @@ -67,6 +67,22 @@ class BatchAnnotationFixture(t.NamedTuple): ] +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 @@ -206,10 +222,14 @@ async def _call() -> t.Any: assert structured["failed"] == 0 assert structured["stopped_at"] is None - serialized = json.dumps(structured, separators=(",", ":"), sort_keys=True) + 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 in serialized + assert second_payload not in serialized first, second = structured["results"] assert first["index"] == 0 @@ -222,7 +242,13 @@ async def _call() -> t.Any: "text": "[... batch truncated nested content ...]", } ] - assert second["structured_content"] == {"value": second_payload} + assert second["structured_content"] is None + assert second["content"] == [ + { + "type": "text", + "text": "[... batch truncated nested content ...]", + } + ] def test_call_readonly_tools_batch_rejects_mutating_inner_tool() -> None: From 042fd684661a81d0d5593c4e6502f43e1fcf8527 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 08:28:01 -0500 Subject: [PATCH 08/10] mcp(fix[batch_tools]): Limit generic batch operation count why: A batch with enough small row results can exceed the response cap even after nested payload truncation, because row metadata alone still serializes into the FastMCP response envelope. what: - Reject generic tool batches above a fixed operation-count cap - Add a public FastMCP regression for row-only oversized batch responses --- src/libtmux_mcp/tools/batch_tools.py | 5 +++ tests/test_batch_tools.py | 61 ++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/src/libtmux_mcp/tools/batch_tools.py b/src/libtmux_mcp/tools/batch_tools.py index f5f29bae..65d760bd 100644 --- a/src/libtmux_mcp/tools/batch_tools.py +++ b/src/libtmux_mcp/tools/batch_tools.py @@ -44,6 +44,8 @@ } ) +MAX_BATCH_OPERATIONS = 1_000 + _BATCH_TRUNCATED_CONTENT: list[dict[str, t.Any]] = [ { "type": "text", @@ -246,6 +248,9 @@ async def _call_tools_batch( 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) diff --git a/tests/test_batch_tools.py b/tests/test_batch_tools.py index 0a4f3a44..96d31154 100644 --- a/tests/test_batch_tools.py +++ b/tests/test_batch_tools.py @@ -36,6 +36,21 @@ class BatchResponseLimitFixture(t.NamedTuple): ] +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.""" @@ -251,6 +266,52 @@ async def _call() -> t.Any: ] +@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 From 03326692f244dbec1059e67362d65840dd47d05d Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 08:43:53 -0500 Subject: [PATCH 09/10] mcp(docs[CHANGES]): Note bounded batch responses why: The unreleased batch entry predated response capping, which keeps large aggregate results within the server response limit and reports the truncation to callers. what: - Note bounded batch responses in the tier-aware tool batching entry --- CHANGES | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES b/CHANGES index f4c44cf6..ed2e52da 100644 --- a/CHANGES +++ b/CHANGES @@ -10,7 +10,7 @@ _Notes on upcoming releases will be added here_ **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. (#79) +{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 and the result flags the truncation. (#79) ## libtmux-mcp 0.1.0a13 (2026-06-13) From 98d2e82d969f58c075778d70a5a739da177c7f94 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sun, 14 Jun 2026 08:48:32 -0500 Subject: [PATCH 10/10] mcp(docs[CHANGES]): Note the batch operation-count limit why: The unreleased batch entry documented the silent response bound but not the hard rejection callers hit when a batch carries too many operations. what: - Note that oversized batch operation lists are rejected --- CHANGES | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES b/CHANGES index ed2e52da..ad8aaa66 100644 --- a/CHANGES +++ b/CHANGES @@ -10,7 +10,7 @@ _Notes on upcoming releases will be added here_ **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 and the result flags the truncation. (#79) +{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)