Compare commits
5 Commits
b9f686a2ef
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| d90614a387 | |||
| d00259bf5a | |||
| 589541313d | |||
| 384c4ff70e | |||
| 6e41cd1f07 |
2
.gitignore
vendored
Normal file
2
.gitignore
vendored
Normal file
@@ -0,0 +1,2 @@
|
|||||||
|
# JetBrains IDEs
|
||||||
|
.idea/
|
||||||
169
README.md
169
README.md
@@ -0,0 +1,169 @@
|
|||||||
|
# H3C CLI MCP Server
|
||||||
|
[English](#English) | [中文](#中文)
|
||||||
|
|
||||||
|
## English
|
||||||
|
A Telnet-based MCP tool for H3C network device CLI, optimized for H3C IOS devices.
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
- **Smart Connection Management**: Auto terminal activation, TCP warm-up, auto disable pagination
|
||||||
|
- **Device Mode Detection**: Auto detect user mode, privileged mode, config mode
|
||||||
|
- **Smart Wait Mechanism**: Returns immediately when device prompt detected
|
||||||
|
- **Long-running Command Optimization**: Auto detect ping, traceroute and extend wait time
|
||||||
|
|
||||||
|
|
||||||
|
### Tools
|
||||||
|
|
||||||
|
- **telnet_connect**: Establish Telnet connection, returns session ID and device mode
|
||||||
|
- **telnet_execute**: Execute command in session, returns output and device mode
|
||||||
|
- **telnet_list_sessions**: List all active sessions
|
||||||
|
- **telnet_disconnect**: Disconnect session
|
||||||
|
|
||||||
|
### Installation
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git clone https://git.youdu.xin/wxy/h3c-cli-mcp.git
|
||||||
|
cd h3c-cli-mcp
|
||||||
|
pip install .
|
||||||
|
```
|
||||||
|
|
||||||
|
### Usage
|
||||||
|
|
||||||
|
#### Run as MCP Server
|
||||||
|
|
||||||
|
```bash
|
||||||
|
h3c-cli-mcp
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Configure MCP Client
|
||||||
|
|
||||||
|
Add to MCP client config:
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"mcpServers": {
|
||||||
|
"h3c-cli-mcp": {
|
||||||
|
"command": "h3c-cli-mcp"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
### System Prompt
|
||||||
|
```text
|
||||||
|
You are a network operations assistant capable of remotely operating network device CLIs via the h3c-cli-mcp tool.
|
||||||
|
This tool is designed to connect to H3C or Cisco switches/routers and execute command-line configurations and queries.
|
||||||
|
|
||||||
|
Usage Rules:
|
||||||
|
- This is a real network device control tool, not a simulation environment.
|
||||||
|
- Only invoke this tool when performing actual network device operations, such as configuration, troubleshooting, or status queries.
|
||||||
|
- Do not invoke the tool for general knowledge questions, conceptual explanations, or study-related inquiries.
|
||||||
|
|
||||||
|
Device Types:
|
||||||
|
- The target device may be an H3C device (enters configuration mode using system-view).
|
||||||
|
- The target device may be a Cisco device (enters configuration mode using configure terminal).
|
||||||
|
- Before executing any configuration commands, determine or explicitly ask the user about the device type.
|
||||||
|
|
||||||
|
Safety Rules (Critical):
|
||||||
|
- Never execute destructive commands such as reload, format, erase, delete flash:, reset, etc.
|
||||||
|
- Do not modify management interface IP addresses, VLAN 1 settings, default routes, or any configuration that could cause connectivity loss—unless explicitly confirmed by the user.
|
||||||
|
- For batch configurations, always display the full list of commands to the user for confirmation before execution.
|
||||||
|
|
||||||
|
Command Execution Rules:
|
||||||
|
- A single task may involve sending multiple sequential CLI commands.
|
||||||
|
- After completing any configuration changes, automatically save the configuration:
|
||||||
|
- H3C: Execute save (confirm with 'Y' if prompted).
|
||||||
|
- Cisco: Execute write memory.
|
||||||
|
|
||||||
|
Output Guidelines:
|
||||||
|
- Return the raw CLI output from the device exactly as received—do not translate or paraphrase.
|
||||||
|
- If a command fails, return the complete error message from the device.
|
||||||
|
- Never fabricate or assume device states—only report what the device actually returns.
|
||||||
|
|
||||||
|
Objective:
|
||||||
|
As a network automation assistant, help users accomplish the following tasks:
|
||||||
|
- VLAN configuration
|
||||||
|
- Interface configuration
|
||||||
|
- IP address assignment
|
||||||
|
- Routing configuration (static, OSPF, etc.)
|
||||||
|
- ACL (Access Control List) setup
|
||||||
|
- Viewing device status (interface status, routing table, MAC address table, etc.)
|
||||||
|
```
|
||||||
|
|
||||||
|
## 中文
|
||||||
|
一款基于 Telnet 的 MCP 工具,专为 H3C 网络设备 CLI 设计,针对 H3C Comware(类 IOS)设备进行了优化。
|
||||||
|
|
||||||
|
### 功能特性
|
||||||
|
|
||||||
|
智能连接管理:自动激活终端、TCP 预热、自动关闭分页显示
|
||||||
|
设备模式识别:自动检测用户模式、特权模式、配置模式
|
||||||
|
智能等待机制:检测到设备提示符后立即返回,无需等待超时
|
||||||
|
长时命令优化:自动识别 ping、traceroute 等命令并延长等待时间
|
||||||
|
|
||||||
|
### 工具列表
|
||||||
|
|
||||||
|
telnet_connect:建立 Telnet 连接,返回会话 ID 和设备当前模式
|
||||||
|
telnet_execute:在指定会话中执行命令,返回输出结果和设备模式
|
||||||
|
telnet_list_sessions:列出所有活跃会话
|
||||||
|
telnet_disconnect:断开指定会话
|
||||||
|
|
||||||
|
### 安装方法
|
||||||
|
```bash
|
||||||
|
|
||||||
|
git clone https://git.youdu.xin/wxy/h3c-cli-mcp.git
|
||||||
|
cd h3c-cli-mcp
|
||||||
|
pip install .
|
||||||
|
```
|
||||||
|
|
||||||
|
### 使用方式
|
||||||
|
作为 MCP 服务器运行
|
||||||
|
```bash
|
||||||
|
|
||||||
|
h3c-cli-mcp
|
||||||
|
```
|
||||||
|
|
||||||
|
### 在 MCP 客户端中配置
|
||||||
|
将以下内容添加到 MCP 客户端配置文件中:
|
||||||
|
```json
|
||||||
|
|
||||||
|
{
|
||||||
|
"mcpServers": {
|
||||||
|
"h3c-cli-mcp": {
|
||||||
|
"command": "h3c-cli-mcp"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
系统提示词
|
||||||
|
```text
|
||||||
|
你是一个网络运维助手,可以通过 h3c-cli-mcp 工具远程操作网络设备 CLI。
|
||||||
|
该工具用于连接 H3C 或 Cisco 交换机/路由器,并执行命令行配置与查询。
|
||||||
|
使用规则:
|
||||||
|
这是一个真实网络设备控制工具,不是模拟环境。
|
||||||
|
仅在涉及网络设备操作、配置、排错、状态查询时才调用该工具。
|
||||||
|
普通知识问答、原理解释、学习类问题,不要调用工具。
|
||||||
|
设备类型:
|
||||||
|
可能是 H3C 设备(使用 system-view)
|
||||||
|
可能是 Cisco 设备(使用 configure terminal)
|
||||||
|
在执行配置前,应先判断或询问设备类型
|
||||||
|
安全规则(非常重要):
|
||||||
|
不要执行 reload、format、erase、delete flash、reset 等破坏性命令
|
||||||
|
不要修改管理口 IP、VLAN1、默认路由等可能导致断连的配置,除非用户明确确认
|
||||||
|
批量配置前应先展示命令内容给用户确认
|
||||||
|
命令执行规则:
|
||||||
|
一次任务可以发送多条连续命令
|
||||||
|
配置完成后应自动执行保存配置操作
|
||||||
|
H3C: save
|
||||||
|
Cisco: write memory
|
||||||
|
输出规范:
|
||||||
|
设备回显是原始 CLI 输出,不需要翻译
|
||||||
|
如命令执行失败,应把错误信息完整返回
|
||||||
|
不要编造设备状态
|
||||||
|
目标:
|
||||||
|
作为网络自动化助手,帮助用户完成:
|
||||||
|
VLAN 配置
|
||||||
|
接口配置
|
||||||
|
IP 地址配置
|
||||||
|
路由配置
|
||||||
|
ACL 配置
|
||||||
|
查看设备状态(接口、路由表、MAC 表等)
|
||||||
|
```
|
||||||
@@ -0,0 +1,215 @@
|
|||||||
|
"""
|
||||||
|
Multi-Vendor Telnet MCP Server
|
||||||
|
Supports Cisco IOS & H3C Comware devices
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import uuid
|
||||||
|
import re
|
||||||
|
import json
|
||||||
|
from datetime import datetime
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
import telnetlib3
|
||||||
|
from mcp.server.fastmcp import FastMCP
|
||||||
|
|
||||||
|
mcp = FastMCP("Multi-Vendor CLI MCP Server")
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# 🔍 提示符识别(核心:厂商判断 + 模式识别)
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
def detect_vendor_and_mode(output: str) -> tuple[str, str]:
|
||||||
|
if not output:
|
||||||
|
return "unknown", "unknown"
|
||||||
|
|
||||||
|
clean = output.replace('\x08', '').replace(' \b', '')
|
||||||
|
|
||||||
|
# H3C
|
||||||
|
h3c_match = re.search(r'[\r\n](\[[^\]]+\]|<[^>]+>)\s*$', clean)
|
||||||
|
if h3c_match:
|
||||||
|
return "h3c", h3c_match.group(1)
|
||||||
|
|
||||||
|
# Cisco
|
||||||
|
cisco_match = re.search(
|
||||||
|
r'[\r\n]([A-Za-z0-9._-]+(\([a-z0-9-]+\))?[#>])\s*$',
|
||||||
|
clean
|
||||||
|
)
|
||||||
|
if cisco_match:
|
||||||
|
return "cisco", cisco_match.group(1)
|
||||||
|
|
||||||
|
return "unknown", "unknown"
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# 🧩 会话模型
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class TelnetSession:
|
||||||
|
session_id: str
|
||||||
|
host: str
|
||||||
|
port: int
|
||||||
|
reader: telnetlib3.TelnetReader
|
||||||
|
writer: telnetlib3.TelnetWriter
|
||||||
|
vendor: str = "unknown"
|
||||||
|
connected_at: datetime = field(default_factory=datetime.now)
|
||||||
|
|
||||||
|
def to_dict(self):
|
||||||
|
return {
|
||||||
|
"sessionId": self.session_id,
|
||||||
|
"host": self.host,
|
||||||
|
"port": self.port,
|
||||||
|
"vendor": self.vendor,
|
||||||
|
"connectedAt": self.connected_at.isoformat(),
|
||||||
|
}
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# 🧠 会话管理器(多厂商核心)
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
class TelnetSessionManager:
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.sessions: dict[str, TelnetSession] = {}
|
||||||
|
|
||||||
|
async def connect(self, host: str, port: int, timeout: int = 5000) -> str:
|
||||||
|
reader, writer = await asyncio.wait_for(
|
||||||
|
telnetlib3.open_connection(host, port),
|
||||||
|
timeout=timeout / 1000.0
|
||||||
|
)
|
||||||
|
|
||||||
|
session_id = str(uuid.uuid4())
|
||||||
|
session = TelnetSession(session_id, host, port, reader, writer)
|
||||||
|
self.sessions[session_id] = session
|
||||||
|
|
||||||
|
# 激活终端
|
||||||
|
for _ in range(3):
|
||||||
|
writer.write("\r\n")
|
||||||
|
await writer.drain()
|
||||||
|
await asyncio.sleep(0.2)
|
||||||
|
|
||||||
|
await self._drain(reader)
|
||||||
|
|
||||||
|
# 识别厂商
|
||||||
|
writer.write("\r\n")
|
||||||
|
await writer.drain()
|
||||||
|
await asyncio.sleep(0.3)
|
||||||
|
output = await self._read_quick(reader)
|
||||||
|
vendor, _ = detect_vendor_and_mode(output)
|
||||||
|
session.vendor = vendor
|
||||||
|
|
||||||
|
# 厂商初始化
|
||||||
|
if vendor == "cisco":
|
||||||
|
writer.write("terminal length 0\r\n")
|
||||||
|
elif vendor == "h3c":
|
||||||
|
writer.write("screen-length 0 temporary\r\n")
|
||||||
|
|
||||||
|
await writer.drain()
|
||||||
|
await asyncio.sleep(0.3)
|
||||||
|
await self._drain(reader)
|
||||||
|
|
||||||
|
return session_id
|
||||||
|
|
||||||
|
async def execute(self, session_id: str, command: str, wait_ms: int = 3000) -> str:
|
||||||
|
session = self.sessions[session_id]
|
||||||
|
session.writer.write(command + "\r\n")
|
||||||
|
await session.writer.drain()
|
||||||
|
|
||||||
|
output = ""
|
||||||
|
start = asyncio.get_event_loop().time()
|
||||||
|
|
||||||
|
if session.vendor == "h3c":
|
||||||
|
prompt_pattern = re.compile(r'[\r\n](\[[^\]]+\]|<[^>]+>)\s*$')
|
||||||
|
else:
|
||||||
|
prompt_pattern = re.compile(r'[\r\n]([A-Za-z0-9._-]+(\([a-z0-9-]+\))?[#>])\s*$')
|
||||||
|
|
||||||
|
while True:
|
||||||
|
if asyncio.get_event_loop().time() - start > wait_ms / 1000:
|
||||||
|
break
|
||||||
|
try:
|
||||||
|
data = await asyncio.wait_for(session.reader.read(4096), timeout=0.2)
|
||||||
|
if data:
|
||||||
|
output += data
|
||||||
|
clean = output.replace('\x08', '').replace(' \b', '')
|
||||||
|
if prompt_pattern.search(clean):
|
||||||
|
await asyncio.sleep(0.2)
|
||||||
|
break
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
return output
|
||||||
|
|
||||||
|
async def disconnect(self, session_id: str):
|
||||||
|
session = self.sessions.pop(session_id)
|
||||||
|
session.writer.close()
|
||||||
|
|
||||||
|
def list_sessions(self):
|
||||||
|
return [s.to_dict() for s in self.sessions.values()]
|
||||||
|
|
||||||
|
async def _drain(self, reader):
|
||||||
|
try:
|
||||||
|
while True:
|
||||||
|
await asyncio.wait_for(reader.read(4096), timeout=0.1)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def _read_quick(self, reader):
|
||||||
|
out = ""
|
||||||
|
try:
|
||||||
|
while True:
|
||||||
|
out += await asyncio.wait_for(reader.read(4096), timeout=0.2)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
pass
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
session_manager = TelnetSessionManager()
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# 🛠 MCP 工具
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
async def telnet_connect(host: str, port: int) -> str:
|
||||||
|
session_id = await session_manager.connect(host, port)
|
||||||
|
output = await session_manager.execute(session_id, "", 1000)
|
||||||
|
vendor, mode = detect_vendor_and_mode(output)
|
||||||
|
|
||||||
|
return json.dumps({
|
||||||
|
"success": True,
|
||||||
|
"sessionId": session_id,
|
||||||
|
"vendor": vendor,
|
||||||
|
"deviceMode": mode
|
||||||
|
}, ensure_ascii=False)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
async def telnet_execute(session_id: str, command: str) -> str:
|
||||||
|
output = await session_manager.execute(session_id, command)
|
||||||
|
vendor, mode = detect_vendor_and_mode(output)
|
||||||
|
|
||||||
|
return json.dumps({
|
||||||
|
"success": True,
|
||||||
|
"output": output,
|
||||||
|
"vendor": vendor,
|
||||||
|
"deviceMode": mode
|
||||||
|
}, ensure_ascii=False)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
def telnet_list_sessions() -> str:
|
||||||
|
return json.dumps(session_manager.list_sessions(), ensure_ascii=False)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
async def telnet_disconnect(session_id: str) -> str:
|
||||||
|
await session_manager.disconnect(session_id)
|
||||||
|
return f"Session {session_id} disconnected"
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
mcp.run()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|||||||
@@ -0,0 +1,7 @@
|
|||||||
|
[project]
|
||||||
|
name = "h3c-cli-mcp"
|
||||||
|
version = "1.0.0"
|
||||||
|
dependencies = ["mcp", "telnetlib3"]
|
||||||
|
|
||||||
|
[project.scripts]
|
||||||
|
h3c-cli-mcp = "h3c_cli_mcp.server:main"
|
||||||
|
|||||||
Reference in New Issue
Block a user