Middleware¶

Middleware for libtmux MCP server.

Provides the project’s middleware infrastructure, in definition order:

SafetyMiddleware gates tools by safety tier based on the LIBTMUX_SAFETY environment variable. Tools tagged above the configured tier are hidden from listing and blocked from execution.
ToolErrorResultMiddleware converts tool-call failures into ToolResult(is_error=True) results that carry the clean error message plus a structured meta payload, instead of fastmcp’s stock -32603 catch-all that prefixed every expected failure with "Internal error: ".
AuditMiddleware emits a structured log record for each tool invocation (name, duration, outcome, client/request ids, and a summary of arguments with payload-bearing fields redacted to a length + SHA-256 prefix).
ReadonlyRetryMiddleware retries transient libtmux failures, but only for readonly tools — re-running a mutating tool would silently double side effects.
TailPreservingResponseLimitingMiddleware is a backstop cap for oversized tool output. Unlike FastMCP’s stock ResponseLimitingMiddleware it preserves the tail of the response — terminal scrollback has its active prompt and the output agents actually need at the bottom, so dropping the head is always the correct direction.

class libtmux_mcp.middleware.SafetyMiddleware ¶ class[source] class[source] class libtmux_mcp.middleware.SafetyMiddleware ¶

Bases: Middleware

Gate tools by safety tier.

Parameters:: max_tier (str) – Maximum allowed tier. One of TAG_READONLY, TAG_MUTATING, or TAG_DESTRUCTIVE.

libtmux_mcp.middleware._schema_validation_error(error) ¶ function[source] function[source] libtmux_mcp.middleware._schema_validation_error(error) ¶

Return the Pydantic validation error behind a schema failure.

Parameters:: error (BaseException)
Return type:: ValidationError | None

libtmux_mcp.middleware._is_schema_validation_error(error) ¶ function[source] function[source] libtmux_mcp.middleware._is_schema_validation_error(error) ¶

Return True for fastmcp argument-schema validation failures.

fastmcp validates tool arguments against the input schema before tool code runs, raising a bare pydantic.ValidationError — too early for the handle_tool_errors decorators to classify. Bad arguments are agent-correctable (fix the call and retry), so they get the same expected/WARNING treatment as ExpectedToolError.

Output validation cannot be mistaken for this case: fastmcp’s tool layer converts output-shape failures into error results itself, so they never reach the middleware as exceptions.

Parameters:: error (BaseException)
Return type:: bool

libtmux_mcp.middleware._validation_errors_without_inputs(error) ¶ function[source] function[source] libtmux_mcp.middleware._validation_errors_without_inputs(error) ¶

Return validation errors without rejected input values.

Parameters:: error (ValidationError)
Return type:: list[dict[str, Any]]

libtmux_mcp.middleware._format_schema_validation_error(error) ¶ function[source] function[source] libtmux_mcp.middleware._format_schema_validation_error(error) ¶

Format a Pydantic validation error without raw input values.

Parameters:: error (BaseException)
Return type:: str

libtmux_mcp.middleware._strip_validation_error_inputs(value) ¶ function[source] function[source] libtmux_mcp.middleware._strip_validation_error_inputs(value) ¶

Remove raw input payloads from structured validation errors.

Parameters:: value (Any)
Return type:: Any

class libtmux_mcp.middleware._FastMCPValidationLogFilter ¶ class[source] class[source] class libtmux_mcp.middleware._FastMCPValidationLogFilter ¶

Bases: Filter

Redact FastMCP invalid-argument warning payloads.

libtmux_mcp.middleware.install_fastmcp_validation_log_filter() ¶ function[source] function[source] libtmux_mcp.middleware.install_fastmcp_validation_log_filter() ¶

Install the FastMCP validation log redaction filter once.

Return type:: None

libtmux_mcp.middleware._CLIENT_SCHEDULING_FLAG = 'wait_for_previous' ¶ data data libtmux_mcp.middleware._CLIENT_SCHEDULING_FLAG = 'wait_for_previous' ¶: Scheduling flag some MCP clients (notably Gemini CLI when batching several tool calls in one turn) merge into the tool’s arguments. Recognized only to word the rejection helpfully — the argument is still rejected, never silently stripped, so genuine argument typos from other clients stay loud. Contrast MemPalace/mempalace #322, which strips the key, and #647, which whitelists arguments against the schema — silent dropping would let a mis-named flag on a mutating tool (e.g. enter on send_keys) run with defaults.

libtmux_mcp.middleware._unexpected_kwargs(error) ¶ function[source] function[source] libtmux_mcp.middleware._unexpected_kwargs(error) ¶

Argument names rejected as unexpected by schema validation.

