Utilities¶

Shared utilities for libtmux MCP server.

Provides server caching, object resolution, serialization, and error handling for all MCP tool functions.

exception libtmux_mcp._utils.ExpectedToolError ¶ exception[source] exception[source] exception libtmux_mcp._utils.ExpectedToolError ¶

Bases: ToolError

ToolError for expected, agent-correctable failures.

Defaults the error’s log_level to WARNING (honored by fastmcp >= 3.3 when logging tool/resource failures) so routine validation errors, missing objects, and tier denials do not surface as ERROR records. Unexpected failures keep stock ToolError and its ERROR default — those are the ones operators must see.

Parameters:

*args (object) – Positional arguments forwarded to ToolError (typically the error message).
log_level (int) – Level fastmcp’s server layer logs this failure at. Defaults to logging.WARNING.
suggestion (str, optional) – Agent-facing recovery hint. ToolErrorResultMiddleware appends it to the error result’s text and mirrors it into the result’s meta.

Examples

>>> import logging
>>> ExpectedToolError("Pane not found: %5").log_level == logging.WARNING
True

An explicit level still wins:

>>> err = ExpectedToolError("noisy", log_level=logging.INFO)
>>> err.log_level == logging.INFO
True

Catch sites that handle ToolError keep working — this is a plain subclass:

>>> isinstance(ExpectedToolError("x"), ToolError)
True

An optional suggestion carries an agent-facing recovery hint; libtmux_mcp.middleware.ToolErrorResultMiddleware surfaces it in the error result’s text and meta:

>>> err = ExpectedToolError("Pane not found: %5",
...     suggestion="Call list_panes to discover valid pane ids.")
>>> err.suggestion
'Call list_panes to discover valid pane ids.'
>>> ExpectedToolError("no hint").suggestion is None
True

class libtmux_mcp._utils.CallerIdentity ¶ class[source] class[source] class libtmux_mcp._utils.CallerIdentity ¶

Bases: object

Identity of the tmux pane hosting this MCP server process.

Parsed from the TMUX and TMUX_PANE environment variables that tmux injects into every child of a pane. TMUX has the format socket_path,server_pid,session_id (see tmux environ.c:281).

Used to scope self-protection checks to the caller’s own tmux server — a pane ID like %1 is only unique within a single server, so comparisons must also verify the socket path matches.

libtmux_mcp._utils._get_caller_identity() ¶ function[source] function[source] libtmux_mcp._utils._get_caller_identity() ¶

Return the caller’s tmux identity, or None if not inside tmux.

Reads TMUX for socket_path/server_pid/session_id and TMUX_PANE for the pane id. Tolerant of missing/malformed TMUX values — callers should check individual fields rather than relying on all being populated.

Return type:: CallerIdentity | None

libtmux_mcp._utils._compute_is_caller(pane) ¶ function[source] function[source] libtmux_mcp._utils._compute_is_caller(pane) ¶

Decide whether pane is the MCP caller’s own tmux pane.

The returned value is used as the is_caller annotation on PaneInfo, PaneSnapshot, and PaneContentMatch.

Tri-state semantics match the original bare-equality check:

None — process is not inside tmux at all (neither TMUX nor TMUX_PANE are set). No caller exists, so the annotation carries no signal.
True — the caller’s TMUX_PANE matches pane.pane_id and _caller_is_strictly_on_server() confirms the caller’s socket realpath equals the target’s.
False — the pane ids differ, or they match but the socket does not (or cannot be proven to). A bare pane-id equality check would have returned True here, which is the cross-socket false-positive fixed by tmux-python/libtmux-mcp #19.

Uses _caller_is_strictly_on_server() rather than _caller_is_on_server(): the kill-guard comparator is conservative-True-when-uncertain (right for blocking destructive actions, wrong for an informational annotation that should demand a positive match). The strict variant declines the basename fallback, the unresolvable-target branch, and the socket-path-unset branch so ambiguous cases resolve to False.

Parameters:: pane (Pane)
Return type:: bool | None

libtmux_mcp._utils._effective_socket_path(server) ¶ function[source] function[source] libtmux_mcp._utils._effective_socket_path(server) ¶

Return the filesystem socket path a Server will actually use.

