cliptunnel-mcp

通过剪贴板远程操作锁定的机器。

功能

cliptunnel-mcp 将共享剪贴板转变为两台机器之间可靠的控制通道。当远程机器位于 Citrix 会话、锁定的 VDI 或任何阻止 SSH、文件传输和网络但仍暴露剪贴板的环境中时，ClipTunnel 通过该单一通道传输命令，并将其作为 Model Context Protocol 工具暴露出来。

该包包含三层：

• 协议 — 一种线上格式（CT1），包含 base64 负载、序列号和类型化消息（命令、响应、错误、确认）。

• 端点 — Controller（操作方）和 Agent（远程方），通过注入的 Transport 连接。两者都运行后台线程，具有 ARQ 重传、基于序列号的去重和代际安全生命周期。

• MCP 服务器 — 一个 FastMCP 应用程序，通过 stdio 将 Controller 的辅助函数暴露为 remote_shell、remote_fs_*、remote_upload 和 remote_download 工具。

核心包零依赖。MCP 服务器需要可选的 [server] 额外依赖（mcp>=1.2,<2）。

Related MCP server: Sky Windows Remote Executor

架构

两端共享一个最后写入者胜出的剪贴板槽。该协议使用停止等待 ARQ：Controller 写入一条命令，Agent 立即 ACK，在工作池中处理命令，然后写入一条类型化响应（R 或 E），并重传直到 Controller 的匹配 ACK 到达。Controller 一次发送一条命令，并在响应返回时解析 future。

线上格式

CT1|<from>|<to>|<seq>|<type>|<payload>

安装

pip install cliptunnel-mcp          # core + cliptunnel-agent binary
pip install cliptunnel-mcp[server]  # adds cliptunnel-mcp server binary (mcp>=1.2,<2)

两种模式都安装控制台入口点：

快速开始

Agent（远程机器）

运行 Agent 的最简方式是使用已安装的二进制文件：

cliptunnel-agent

防病毒 / EDR 变通方案（Windows）：未签名的 .exe 入口点可能被隔离。请改用 python -m — 它通过已受信任的 Python 解释器运行，无需生成二进制文件：

python -m cliptunnel_mcp.agent    # instead of cliptunnel-agent
python -m cliptunnel_mcp.server   # instead of cliptunnel-mcp

这会构建一个由系统剪贴板支持的 ClipboardTransport（macOS 上为 pbcopy/pbpaste，Windows 上为 user32，Wayland 上为 wl-copy/wl-paste，X11 上为 xclip/xsel），并将 operations.dispatch 作为命令处理器。Agent 监视剪贴板通道，ACK 命令，在工作池中处理它们，并写回响应。按 Ctrl+C 停止。

Controller + MCP 服务器（操作方机器）

在操作方，配置您的 MCP 客户端（Claude Desktop、Cursor、Pi 等）以启动服务器二进制：

{
  "mcpServers": {
    "cliptunnel": {
      "command": "cliptunnel-mcp",
      "args": []
    }
  }
}

如果 cliptunnel-mcp 二进制被防病毒软件阻止，请使用 python -m：

{
  "mcpServers": {
    "cliptunnel": {
      "command": "python",
      "args": ["-m", "cliptunnel_mcp.server"]
    }
  }
}

服务器二进制文件注入一个由 ClipboardTransport 支持的 Controller，并通过 stdio 运行 FastMCP 应用程序。所有 remote_* 工具立即可用。

注意：MCP 服务器需要 pip install cliptunnel-mcp[server]。

仅 Controller（无 MCP）

用于无 MCP 客户端的编程使用：

from cliptunnel_mcp.clipboard_transport import ClipboardTransport
from cliptunnel_mcp import Controller
import json

controller = Controller(transport=ClipboardTransport())

# Async — returns a Future
future = controller.send_command(json.dumps({"op": "shell", "cmd": "whoami"}))
result = future.result(timeout=30)

# Sync — blocks until response or timeout
output = controller.send_command_sync(json.dumps({"op": "fs.read", "path": "/etc/hostname"}))

编程式 Agent

如果您需要自定义处理器或传输：

from cliptunnel_mcp.clipboard_transport import ClipboardTransport
from cliptunnel_mcp import Agent
from cliptunnel_mcp.operations import dispatch

agent = Agent(transport=ClipboardTransport(), handler=dispatch)
# Blocks until agent.close() — run in a thread or manage lifecycle yourself.

API 接口

Controller

操作方端点。异步发送命令，一次调度一个，并在响应到达时解析 future。

构造函数参数：transport（必需）、timeout、retries、poll_interval、ack_timeout、initial_seq、persist_seq、seq_store。

Agent