Reads pydantic’s structured errors() from error or its __cause__ (same posture as _is_schema_validation_error()) and returns the names flagged unexpected_keyword_argument. Empty list for every other failure shape.

Parameters:: error (BaseException)
Return type:: list[str]

libtmux_mcp.middleware._client_label(context) ¶ function[source] function[source] libtmux_mcp.middleware._client_label(context) ¶

"name version" of the connected client, when the handshake exposed it.

Walks fastmcp_context.session.client_params.clientInfo — the MCP initialize handshake’s client identity. Every hop can be absent (unit-test contexts, background tasks, clients that omit clientInfo), so any failure resolves to None. Used only to word error suggestions; never gates behavior.

Parameters:: context (MiddlewareContext | None)
Return type:: str | None

libtmux_mcp.middleware._error_tool_result(error, context=None) ¶ function[source] function[source] libtmux_mcp.middleware._error_tool_result(error, context=None) ¶

Build a rich ToolResult(is_error=True) from a tool failure.

The text block carries the error message exactly as raised — no transform-layer prefix — with the recovery suggestion appended when the error provides one. meta mirrors the details in machine-readable form:

error_type — class name of the originating exception (__cause__ when the raise site chained one, so agents see PaneNotFound rather than the ToolError wrapper).
expected — True for agent-correctable failures (ExpectedToolError and argument-schema validation errors), False for operator faults and potential server bugs.
suggestion — recovery hint. Carried by the error when the raise site provided one; synthesized for schema-validation failures that rejected unexpected arguments, so the agent knows to drop or fix exactly those names (with a client-flag note for _CLIENT_SCHEDULING_FLAG leaks, naming the client via context when the handshake exposed it).

structured_content is deliberately left unset: tools declare output schemas for their success payloads, and clients validate structuredContent against them — an error-shaped payload there would fail validation on strict clients.

Parameters:

error (Exception)
context (MiddlewareContext | None)

Return type:

ToolResult

class libtmux_mcp.middleware.ToolErrorResultMiddleware ¶ class[source] class[source] class libtmux_mcp.middleware.ToolErrorResultMiddleware ¶

Bases: ErrorHandlingMiddleware

Convert tool-call failures into rich ToolResult errors.

Replaces the stock ErrorHandlingMiddleware behavior for tools/call only. The stock transform_errors=True path funnels every non-MCP exception through a -32603 catch-all, so agents received "Internal error: Pane not found: %5" — the transform mangled every expected failure message. This subclass intercepts tool-call exceptions first (on_call_tool is the innermost hook of a middleware’s chain) and returns _error_tool_result() instead; non-tool messages fall through to the inherited on_message, preserving the MCP -32002 resource-not-found transform this middleware was originally adopted for.

Logging honors FastMCPError.log_level (fastmcp >= 3.3): the expected failures demoted to WARNING by ExpectedToolError no longer get re-shouted at ERROR by the stock _log_error. Argument-schema validation failures — raised by fastmcp before tool code can classify them — are treated as expected too (see _is_schema_validation_error()), and when the rejected arguments include unexpected names the error result carries a synthesized suggestion telling the agent which names to drop or fix (see _error_tool_result()).

Ordering invariant: must sit outside AuditMiddleware, ReadonlyRetryMiddleware, and SafetyMiddleware. All three depend on exception semantics — audit detects failures by catching, retry matches LibTmuxException via __cause__, and safety’s tier denials must propagate as exceptions for audit to record them — so converting the exception to a result any deeper in the stack would silently break all three.

_log_error(error, context) ¶ method[source] method[source] _log_error(error, context) ¶

Log at the error’s own log_level instead of a flat ERROR.

Mirrors the stock implementation (error-count tracking, optional traceback, error callback) but routes the record through logger.log with FastMCPError.log_level. Exceptions that don’t carry one default to ERROR — except argument-schema validation failures, which are agent-correctable and log at WARNING like every other invalid argument (see _is_schema_validation_error()).

Parameters:

error (Exception)
context (MiddlewareContext)

Return type:

None

libtmux_mcp.middleware._SENSITIVE_ARG_NAMES: frozenset[str] = frozenset({'command', 'content', 'environment', 'keys', 'shell', 'text', 'value'}) ¶ data data libtmux_mcp.middleware._SENSITIVE_ARG_NAMES: frozenset[str] = frozenset({'command', 'content', 'environment', 'keys', 'shell', 'text', 'value'}) ¶: Argument names that carry user-supplied payloads we never want in logs. keys (send_keys), text (paste_text), command (run_command), value (set_environment), content (load_buffer), shell (respawn_pane), and environment (respawn_pane) can contain commands, secrets, or arbitrary large strings. Matched by exact name, case-sensitive, to mirror the tool signatures.

