mcp2term(该词可能是一个特定领域或项目中的术语,直接翻译为中文可能无具体含义,但可理解为“将MCP(可能指某种模型、条件或参数)转换为术语”或根据上下文具体解释)
一个模型上下文协议(MCP)服务器的实现,该服务器提供安全且可审计的系统shell访问权限。服务器实时流式传输标准输出(stdout)和标准错误(stderr),同时捕获丰富的元数据供插件和下游消费者使用。
特点/功能
- 完整命令执行 具有可配置的外壳、工作目录、环境变量和超时设置。
- 网络直播 通过MCP日志通知来捕获标准输出(stdout)和标准错误(stderr),以便客户端能够实时观察进度。
- 健壮的分块流式传输 该功能能够处理大量标准输出/标准错误信息,而不会出现阻塞或截断情况。
- 插件架构 该机制公开了包中定义的所有函数、类和变量,使得扩展能够观察命令生命周期或注入自定义行为。
- 远程文件管理 工具,允许安全地创建文件、打印、替换行范围、精确查找行以及通过(某种方式)进行统一的差异补丁应用
manage_file工具和filetool客户端命令。 - 自动ngrok隧道建立 因此,HTTP传输无需额外的手动设置即可访问。
- 类型化的寿命上下文 与MCP工具共享,用于依赖项访问和生命周期管理。
- 结构化工具响应 包括时间信息,以便代理能够轻松处理结果。
- 控制台镜像 因此,操作员默认情况下总是在托管终端上看到命令流、标准输出(stdout)和标准错误(stderr)。
- 自动导出启动目录 这个(指令/方法)会为服务器添加前置目录
这句话的开头部分“was started from to”在中文里可能不太直接对应,因为它不是一个完整的句子或短语。但如果我们要尝试翻译其可能的意图,可以将其理解为“始于……到……”,不过这样的翻译需要上下文来确定具体的含义。例如,如果完整句子是“was started from the beginning to the end”,那么翻译就是“从开始到结束”。但在这里,仅凭“was started from to”这个短语,我们无法给出一个确切的翻译,只能提供一个大致的方向或解释 PYTHONPATH 所以通过(某种方式)调用的Python工具 run_command 可以立即解析本地包。
安装
pip install -e .该项目的目标是支持 Python 3.12 或更高版本。
配置
ServerConfig 从环境变量中读取设置:
| 变量 | 描述 | 默认值 |
|---|---|---|
MCP2TERM_SHELL 用于命令的可执行Shell。 | /bin/bash | |
MCP2TERM_WORKDIR | 命令的工作目录。 | 当前目录 |
MCP2TERM_INHERIT_ENV | When(何时) true继承父级环境。 | true |
MCP2TERM_EXTRA_ENV | JSON 对象已合并到命令环境中。 | {} |
MCP2TERM_PLUGINS 以逗号分隔的点号模块路径,用于加载为插件。 | *(无)* | |
MCP2TERM_COMMAND_TIMEOUT | 命令的默认超时时间(以秒为单位)。 | 无限制 |
MCP2TERM_STREAM_CHUNK_SIZE 在流式传输时,每块从标准输出/标准错误读取的字节数。 | 65536 | |
MCP2TERM_LONG_COMMAND_NOTICE_DELAY 在发出长时间运行命令通知之前等待的秒数。 | 2.0 | |
MCP2TERM_LONG_COMMAND_NOTICE_INTERVAL 长时间运行命令通知之间的间隔(以秒为单位)。 | 5.0 | |
MCP2TERM_CONSOLE_ECHO | 镜像命令并将输出到服务器控制台(true/false)。 | true |
MCP2TERM_CHAT_TERMINAL | 设置为 disabled 以抑制控制台集成的消息桥。接受但忽略旧值。 | *(未使用)* |
运行服务器
mcp2term --transport stdio改变 --transport to sse 或者 streamable-http 使用相应的MCP传输方式。 --log-level 控制输出详细程度,并且 --mount-path 在相关情况下,会覆盖HTTP挂载位置。
当服务器运行时,它会将每个执行的命令、标准输出(stdout)数据块和标准错误(stderr)数据块镜像到托管控制台。设置 MCP2TERM_CONSOLE_ECHO=false 在将服务器嵌入到对日志敏感的环境中时,抑制镜像功能。
当与……一起运行时 streamable-http 传输MCP端点由(某处)提供服务 /mcp 路径(或 --mount-path 更多 /mcp 当提供自定义挂载时)。命令行界面(CLI)会打印出完整的URL,包括 /mcp 后缀,用于使像ngrok这样的隧道目标易于复制。
MCP工具
服务器提供了两种用于远程命令管理的工具:
run_command(command: str, working_directory: Optional[str], environment: Optional[dict[str, str]], timeout: Optional[float]], command_id: Optional[str])
该工具返回包含以下内容的结构化JSON:
command_id分配给调用的唯一标识符command执行的命令字符串working_directory解析后的工作目录return_code进程退出代码(非零表示失败)stdout/stderr汇总输出started_at/finished_atISO 8601时间戳duration执行时长(以秒为单位)timed_out布尔标志,指示是否发生了超时
当命令运行时,服务器会以MCP日志消息的形式发出stdout和stderr数据块,并通过异步流保持顺序。客户端可以重用这些消息 command_id 在提出后续请求时考虑相关价值观。
cancel_command(command_id: str, signal_value: Optional[str | int])
发送 cancel_command 转发一个信号(默认为 SIGINT) 到由(某个标识或条件)识别的运行进程 command_id该响应包含数字 signal,其象征意义 signal_name,以及一个 delivered 标志用于确认发送信号时该进程是否仍在运行。
send_stdin(command_id: str, data: Optional[str], eof: bool = False)
使用 send_stdin 向交互式命令流式传输额外输入。该工具接受可选的文本有效载荷和一个 eof 旗帜 一旦所有所需数据都已交付,该程序就会关闭标准输入管道。响应报告输入是否被接受,因此 客户端可以重试或提供有用的诊断信息。
manage_file(path: str, *, operation: str, content: Optional[str] = None, pattern: Optional[str] = None, line: Optional[int] = None, start_line: Optional[int] = None, end_line: Optional[int] = None, encoding: str = "utf-8", create_parents: bool = False, overwrite: bool = False, create_if_missing: bool = True, escape_profile: str = "auto", follow_symlinks: bool = True, use_regex: bool = False, ignore_case: bool = False, max_replacements: Optional[int] = None, anchor: Optional[str] = None, anchor_use_regex: bool = False, anchor_ignore_case: bool = False, anchor_after: bool = False, anchor_occurrence: Optional[int] = None)
manage_file 为……提供动力/能量 filetool 客户端命令,并提供了一系列广泛的、支持行感知的编辑操作 escape_profile 参数控制内联程度 --content 有效载荷在到达服务器之前会被规范化:
auto(默认) 保持原有行为并进行扩展\n,\t,\r,以及\0当有效载荷原本为单行时的序列情况。none禁用所有内联解码,因此有效载荷会原样到达,非常适合对二进制友好的工作流程或当反斜杠具有语义意义时使用。- 可以通过扩展来注册额外的配置文件,以强制执行特定于组织的转义规则。所选配置文件将通过(某种机制)传递给插件
FileOperationEvent有效载荷(或:数据负载)以便可观测性工具能够做出适当响应。
最近的更新为工具箱增加了文件顶部编辑和模式驱动的替换功能:
prepend在文件开头注入内容,并且尊重(原有内容/格式等,具体根据上下文确定)--create-if-missing因此,你可以用一条命令来初始化带有头部信息的全新文件。insert现在支持通过字面量或正则表达式锚点进行匹配--anchor,--anchor-after,--anchor-ignore-case,以及--anchor-occurrence这使得相对于哨兵文本进行更改变得容易,而无需逐行计算。substitute --pattern PATTERN --content TEXT在流式处理结构化元数据时执行基于字面量或正则表达式的替换(匹配模式、替换次数以及标志等)--ignore-case或者--max-replacements)转接回来电者。
示例用法:
# Create a multi-line file from a single-shell command using the default profile.
filetool write docs/roadmap.txt --content 'phase-one\\nphase-two\\nphase-three'
# Append literal escape sequences without rewriting them by selecting the "none" profile.
filetool append docs/roadmap.txt --content 'literal\\nvalue' --escape-profile none
# Use stdin for bulk updates while still labelling the request for plugins.
cat release.diff | filetool patch docs/roadmap.txt --stdin --escape-profile auto插件
插件实现 PluginProtocol (通过模块级别 PLUGIN 对象) 并且可以注册 CommandStreamListener 实例用于观察命令生命周期事件。当服务器启动时,它会加载列表中列出的模块 MCP2TERM_PLUGINS,暴露了整个 mcp2term 通过插件注册表进行命名空间的检查或扩展。
一个最小化的插件框架:
from dataclasses import dataclass
from mcp2term.plugin import CommandStreamListener, PluginProtocol, PluginRegistry
@dataclass
class EchoListener(CommandStreamListener):
async def on_command_stdout(self, event):
print(event.data, end="")
async def on_command_start(self, event):
print(f"Starting: {event.request.command}")
async def on_command_stderr(self, event):
print(f"[stderr] {event.data}", end="")
async def on_command_complete(self, event):
print(f"Finished with {event.return_code}")
class ShellEchoPlugin(PluginProtocol):
name = "shell-echo"
version = "1.0.0"
def activate(self, registry: PluginRegistry):
registry.register_command_listener(EchoListener())
class AuditListener:
async def on_file_operation(self, event):
print(f"{event.operation} {event.path}: {event.result.message}")
registry.register_file_operation_listener(AuditListener())
PLUGIN = ShellEchoPlugin()通过(某种方式)注册的听众 register_file_operation_listener 接收 FileOperationEvent 包含原始请求参数、解析后的路径以及(其他相关信息的)实例 FileOperationResult以及在处理过程中发出的任何警告。这使得它 构建审计、通知或同步插件非常直接 实时响应远程编辑,而无需修改核心服务器。
发展
使用以下命令运行测试套件:
pytest测试被参数化,以便在启用或禁用依赖存根的情况下运行,从而确保所有执行路径都得到验证。
Ngrok 集成
默认情况下 mcp2term 每当你运行服务器时,它就会打开一个ngrok隧道 sse 或者 streamable-http 传输。该隧道使用ngrok代理暴露本地HTTP端点,该代理必须已经过身份验证(例如通过 ngrok config add-authtoken)。 除非被覆盖,否则服务器现在会请求保留域名 alpaca-model-easily.ngrok-free.app 因此,客户端总是能收到一个可预测的主机名。
通过以下环境变量控制集成:
| 变量 | 描述 | 默认值 |
|---|---|---|
MCP2TERM_NGROK_ENABLE | 启用或禁用自动隧道创建。 | true |
MCP2TERM_NGROK_TRANSPORTS | 应该被隧道化的以逗号分隔的传输方式(stdio, sse, streamable-http)。 | sse,streamable-http |
MCP2TERM_NGROK_BIN | 到(某处的)路径 ngrok 可执行文件。 | ngrok |
MCP2TERM_NGROK_API_URL 本地ngrok API的基准URL。 | http://127.0.0.1:4040 | |
MCP2TERM_NGROK_REGION | 可选的 ngrok 区域目标。 | *(无)* |
MCP2TERM_NGROK_LOG_LEVEL | ngrok 日志级别 (debug, info, warn, error)。 | info |
MCP2TERM_NGROK_EXTRA_ARGS 传递给 ngrok 的额外 CLI 参数的 JSON 数组。 | [] | |
MCP2TERM_NGROK_ENV JSON 对象已合并到 ngrok 进程环境中。 | {} | |
MCP2TERM_NGROK_START_TIMEOUT 等待隧道配置的秒数。 | 15 | |
MCP2TERM_NGROK_POLL_INTERVAL | 隧道状态检查之间的间隔时间(秒)。 | 0.5 |
MCP2TERM_NGROK_REQUEST_TIMEOUT API调用的HTTP超时时间。 | 5 | |
MCP2TERM_NGROK_SHUTDOWN_TIMEOUT 等待 ngrok 优雅终止的秒数。 | 5 | |
MCP2TERM_NGROK_CONFIG | 可选的 ngrok 配置文件路径。 | *(无)* |
MCP2TERM_NGROK_HOSTNAME / MCP2TERM_NGROK_DOMAIN / MCP2TERM_NGROK_EDGE 自定义主机绑定以向 ngrok 发出请求。 | alpaca-model-easily.ngrok-free.app 对于领域 |
使用 --disable-ngrok 运行时的标志 mcp2term 对于单次调用,选择不使用隧道技术。 该配置还记录了服务器进程所在的目录 推出并出口至 PYTHONPATH这就像跑步一样 export PYTHONPATH=$(pwd) 在启动服务器之前,以便任何Python代码 通过……执行 run_command 即使在(某种情况下),也继承相同的模块搜索路径 工作目录被覆盖。
