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