Skip to main content
Glama
jordi-murgo

cliptunnel-mcp

by jordi-murgo

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

架构

Mermaid diagram

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

线格式

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

字段

值

CT1

协议签名 + 版本

from

C(Controller)或 A(Agent)

to

C 或 A

seq

正整数,每个 Controller 会话单调递增

type

C(命令)、R(响应)、E(错误)、A(确认)

payload

Base64 编码的 UTF-8

安装

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

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

二进制文件

额外依赖

用途

cliptunnel-agent

(无)

在本地操作系统剪贴板上运行 Agent。

cliptunnel-mcp

[server]

通过 stdio 运行 MCP 服务器。

快速开始

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 监视剪贴板插槽,确认命令,在工作池中处理它们,并写回响应。按 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。

方法

描述

send_command(command: str) -> Future

排队命令;返回一个 Future,在成功时解析为响应负载,失败时解析为 None。

send_command_sync(command: str) -> str | None

发送并阻塞直到响应或 timeout 秒。

close()

停止后台线程。幂等。

构造函数参数:transport(必需)、timeout、retries、poll_interval、ack_timeout、initial_seq、persist_seq、seq_store。

Agent

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

方法

描述

close()

停止此 agent 生成。幂等;绝不遗留线程。

构造函数参数: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"}')

协议原语

符号

描述

pack(msg) -> str

将 Message 序列化为线格式。

unpack(raw) -> Message | None

解析线字符串;输入格式错误时返回 None。

validate(raw, my_role) -> bool

如果 raw 格式正确且寻址到 my_role,则返回 True。

Message

数据类:frm、to、seq、mtype、payload。

MsgType

枚举:COMMAND、RESPONSE、ERROR、ACK。

Role

枚举:CONTROLLER、AGENT。

SeqTracker

每个序列的去重状态:新 → 处理中 → 完成。

传输协议

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 处理程序支持以下操作:

操作

参数

返回

shell

cmd

JSON: {stdout, stderr, returncode}

fs.read

path

JSON: {content, lines}

fs.write

path, content

wrote N bytes to PATH

fs.list

path

JSON: [{name, size, is_dir}]

fs.delete

path

deleted PATH

fs.replace

path, old, new

replaced 1 occurrence in PATH(精确一次匹配)

fs.search

path, pattern

JSON: [{line, content}](正则)

fs.find

path, pattern

JSON: [PATH, ...](glob,** 递归)

fs.bin_read

path

JSON: {path, size, b64}

fs.bin_write

path, b64

wrote N bytes to PATH

MCP 工具

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

工具

描述

remote_shell

执行 shell 命令;自动同步(10 秒)然后异步,使用 job_id 轮询。

remote_shell_result

轮询异步 shell 命令的结果。

remote_fs_read

读取文件。

remote_fs_write

创建或覆盖文件(创建父目录)。

remote_fs_list

列出目录条目。

remote_fs_delete

删除文件。

remote_fs_replace

在文件中搜索并替换(精确一次匹配)。

remote_fs_search

在文件中进行正则搜索。

remote_fs_find

在目录下使用 glob 查找文件。

remote_fs_bin_read

以 base64 读取二进制文件。

remote_fs_bin_write

将 base64 内容写入二进制文件。

remote_upload

将本地文件上传到远程机器。

remote_download

将远程文件下载到本地机器。

生命周期与合并语义

  • 一次一个命令:Controller 串行分发命令。待处理命令的 seq 与槽位写入原子地一起发布,因此读取方绝不会在分发器之前观察到该命令。

  • 立即 ACK:Agent 在处理每条命令之前先进行 ACK,从而为 Controller 释放槽位。

  • 一次一个响应:Agent 只持有一个待处理的响应信封。新命令绝不会隐式 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 自带 ClipboardTransport,这是一个由操作系统剪贴板支撑的传输层。在 Wayland 上,它使用 wl-paste --watch 进行事件驱动的变更检测(空闲时零轮询、零 CPU)。在 macOS、Windows 和 X11 上,它每 100 毫秒轮询一次,并使用基于哈希的变更检测。它同时实现了 Transport 和 RevisionMonitor,因此两个端点都能获得感知变更的等待。二进制文件 cliptunnel-agent 和 cliptunnel-mcp 会自动使用它。

对于自定义设置——例如 Citrix 剪贴板重定向、共享 Gist、网络管道——请实现 Transport 协议(read() -> str、write(str) -> None),并可选择实现 RevisionMonitor(revision + wait_for_change)。直接将其注入 Controller 或 Agent。

平台支持

平台

状态

剪贴板后端

变更检测

macOS

已测试

pbcopy/pbpaste(内置)

轮询(100 毫秒)

Windows

已测试

ctypes + user32(无额外依赖)

轮询(100 毫秒)

Linux / Wayland

已测试

wl-copy/wl-paste(wl-clipboard 包)

事件驱动

Linux / X11

核心可用

xclip(备选:xsel)

轮询(100 毫秒)

开发

# 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

测试套件使用确定性的 ClipboardSlot 测试替身,它模拟了带修订和有限等待的“最后写入者胜出”通道。无需剪贴板硬件。

限制

  • 仅文本剪贴板:协议承载 UTF-8 字符串。二进制文件会进行 base64 编码,这会使在线传输大小大约翻倍。

  • 单一槽位:剪贴板一次只能保存一个值。ARQ 协议通过它串行化所有流量,因此吞吐量受剪贴板往返延迟的限制。

  • 无加密:线上格式是纯 base64。如果剪贴板可被观察,请在传输层或处理器中使用加密层。

许可证

MIT — 参见 LICENSE.

Related MCP Connectors

Related MCP Servers