environment is dict-shaped (dict[str, str]); the redaction logic in _summarize_args() recognises this and digests each value while leaving the keys (env var names like DATABASE_URL) visible — env var names are operator-debug-useful, but their values are the secret. All other entries are scalar strings; mixing the two is intentional.

Note on shell and environment redaction: this redacts the MCP audit log only. respawn_pane(shell="env SECRET=... bash") and environment={"AWS_SECRET_KEY": "..."} may briefly expose the values via the OS process table and tmux’s pane_current_command metadata until the spawned shell takes over — see docs/topics/safety.md.

libtmux_mcp.middleware._NESTED_ARG_LIST_NAMES: frozenset[str] = frozenset({'operations'}) ¶ data data libtmux_mcp.middleware._NESTED_ARG_LIST_NAMES: frozenset[str] = frozenset({'operations'}) ¶: Nested argument containers that may contain sensitive argument names. 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.

libtmux_mcp.middleware._MAX_LOGGED_STR_LEN: int = 200 ¶ data data libtmux_mcp.middleware._MAX_LOGGED_STR_LEN: int = 200 ¶: String arguments longer than this get truncated in the log summary to keep records bounded. Non-sensitive strings only — sensitive ones are replaced entirely by their digest.

libtmux_mcp.middleware._redact_digest(value) ¶ function[source] function[source] libtmux_mcp.middleware._redact_digest(value) ¶

Return a length + SHA-256 prefix summary of value.

The digest is stable and deterministic, which lets operators correlate the same payload across log lines without ever recording the payload itself.

Examples

>>> _redact_digest("hello")
{'len': 5, 'sha256_prefix': '2cf24dba5fb0'}
>>> _redact_digest("")
{'len': 0, 'sha256_prefix': 'e3b0c44298fc'}

Parameters:: value (str)
Return type:: dict[str, Any]

libtmux_mcp.middleware._redacted_value_shape(value) ¶ function[source] function[source] libtmux_mcp.middleware._redacted_value_shape(value) ¶

Return non-payload metadata for a value that cannot be logged.

Parameters:: value (Any)
Return type:: dict[str, Any]

libtmux_mcp.middleware._summarize_send_keys_operation_args(args) ¶ function[source] function[source] libtmux_mcp.middleware._summarize_send_keys_operation_args(args) ¶

Summarize one send_keys_batch operation for audit logging.

Parameters:: args (dict[str, Any])
Return type:: dict[str, Any]

libtmux_mcp.middleware._summarize_tool_batch_operation_args(args) ¶ function[source] function[source] libtmux_mcp.middleware._summarize_tool_batch_operation_args(args) ¶

Summarize one generic tool-batch operation for audit logging.

Parameters:: args (dict[str, Any])
Return type:: dict[str, Any]

libtmux_mcp.middleware._summarize_nested_operation_args(args) ¶ function[source] function[source] libtmux_mcp.middleware._summarize_nested_operation_args(args) ¶

Summarize a known nested operation shape.

Parameters:: args (dict[str, Any])
Return type:: dict[str, Any]

libtmux_mcp.middleware._summarize_args(args) ¶ function[source] function[source] libtmux_mcp.middleware._summarize_args(args) ¶

Summarize tool arguments for audit logging.

Sensitive keys get replaced by a digest; over-long strings get truncated with a marker; everything else passes through as-is. Sensitive values that are dict-shaped (e.g. environment on respawn_pane) have each value digested while keys remain visible — env-var-name-like keys are operator-debug-useful and rarely sensitive, while their values usually are. Known nested operation lists are summarized recursively so batched tool calls keep target metadata while redacting inner payloads.

Examples

Non-sensitive scalars pass through unchanged:

>>> _summarize_args({"pane_id": "%1", "bracket": True})
{'pane_id': '%1', 'bracket': True}

Sensitive payload names are replaced by a digest dict:

>>> _summarize_args({"keys": "rm -rf /"})["keys"]["len"]
8

Sensitive dict-shaped payloads keep their keys but digest values:

>>> redacted = _summarize_args({"environment": {"FOO": "bar"}})
>>> redacted["environment"]["FOO"]["len"]
3
>>> "bar" in str(redacted)
False

Parameters:: args (dict[str, Any])
Return type:: dict[str, Any]

class libtmux_mcp.middleware.AuditMiddleware ¶ class[source] class[source] class libtmux_mcp.middleware.AuditMiddleware ¶

Bases: Middleware

Emit a structured log record per tool invocation.

Records carry: tool name, outcome (ok/error), duration in ms, error type on failure, the fastmcp client_id / request_id when available, and a redacted argument summary. The logger name is libtmux_mcp.audit by default so operators can route it independently (e.g. to a JSON file) without touching the module libtmux_mcp logger.

Parameters:: logger_name (str) – Name of the logging logger used for audit records.