远程端端点。监视槽，立即 ACK 命令，在工作池中处理它们，并一次写入一条类型化响应并重传。

构造函数参数：transport（必需）、handler（必需）、poll_interval、max_workers、response_ack_timeout。

dispatch

默认 Agent 处理器。解析 JSON 负载并路由到匹配的操作。

from cliptunnel_mcp.operations import dispatch

output, is_error = dispatch('{"op": "shell", "cmd": "echo hello"}')

协议原语

传输协议

class Transport(Protocol):
    def read(self) -> str: ...
    def write(self, value: str) -> None: ...

class RevisionMonitor(Protocol):
    @property
    def revision(self) -> int: ...
    def wait_for_change(self, after: int, timeout: float = 1.0) -> int: ...

传输必须实现 read/write（最后写入者胜出）。实现 RevisionMonitor（或暴露 wait_for_revision / wait_for_change）可启用变更感知等待，而不是轮询。

操作

dispatch 处理器支持以下操作：

MCP 工具

服务器通过 stdio 暴露 13 个工具：

生命周期与关闭语义

• 一度に1つのコマンド: Controllerはコマンドを直列にディスパッチします。保留中のコマンドのseqはスロット書き込みと原子的に公開されるため、リーダーがディスパッチャーより先にコマンドを観測することはありません。

• 即時ACK: Agentは処理前にすべてのコマンドにACKを返し、Controllerのためにスロットを解放します。

• 一度に1つのレスポンス: Agentは保留中のレスポンスエンベロープを正確に1つ保持します。新しいコマンドが保留中のレスポンスを暗黙的にACKすることはありません — Controllerの一致するA(seq)だけがそれを解放します。

• 再送: 両側ともACKタイムアウト時に再送します。Controllerはretries回（デフォルト3回）まで再試行します。Agentはresponse_ack_timeout秒（デフォルト1.0）ごとにレスポンスを再送します。

• 重複排除: AgentのSeqTrackerはseqごとの状態（new → processing → done）を追跡します。重複コマンドはACKされ、完了済みのものはキャッシュされた型付きレスポンスを再生し、処理中のものはすでに処理されています。

• 古いメッセージのガード: Controllerはseq <= min_seqのR/Eをスキップします — 前のセッションからの古いスロット内容です。

• 世代セーフ: すべての停止状態とキューは各インスタンスにローカルです。AgentまたはControllerを閉じて新しいものを作成しても、スレッドが取り残されることはありません。

• ペース調整された書き込み: Controllerは境界のある書き込み間隔（ポーリング間隔の2倍）を強制するため、Agentは各メッセージが上書きされる前に読み取ることができます。

バックエンドの選択

ClipTunnelはOSクリップボードをバックエンドとするClipboardTransportを同梱しています。Waylandではwl-paste --watchを使用してイベント駆動型の変更検出を行います（アイドル時のポーリングゼロ、CPUゼロ）。macOS、Windows、X11では100msごとにハッシュベースの変更検出でポーリングします。TransportとRevisionMonitorの両方を実装しているため、両エンドポイントで変更認識型の待機が可能です。バイナリcliptunnel-agentとcliptunnel-mcpは自動的にこれを使用します。

カスタムセットアップ（Citrixクリップボードリダイレクション、共有Gist、ネットワークパイプ）の場合は、Transportプロトコル（read() -> str、write(str) -> None）と、必要に応じてRevisionMonitor（revision + wait_for_change）を実装します。それをControllerまたはAgentに直接注入します。

プラットフォームサポート

開発

# Create a virtual environment
uv venv && source .venv/bin/activate

# Install in development mode
uv pip install -e . pytest

# Run the test suite (161 tests)
python -m pytest -q
# or
python -m unittest discover -s tests -t .

# Bare mode — no install, just PYTHONPATH
PYTHONPATH=src:. python -m pytest -q

テストスイートは、リビジョンと境界のある待機を備えたlast-writer-winsチャネルをモデル化した決定的なClipboardSlotテストダブルを使用します。クリップボードハードウェアは不要です。

制限事項

• テキストのみのクリップボード: プロトコルはUTF-8文字列を運びます。バイナリファイルはbase64エンコードされるため、ワイヤー上でサイズが約2倍になります。

• 単一スロット: クリップボードは一度に1つの値を保持します。ARQプロトコルはすべてのトラフィックをスロット経由で直列化するため、スループットはクリップボードのラウンドトリップレイテンシによって制限されます。

• 暗号化なし: ワイヤーフォーマットはプレーンなbase64です。クリップボードが観測可能な場合は、トランスポートまたはハンドラーに暗号化レイヤーを使用してください。

ライセンス

MIT — LICENSEを参照してください。