libtmux leaves Server.socket_path as None when only socket_name (or neither) was supplied, but tmux still resolves to a real path under ${TMUX_TMPDIR:-/tmp}/tmux-<uid>/<name>. This helper reproduces that resolution so _caller_is_on_server() can compare against the caller’s TMUX socket path.

Resolution order:

Server.socket_path if libtmux already has it.
tmux display-message -p '#{socket_path}' against the target server — authoritative because tmux itself reports the path it is actually using, regardless of our process environment. Necessary on macOS where $TMUX_TMPDIR under launchd diverges from the interactive shell (see docs/topics/safety.md for the self-kill guard gap this closes).
Fallback: reconstruct from $TMUX_TMPDIR + euid + socket name. This path is reached only when the target server is unreachable (e.g. not running), in which case no self-kill is possible and the conservative caller check still blocks via _caller_is_on_server’s None-socket branch.

Parameters:: server (Server)
Return type:: str | None

libtmux_mcp._utils._caller_is_on_server(server, caller) ¶ function[source] function[source] libtmux_mcp._utils._caller_is_on_server(server, caller) ¶

Return True if caller looks like it is on the same tmux server.

Compares socket paths via os.path.realpath() so symlinked temp dirs still match, then falls back to basename comparison when realpath disagrees — the authoritative caller-side $TMUX name and the target’s declared socket_name are both unaffected by $TMUX_TMPDIR divergence (the macOS launchd case), so a last-chance name match still blocks a self-kill when the path comparison was fooled by env mismatch.

Decision table:

caller is None → False. The process isn’t inside tmux at all, so there is no caller-side pane to protect and no self-kill is possible.
caller has a pane id but no socket path (e.g. TMUX_PANE set without TMUX) → True. We can’t rule out that the caller is on the target server, so err on the side of blocking a destructive action.
target server has no resolvable socket path → True. Same conservative reasoning.
realpath of caller’s socket path matches target’s effective path → True (primary positive signal).
basename of caller’s socket path equals target’s socket_name (or "default") → True. Conservative last-chance block for env-mismatch scenarios where reconstruction produced a wrong path but the name was authoritative on both sides. Trades off one exotic false positive (two daemons with identical socket_name under different tmpdirs) for a real safety property.
Otherwise → False.

When a conservative block is a false positive, the caller’s error message directs the user to run tmux manually.

Parameters:

server (Server)
caller (CallerIdentity | None)

Return type:

bool

libtmux_mcp._utils._caller_is_strictly_on_server(server, caller) ¶ function[source] function[source] libtmux_mcp._utils._caller_is_strictly_on_server(server, caller) ¶

Return True only on a confirmed socket-path match.

Counterpart to _caller_is_on_server() for the informational is_caller annotation. The destructive-action guard is biased toward True-when-uncertain so a macOS $TMUX_TMPDIR divergence cannot fool it into permitting self-kill; the annotation cannot absorb that bias — ambiguous cases are exactly the cross-socket false positives documented by tmux-python/libtmux-mcp #19. This function therefore declines every branch other than a confirmed realpath match.

Decision table:

caller is None → False. No caller identity.
caller.socket_path unset (TMUX_PANE set without TMUX) → False. We cannot verify the caller is on this server.
target server’s effective socket path unresolvable → False.
realpath of caller’s socket path equals target’s effective path → True. Primary and only positive signal.
Fallback on OSError from realpath: exact string match → True. Still a positive signal, just without the resolve step.
Otherwise → False (including the basename-only match that _caller_is_on_server() permits as a conservative block).

Parameters:

server (Server)
caller (CallerIdentity | None)

Return type:

bool

libtmux_mcp._utils.ANNOTATIONS_SHELL: dict[str, bool] = {'destructiveHint': False, 'idempotentHint': False, 'openWorldHint': True, 'readOnlyHint': False} ¶ data data libtmux_mcp._utils.ANNOTATIONS_SHELL: dict[str, bool] = {'destructiveHint': False, 'idempotentHint': False, 'openWorldHint': True, 'readOnlyHint': False} ¶

Annotations for tools that move user-supplied payloads into a shell context. Six consumers today:

send_keys, run_command, paste_text, pipe_pane — the canonical shell-driving tools; caller’s keys/command/text/stream reaches the shell prompt or pipes into an external command respectively.
load_buffer, paste_buffer — load_buffer stages content into a tmux paste buffer; paste_buffer pushes that content into a target pane where the shell receives it as input. The two are split into a stage/fire pair so callers can validate before paste, but both participate in the same open-world transfer.

Distinguished from ANNOTATIONS_CREATE by openWorldHint=True: the effects of these tools extend into whatever command or content the caller supplies, which is the canonical open-world MCP interaction.

libtmux_mcp._utils.DISCOVERY_META: dict[str, Any] = {'anthropic/alwaysLoad': True} ¶ data data libtmux_mcp._utils.DISCOVERY_META: dict[str, Any] = {'anthropic/alwaysLoad': True} ¶: Per-tool MCP meta payload that hints clients to keep this tool always visible (not deferred). FastMCP passes meta opaquely (verified vs ~/study/python/fastmcp/src — no special handling); honoring is delegated to Claude Code, where alwaysLoad is documented at https://code.claude.com/docs/en/mcp (v2.1.121+).

Best-effort by design — safe no-op for clients that don’t index the anthropic/* namespace. Apply only to read-tier discovery anchors (list_panes, list_windows, snapshot_pane); each always-loaded tool consumes a fixed schema budget in clients that honour the hint, so widening the set has a real cost.

libtmux_mcp._utils.ANNOTATIONS_MUTATING_DESTRUCTIVE: dict[str, bool] = {'destructiveHint': True, 'idempotentHint': False, 'openWorldHint': False, 'readOnlyHint': False} ¶ data data libtmux_mcp._utils.ANNOTATIONS_MUTATING_DESTRUCTIVE: dict[str, bool] = {'destructiveHint': True, 'idempotentHint': False, 'openWorldHint': False, 'readOnlyHint': False} ¶: Annotations for tools that stay in the mutating tier (so they remain visible to default-profile agents) but whose default behaviour can terminate processes or otherwise lose state.

Canonical users include respawn_pane and clear_pane: tier=mutating because shell recovery and scrollback cleanup are part of normal agent workflows, while the hints still disclose process termination or state loss.

Distinct from ANNOTATIONS_DESTRUCTIVE (same hint values) because the tier tag differs: ANNOTATIONS_DESTRUCTIVE is paired with TAG_DESTRUCTIVE everywhere it is used; this preset is paired with TAG_MUTATING. The distinct name documents intent at the call site.

libtmux_mcp._utils._tmux_argv(server, *tmux_args) ¶ function[source] function[source] libtmux_mcp._utils._tmux_argv(server, *tmux_args) ¶

Build a full tmux argv list honouring socket_name and socket_path.

Internal helper shared by every module that has to invoke the tmux binary directly via subprocess.run() (the buffer, wait-for, and paste_text tools). libtmux’s own Server.cmd() wraps the same logic but does not expose a timeout, so tools that need bounded blocking have to shell out themselves — and when they do they must honour the caller’s socket.

Parameters:

server (libtmux.server.Server) – The resolved server whose socket to target.
*tmux_args (str) – tmux subcommand and its flags, e.g. "load-buffer", "-b", name.

Returns:

Complete argv ready for subprocess.run().

Return type:

list[str]

Examples

>>> class _S:
...     tmux_bin = "tmux"
...     socket_name = "s"
...     socket_path = None
>>> _tmux_argv(t.cast("Server", _S()), "list-sessions")
['tmux', '-L', 's', 'list-sessions']

>>> class _P:
...     tmux_bin = "tmux"
...     socket_name = None
...     socket_path = "/tmp/tmux-1000/default"
>>> _tmux_argv(t.cast("Server", _P()), "ls")
['tmux', '-S', '/tmp/tmux-1000/default', 'ls']

libtmux_mcp._utils._get_server(socket_name=None, socket_path=None) ¶ function[source] function[source] libtmux_mcp._utils._get_server(socket_name=None, socket_path=None) ¶

Get or create a cached Server instance.

Parameters:

socket_name (str, optional) – tmux socket name (-L). Falls back to LIBTMUX_SOCKET env var.
socket_path (str, optional) – tmux socket path (-S). Falls back to LIBTMUX_SOCKET_PATH env var.

Returns:

A cached libtmux Server instance.

Return type:

Server

libtmux_mcp._utils._invalidate_server(socket_name=None, socket_path=None) ¶ function[source] function[source] libtmux_mcp._utils._invalidate_server(socket_name=None, socket_path=None) ¶

Evict a server from the cache.

Parameters:

socket_name (str, optional) – tmux socket name used in the cache key.
socket_path (str, optional) – tmux socket path used in the cache key.

Return type:

None

libtmux_mcp._utils._resolve_session(server, session_name=None, session_id=None) ¶ function[source] function[source] libtmux_mcp._utils._resolve_session(server, session_name=None, session_id=None) ¶

Resolve a session by name or ID.

Parameters:

server (Server) – The tmux server.
session_name (str, optional) – Session name to look up.
session_id (str, optional) – Session ID (e.g. ‘$1’) to look up.

Return type:

Session

Raises:

exc.TmuxObjectDoesNotExist – If no matching session is found.

libtmux_mcp._utils._resolve_window(server, session=None, window_id=None, window_index=None, session_name=None, session_id=None) ¶ function[source] function[source] libtmux_mcp._utils._resolve_window(server, session=None, window_id=None, window_index=None, session_name=None, session_id=None) ¶

Resolve a window by ID, index, or default.

Parameters:

server (Server) – The tmux server.
session (Session, optional) – Session to search within.
window_id (str, optional) – Window ID (e.g. ‘@1’).
window_index (str, optional) – Window index within the session.
session_name (str, optional) – Session name for resolution.
session_id (str, optional) – Session ID for resolution.

Return type:

Window

Raises:

exc.TmuxObjectDoesNotExist – If no matching window is found.

libtmux_mcp._utils._resolve_pane(server, pane_id=None, session_name=None, session_id=None, window_id=None, window_index=None, pane_index=None) ¶ function[source] function[source] libtmux_mcp._utils._resolve_pane(server, pane_id=None, session_name=None, session_id=None, window_id=None, window_index=None, pane_index=None) ¶

Resolve a pane by ID or hierarchical targeting.

Parameters:

server (Server) – The tmux server.
pane_id (str, optional) – Pane ID (e.g. ‘%1’). Globally unique within a server.
session_name (str, optional) – Session name for hierarchical resolution.
session_id (str, optional) – Session ID for hierarchical resolution.
window_id (str, optional) – Window ID for hierarchical resolution.
window_index (str, optional) – Window index for hierarchical resolution.
pane_index (str, optional) – Pane index within the window.

Return type:

Pane

Raises:

exc.TmuxObjectDoesNotExist – If no matching pane is found.

libtmux_mcp._utils._coerce_dict_arg(name, value) ¶ function[source] function[source] libtmux_mcp._utils._coerce_dict_arg(name, value) ¶

Coerce a tool parameter to a dict, accepting JSON-string form.

Workaround: Cursor’s composer-1/composer-1.5 models and some other MCP clients serialize dict params as JSON strings instead of objects. Claude and GPT models through Cursor work fine; the bug is model-specific. This helper is the canonical place to absorb the string form so each tool can stay dict-typed on the Python side. Callers pass name so the error messages identify the offending parameter.

See:: https://forum.cursor.com/t/145807 https://github.com/anthropics/claude-code/issues/5504

Parameters:

name (str) – Parameter name, used in error messages.
value (dict, str, or None) – Either an already-decoded dict, a JSON string of a dict, or None.

Returns:

The decoded dict, or None if the input was None or an empty string.

Return type:

dict or None

Raises:

ExpectedToolError – If value is a string that is not valid JSON, or decodes to a JSON value that is not an object.

libtmux_mcp._utils._apply_filters(items, filters, serializer) ¶ function[source] function[source] libtmux_mcp._utils._apply_filters(items, filters, serializer) ¶

Apply QueryList filters and serialize results.

Parameters:

items (QueryList) – The QueryList of tmux objects to filter.
filters (dict or str, optional) – Django-style filters as a dict (e.g. {"session_name__contains": "dev"}) or as a JSON string. Some MCP clients require the string form. If None or empty, all items are returned.
serializer (callable) – Serializer function to convert each item to a model.

Returns:

Serialized list of matching items.

Return type:

list

Raises:

ExpectedToolError – If a filter key uses an invalid lookup operator.

libtmux_mcp._utils._serialize_session(session) ¶ function[source] function[source] libtmux_mcp._utils._serialize_session(session) ¶

Serialize a Session to a Pydantic model.

Parameters:: session (Session) – The session to serialize.
Returns:: Session data including id, name, window count.
Return type:: SessionInfo

libtmux_mcp._utils._serialize_window(window) ¶ function[source] function[source] libtmux_mcp._utils._serialize_window(window) ¶

Serialize a Window to a Pydantic model.

Parameters:: window (Window) – The window to serialize.
Returns:: Window data including id, name, index, pane count, layout.
Return type:: WindowInfo

libtmux_mcp._utils._coerce_int(value) ¶ function[source] function[source] libtmux_mcp._utils._coerce_int(value) ¶

Parse a tmux format-string number into int or None.

tmux format variables come back as strings; an empty string means “tmux returned nothing” (e.g. older tmux that doesn’t know the var).

Parameters:: value (str | None)
Return type:: int | None

libtmux_mcp._utils._coerce_bool(value) ¶ function[source] function[source] libtmux_mcp._utils._coerce_bool(value) ¶

Parse a tmux "1"/"0" flag into bool or None.

Mirrors libtmux’s own Pane.at_top / at_bottom typing, which folds "1" to True and everything else to False — except we keep None distinct so callers can tell “tmux didn’t tell us” from “tmux said no”.

Parameters:: value (str | None)
Return type:: bool | None

libtmux_mcp._utils._serialize_pane(pane) ¶ function[source] function[source] libtmux_mcp._utils._serialize_pane(pane) ¶

Serialize a Pane to a Pydantic model.

Parameters:: pane (Pane) – The pane to serialize.
Returns:: Pane data including id, dimensions, geometry, current command, title.
Return type:: PaneInfo

libtmux_mcp._utils._map_exception_to_tool_error(fn_name, e) ¶ function[source] function[source] libtmux_mcp._utils._map_exception_to_tool_error(fn_name, e) ¶

Translate a libtmux / unexpected exception into a ToolError.

Shared between the sync and async handle_tool_errors* decorators so the two paths stay byte-for-byte identical in what agents see.

Expected, agent-correctable failures map to ExpectedToolError (logged at WARNING). Two cases stay at ERROR: a missing tmux binary (operator-environment fault that must be loud) and the unexpected catch-all (potential bug in this server).

Parameters:

fn_name (str)
e (BaseException)

Return type:

ToolError

libtmux_mcp._utils.handle_tool_errors(fn) ¶ function[source] function[source] libtmux_mcp._utils.handle_tool_errors(fn) ¶

Decorate synchronous MCP tool functions with standardized error handling.

Catches libtmux exceptions and re-raises them through _map_exception_to_tool_error() so MCP responses have isError=True with a descriptive message — expected, agent-correctable failures as ExpectedToolError (logged at WARNING), the unexpected catch-all as stock ToolError (logged at ERROR).

The re-raise chains the original exception via from e. Keep it single-level: ReadonlyRetryMiddleware matches libtmux.exc.LibTmuxException by inspecting exactly one __cause__ hop, so wrapping the mapped error again would silently disable readonly retries.

Use handle_tool_errors_async() for async def tools — this wrapper only supports plain sync callables.

Parameters:: fn (Callable[P, R])
Return type:: Callable[P, R]

libtmux_mcp._utils.handle_tool_errors_async(fn) ¶ function[source] function[source] libtmux_mcp._utils.handle_tool_errors_async(fn) ¶

Decorate asynchronous MCP tool functions with standardized error handling.

Async counterpart to handle_tool_errors(). Required for tools that accept a fastmcp.Context parameter because Context’s report_progress/elicit/read_resource methods are coroutines that only run inside async def tools.

Maps the same libtmux exception set to the same messages and error classes as the sync decorator (expected failures as ExpectedToolError at WARNING, the unexpected catch-all as stock ToolError at ERROR) by delegating to a shared helper, and chains the original exception via the same single-level from e that readonly retries depend on.

Parameters:: fn (Callable[P, Coroutine[Any, Any, R]])
Return type:: Callable[P, Coroutine[Any, Any, R]]