libtmux_mcp.middleware.DEFAULT_RESPONSE_LIMIT_BYTES = 1000000 ¶ data data libtmux_mcp.middleware.DEFAULT_RESPONSE_LIMIT_BYTES = 1000000 ¶: Default byte ceiling for TailPreservingResponseLimitingMiddleware. Matches FastMCP’s stock 1 MB default so normal schema-bearing tool responses stay below this global backstop. Tool-level caps remain responsible for terminal-specific truncation metadata.

class libtmux_mcp.middleware.ReadonlyRetryMiddleware ¶ class[source] class[source] class libtmux_mcp.middleware.ReadonlyRetryMiddleware ¶

Bases: Middleware

Retry transient libtmux failures, but only for readonly tools.

Wraps fastmcp’s fastmcp.server.middleware.error_handling.RetryMiddleware so retries are bounded by the safety tier the tool is registered under. Mutating and destructive tools (send_keys, create_session, kill_server, …) pass straight through — re-running them on a transient socket error would silently double side effects, which is unacceptable. Readonly tools (list_sessions, capture_pane, snapshot_pane, …) are safe to retry because they observe state without mutating it.

Default retry trigger is libtmux.exc.LibTmuxException — libtmux wraps the subprocess failures we actually want to retry (socket EAGAIN, transient connect errors). The fastmcp default (ConnectionError, TimeoutError) does NOT match these, so the upstream defaults would be a silent no-op.

Place this in the middleware stack inside AuditMiddleware (so retried calls are audited once each) and outside SafetyMiddleware (so tier-denied tools never reach retry).

__init__(max_retries=1, base_delay=0.1, max_delay=1.0, backoff_multiplier=2.0, retry_exceptions=(libtmux_exc.LibTmuxException,), logger_=None) ¶ method[source] method[source] __init__(max_retries=1, base_delay=0.1, max_delay=1.0, backoff_multiplier=2.0, retry_exceptions=(libtmux_exc.LibTmuxException,), logger_=None) ¶

Configure the underlying retry policy.

Defaults are deliberately small. max_retries=1 keeps audit log noise minimal (one retry per failed call, not three). The 100 ms / 1 s backoff window matches the expected duration of a transient libtmux socket hiccup — a longer backoff would just delay a real failure without adding meaningful retry headroom.

logger_ defaults to logging.getLogger("libtmux_mcp.retry") when not supplied — keeps retry events on the project’s libtmux_mcp.* namespace so operators routing the audit stream capture them. Without this default, fastmcp’s stock RetryMiddleware would log to fastmcp.retry and miss any project-namespace log routing.

Parameters:

max_retries (int)
base_delay (float)
max_delay (float)
backoff_multiplier (float)
retry_exceptions (tuple[type[Exception], ...])
logger_ (Logger | None)

Return type:

None

libtmux_mcp.middleware._TRUNCATION_HEADER_TEMPLATE = '[... truncated {dropped} bytes ...]\n' ¶ data data libtmux_mcp.middleware._TRUNCATION_HEADER_TEMPLATE = '[... truncated {dropped} bytes ...]\n' ¶: Header prefixed to a truncated response. Intentionally matches the format used by the per-tool capture_pane truncation so clients see a consistent marker regardless of which layer fired.

class libtmux_mcp.middleware.TailPreservingResponseLimitingMiddleware ¶ class[source] class[source] class libtmux_mcp.middleware.TailPreservingResponseLimitingMiddleware ¶

Bases: ResponseLimitingMiddleware

Response-limiter that keeps the tail of oversized output.

FastMCP’s stock ResponseLimitingMiddleware truncates the tail of the response (keeps the start, appends a suffix). That’s exactly wrong for terminal scrollback, where the active shell prompt and most recent command output live at the bottom of the buffer. This subclass overrides _truncate_to_result to drop the head instead, prefixing a single truncation-header line so callers can detect the cap fired.

Used as a global backstop for libtmux_mcp.tools.pane_tools.io.capture_pane(), libtmux_mcp.tools.pane_tools.capture_since.capture_since(), libtmux_mcp.tools.pane_tools.meta.snapshot_pane(), and libtmux_mcp.tools.pane_tools.search.search_panes(). Per-tool caps at the tool layer fire first under normal operation; this middleware catches pathological output from future tools that forget to declare their own bounds.

Error results keep their is_error flag through truncation. The stock truncation path rebuilds the result without it, which turns an oversized error (e.g. a validation message echoing a huge argument) into an apparent success — MCP clients then validate the truncated text against the tool’s output schema and fail with a transport-level error instead of delivering the tool error. meta (error_type / expected / suggestion) already survives via the base class, and tail-preservation keeps the suggestion line, which sits at the end of the text.