v6ole/NetBrain_MCP

Fork 0

Files

T

Koreyoshi 3236fa947d Add files via upload

2026-05-22 12:17:24 +08:00

18 KiB

Raw Permalink Blame History

NetBrain MCP 项目教程

本教程详细介绍如何使用NetBrain MCP项目，包括MCP协议的基本概念、实际实现方式和高级使用技巧。

1. MCP协议概述

1.1 什么是MCP？

MCP（Model Context Protocol）是一个开放标准，用于连接LLM（大型语言模型）应用程序与外部数据源和工具。可以将MCP视为"AI领域的USB-C"——一种通用连接器，使AI模型能够与外部工具或数据源建立安全的双向连接。

1.2 NetBrain MCP的实现特点

NetBrain MCP基于FastMCP框架实现，具有以下特点：

32个MCP工具：涵盖设备管理、连接控制、拓扑发现、网络扫描等
13个MCP资源：提供设备、配置、拓扑、扫描等数据资源
13个提示模板：专业的网络运维AI提示模板
混合异步/同步模式：智能适配不同类型的设备连接
多厂商支持：思科、华为、H3C、Juniper等主流设备

2. MCP核心概念

2.1 工具（Tools）

工具是模型控制的函数，允许LLM执行动作和产生副作用。

2.1.1 NetBrain MCP工具分类

设备管理工具（6个）

@mcp.tool()
async def list_devices(vendor: str = None, device_type: str = None, status: str = None, tag: str = None) -> List[Dict[str, Any]]:
    """列出网络设备，支持过滤条件"""

@mcp.tool()  
async def add_device(name: str, ip_address: str, device_type: str, vendor: str, platform: str = None, ...) -> Dict[str, Any]:
    """添加新的网络设备"""

@mcp.tool()
async def get_device(device_id: str) -> Optional[Dict[str, Any]]:
    """获取设备详细信息"""

@mcp.tool()
async def update_device(device_id: str, **kwargs) -> Optional[Dict[str, Any]]:
    """更新设备信息"""

@mcp.tool()
async def delete_device(device_id: str) -> bool:
    """删除设备"""

连接管理工具（5个）

@mcp.tool()
async def connect_device(device_id: str, credential_id: str) -> Dict[str, Any]:
    """连接到网络设备"""

@mcp.tool()
async def disconnect_device(device_id: str, credential_id: str) -> Dict[str, Any]:
    """断开设备连接"""

@mcp.tool()
async def send_command(device_id: str, credential_id: str, command: str, timeout: int = 30) -> Dict[str, Any]:
    """发送单个命令到设备"""

@mcp.tool()
async def send_commands(device_id: str, credential_id: str, commands: List[str], timeout: int = 30) -> List[Dict[str, Any]]:
    """发送多个命令到设备"""

@mcp.tool()
async def get_active_connections() -> List[Dict[str, Any]]:
    """获取当前活动的设备连接"""

拓扑发现工具（6个）

@mcp.tool()
async def discover_topology(device_ids: List[str]) -> Dict[str, Any]:
    """发现网络拓扑结构"""

@mcp.tool()
async def get_topology() -> Dict[str, Any]:
    """获取当前网络拓扑"""

@mcp.tool()
async def clear_topology() -> Dict[str, Any]:
    """清除拓扑缓存"""

@mcp.tool()
async def get_device_neighbors(device_id: str) -> Dict[str, Any]:
    """获取设备的邻居信息"""

@mcp.tool()
async def discover_device_neighbors(device_id: str) -> Dict[str, Any]:
    """发现设备邻居"""

@mcp.tool()
async def get_topology_statistics() -> Dict[str, Any]:
    """获取拓扑统计信息"""

2.1.2 工具实现方式

NetBrain MCP使用FastMCP的装饰器方式注册工具：

# 在server.py中的实际实现
from mcp.server.fastmcp import FastMCP
from network_devices import device_manager
from device_connector import connection_manager

# 创建FastMCP服务器实例
mcp = FastMCP("NetBrain MCP")

@mcp.tool()
async def add_device(
    name: str,
    ip_address: str, 
    device_type: str,
    vendor: str,
    platform: str = None,
    model: str = None,
    os_version: str = None,
    location: str = None,
    description: str = None
) -> Dict[str, Any]:
    """添加新的网络设备到系统中"""
    try:
        # 创建设备对象
        device = NetworkDevice(
            name=name,
            ip_address=ip_address,
            device_type=DeviceType(device_type),
            vendor=DeviceVendor(vendor),
            platform=platform or "",
            model=model or "",
            os_version=os_version or "",
            location=location or "",
            description=description or ""
        )
        
        # 添加设备
        device_id = device_manager.add_device(device)
        
        return {
            "success": True,
            "device_id": device_id,
            "message": f"设备 {name} 添加成功"
        }
        
            except Exception as e:
        logger.error(f"添加设备失败: {str(e)}")
        return {
            "success": False,
            "error": str(e)
        }

2.2 资源（Resources）

资源是应用程序控制的数据对象，通过URI标识，用于为LLM提供上下文信息。

2.2.1 NetBrain MCP资源列表

系统实现了13个MCP资源：

设备相关资源（5个）

@resource_manager.register_resource("device/{device_id}", ResourceType.DEVICE)
async def get_device_resource(device_id: str) -> Dict[str, Any]:
    """获取设备基本信息"""

@resource_manager.register_resource("device/{device_id}/config", ResourceType.CONFIG)
async def get_device_config_resource(device_id: str) -> Dict[str, Any]:
    """获取设备配置信息"""

@resource_manager.register_resource("device/{device_id}/interfaces", ResourceType.DEVICE)
async def get_device_interfaces_resource(device_id: str) -> Dict[str, Any]:
    """获取设备接口信息"""

@resource_manager.register_resource("device/{device_id}/routes", ResourceType.DEVICE)
async def get_device_routes_resource(device_id: str) -> Dict[str, Any]:
    """获取设备路由信息"""

@resource_manager.register_resource("device/{device_id}/neighbors", ResourceType.TOPOLOGY)
async def get_device_neighbors_resource(device_id: str) -> Dict[str, Any]:
    """获取设备邻居信息"""

系统资源（4个）

@resource_manager.register_resource("greeting/{name}", ResourceType.SYSTEM)
async def get_greeting_resource(name: str) -> Dict[str, Any]:
    """问候信息资源（示例）"""

@resource_manager.register_resource("credentials", ResourceType.CREDENTIAL)
async def get_credentials_resource() -> Dict[str, Any]:
    """获取凭据列表"""

@resource_manager.register_resource("system/status", ResourceType.SYSTEM)
async def get_system_status_resource() -> Dict[str, Any]:
    """获取系统状态"""

@resource_manager.register_resource("topology", ResourceType.TOPOLOGY)
async def get_topology_resource() -> Dict[str, Any]:
    """获取网络拓扑"""

2.2.2 资源实现方式

# 在server.py中的实际资源处理器
@mcp.resource("{uri}")
async def mcp_resource_handler(uri: str) -> Any:
    """
    MCP资源处理器，处理所有资源请求
    
    Args:
        uri: 资源URI
        
    Returns:
        资源内容
    """
    try:
        result = await resource_manager.get_resource(uri)
        return result
    except Exception as e:
        logger.error(f"获取资源失败: {uri}, 错误: {str(e)}")
        return {"error": f"获取资源失败: {str(e)}"}

2.3 提示模板（Prompts）

提示模板是用户控制的交互模板，帮助构造有效的LLM提示。

2.3.1 NetBrain MCP提示模板

系统实现了13个专业网络运维提示模板：

template_system.py中的模板（8个）

@template_manager.register_template(
    name="device_diagnosis",
    description="网络设备诊断模板"
)
def device_diagnosis_template(device_info: Dict[str, Any], interfaces_info: str = None, logs: str = None) -> List[Message]:
    """网络设备诊断提示模板，用于分析设备状态和接口信息"""
    system_prompt = f"""你是一位专业的网络工程师，负责诊断{device_info.get('vendor', '未知厂商')}的{device_info.get('device_type', '网络设备')}。
请根据提供的设备信息、接口状态和日志，进行专业分析并提供诊断报告。"""
    
    return [
        SystemMessage(system_prompt),
        UserMessage(f"请诊断设备：{device_info}")
    ]

@template_manager.register_template(
    name="config_review", 
    description="网络设备配置审查模板"
)
def config_review_template(device_info: Dict[str, Any], config_content: str, focus_area: str = None) -> List[Message]:
    """网络配置审查提示模板，用于审查设备配置并提供改进建议"""
    # 实现...

device_prompts.py中的模板（5个）

@template_manager.register_template(
    name="cisco_device_analysis",
    description="思科设备专业分析模板"
)
def cisco_device_analysis_template(device_info: Dict[str, Any], command_outputs: Dict[str, str]) -> List[Message]:
    """思科设备专业分析模板"""
    # 实现...

@template_manager.register_template(
    name="huawei_device_analysis", 
    description="华为设备专业分析模板"
)
def huawei_device_analysis_template(device_info: Dict[str, Any], command_outputs: Dict[str, str]) -> List[Message]:
    """华为设备专业分析模板"""
    # 实现...

2.3.2 模板实现方式

# 在server.py中的模板处理器
@mcp.prompt("{name}")
async def mcp_prompt_handler(name: str, arguments: dict = None) -> str:
    """
    MCP提示模板处理器
    
    Args:
        name: 模板名称
        arguments: 模板参数
        
    Returns:
        渲染后的提示内容
    """
    try:
        result = template_manager.render_template(name, arguments or {})
        if isinstance(result, list):
            # 如果返回消息列表，转换为字符串
            return "\n".join([msg.content for msg in result])
        return result or ""
    except Exception as e:
        logger.error(f"渲染模板失败: {name}, 错误: {str(e)}")
        return f"模板渲染失败: {str(e)}"

3. 实际使用示例

3.1 基本设备管理

3.1.1 添加设备凭据

# 在Claude或其他MCP客户端中输入：
请帮我添加一个思科设备的SSH凭据，用户名是admin，密码是cisco123，端口是22

AI将自动调用add_credential工具：

# 实际调用的工具
await add_credential(
    name="思科SSH凭据",
    username="admin", 
    password="cisco123",
    protocol="ssh",
    port=22
)

3.1.2 添加网络设备

请添加一台思科路由器，名称是Router-01，IP地址是192.168.1.1，型号是ISR4321，使用刚才添加的凭据

AI将调用add_device工具：

await add_device(
    name="Router-01",
    ip_address="192.168.1.1",
    device_type="router",
    vendor="cisco", 
    platform="cisco_iosxe",
    model="ISR4321"
)

3.2 设备连接和命令执行

3.2.1 连接设备

请连接到Router-01设备

AI将调用连接工具：

await connect_device(
    device_id="router-01-uuid",
    credential_id="credential-uuid"
)

3.2.2 执行命令

在Router-01上执行 show version 命令

AI将调用命令执行工具：

await send_command(
    device_id="router-01-uuid",
    credential_id="credential-uuid", 
    command="show version",
    timeout=30
)

3.3 拓扑发现

请发现Router-01的网络拓扑

AI将调用拓扑发现工具：

await discover_topology(
    device_ids=["router-01-uuid"]
)

3.4 使用提示模板

使用设备诊断模板分析Router-01的状态

AI将：

获取设备信息资源
获取设备接口信息
使用device_diagnosis模板进行分析

4. 高级功能

4.1 批量操作

4.1.1 批量命令执行

在Router-01上依次执行以下命令：
1. show running-config
2. show ip route
3. show interface status

AI将调用send_commands工具：

await send_commands(
    device_id="router-01-uuid",
    credential_id="credential-uuid",
    commands=[
        "show running-config",
        "show ip route", 
        "show interface status"
    ]
)

4.1.2 网络扫描

请扫描192.168.1.0/24网段，发现网络设备

AI将调用网络扫描工具：

await scan_network_range(
    network="192.168.1.0/24",
    timeout=5,
    max_concurrent=50,
    ports=[22, 23, 80, 443, 161]
)

4.2 资源访问

4.2.1 获取设备配置

请获取Router-01的配置信息

AI将访问配置资源：

资源URI: device/router-01-uuid/config

4.2.2 获取拓扑信息

显示当前网络拓扑的统计信息

AI将访问拓扑资源：

资源URI: topology/statistics

4.3 智能诊断

4.3.1 使用诊断模板

使用华为设备分析模板检查Switch-01的状态

AI将：

识别设备是华为设备
获取设备相关信息和命令输出
使用huawei_device_analysis模板
提供专业的分析报告

4.3.2 配置审查

使用配置审查模板检查Router-01的安全配置

AI将使用config_review模板，重点关注安全配置。

5. 系统集成

5.1 连接到Claude Desktop

5.1.1 配置文件

编辑Claude Desktop配置文件：

Windows: %APPDATA%\Claude\claude_desktop_config.json macOS: ~/Library/Application Support/Claude/claude_desktop_config.json Linux: ~/.config/Claude/claude_desktop_config.json

{
"mcpServers": {
 "netbrain-mcp": {
    "command": "python",
   "args": ["/path/to/NetBrainMCP/server.py"],
   "env": {
     "LOG_LEVEL": "INFO"
   }
  }
}
}

5.1.2 验证连接

重启Claude Desktop后，在对话中输入：

请列出所有可用的网络设备

如果看到工具图标并得到设备列表响应，说明连接成功。

5.2 连接到Cursor IDE

5.2.1 启动SSE模式

cd /path/to/NetBrainMCP
python server.py --mode sse --port 8000

5.2.2 配置Cursor

打开Cursor → 设置 → 功能 → MCP服务器
添加新服务器：
- 名称: netbrain-mcp
- 类型: sse
- URL: http://localhost:8000/sse

5.3 Web界面使用

5.3.1 启动Web服务器

python server.py --web --port 8080

5.3.2 访问Web界面

在浏览器中访问：http://localhost:8080

Web界面提供：

设备管理页面
多标签页终端
交互式拓扑图

6. 故障排除

6.1 常见问题

6.1.1 MCP服务器无法启动

问题症状：

ModuleNotFoundError: No module named 'mcp'

解决方案：

pip install -r requirements.txt

6.1.2 设备连接失败

问题症状：

{"success": false, "error": "连接超时"}

解决方案：

检查网络连通性：ping 192.168.1.1
验证SSH/Telnet服务：telnet 192.168.1.1 22
确认凭据正确性
检查防火墙设置

6.1.3 Claude Desktop连接问题

问题症状：Claude Desktop不显示工具图标

解决方案：

检查配置文件路径和格式
验证Python路径和脚本路径
查看Claude Desktop日志
重启Claude Desktop

6.2 调试技巧

6.2.1 启用详细日志

export LOG_LEVEL=DEBUG
python server.py

6.2.2 测试工具连接

# 测试Scrapli连接
python -c "
import asyncio
from server import test_scrapli_connection
result = asyncio.run(test_scrapli_connection(
    host='192.168.1.1',
    username='admin',
    password='cisco123',
    platform='cisco_iosxe'
))
print(result)
"

7. 扩展开发

7.1 添加自定义工具

@mcp.tool()
async def custom_network_tool(param1: str, param2: int) -> Dict[str, Any]:
    """自定义网络工具"""
    try:
        # 实现自定义逻辑
        result = perform_custom_operation(param1, param2)
        return {"success": True, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

7.2 添加自定义资源

@resource_manager.register_resource("custom/{id}", ResourceType.CUSTOM)
async def get_custom_resource(id: str) -> Dict[str, Any]:
    """自定义资源处理器"""
    # 实现自定义资源获取逻辑
    return {"data": f"自定义数据 for {id}"}

7.3 添加自定义模板

@template_manager.register_template(
    name="custom_analysis",
    description="自定义分析模板"
)
def custom_analysis_template(data: Dict[str, Any]) -> str:
    """自定义分析模板"""
    return f"基于数据 {data} 进行自定义分析..."

8. 最佳实践

8.1 设备管理最佳实践

统一命名规范：使用一致的设备命名格式
合理分组标签：按功能、位置、环境分组
定期更新信息：保持设备信息的准确性
备份配置数据：定期备份重要配置

8.2 使用技巧

充分利用模板：使用专业模板获得更好的分析结果
批量操作：对多个设备执行相同操作时使用批量工具
资源组合：结合多个资源获得全面的设备视图
缓存机制：利用系统缓存提高查询效率

8.3 安全建议

凭据管理：使用强密码和定期轮换
网络隔离：在安全的网络环境中部署
访问控制：限制MCP服务器的访问权限
日志监控：定期检查操作日志

9. 性能优化

9.1 连接优化

# 调整连接池设置
MAX_CONNECTIONS = 10
CONNECTION_TIMEOUT = 30
COMMAND_TIMEOUT = 60

9.2 缓存优化

# 调整缓存策略
RESOURCE_CACHE_TTL = 300  # 5分钟
CONFIG_CACHE_TTL = 600    # 10分钟
TOPOLOGY_CACHE_TTL = 900  # 15分钟

9.3 并发控制

# 网络扫描并发控制
DEFAULT_MAX_CONCURRENT = 50
SCAN_TIMEOUT = 5

10. 总结

NetBrain MCP项目提供了一个完整的网络运维AI集成平台，通过MCP协议实现：

32个专业工具：涵盖网络设备管理的各个方面
13个数据资源：提供丰富的网络设备和拓扑信息
13个提示模板：专业的网络运维AI指导
多厂商支持：思科、华为等主流设备厂商
灵活部署：支持Claude Desktop、Cursor IDE和Web界面

通过本教程，您应该能够熟练使用NetBrain MCP系统进行AI驱动的网络运维工作。系统的模块化设计和丰富的扩展机制，也为进一步的定制和开发提供了良好的基础。

18 KiB Raw Permalink Blame History Unescape Escape

NetBrain MCP 项目教程

1. MCP协议概述

1.1 什么是MCP？

1.2 NetBrain MCP的实现特点

2. MCP核心概念

2.1 工具（Tools）

2.1.1 NetBrain MCP工具分类

2.1.2 工具实现方式

2.2 资源（Resources）

2.2.1 NetBrain MCP资源列表

2.2.2 资源实现方式

2.3 提示模板（Prompts）

2.3.1 NetBrain MCP提示模板

2.3.2 模板实现方式

3. 实际使用示例

3.1 基本设备管理

3.1.1 添加设备凭据

3.1.2 添加网络设备

3.2 设备连接和命令执行

3.2.1 连接设备

3.2.2 执行命令

3.3 拓扑发现

3.4 使用提示模板

4. 高级功能

4.1 批量操作

4.1.1 批量命令执行

4.1.2 网络扫描

4.2 资源访问

4.2.1 获取设备配置

4.2.2 获取拓扑信息

4.3 智能诊断

4.3.1 使用诊断模板

4.3.2 配置审查

5. 系统集成

5.1 连接到Claude Desktop

5.1.1 配置文件

5.1.2 验证连接

5.2 连接到Cursor IDE

5.2.1 启动SSE模式

5.2.2 配置Cursor

5.3 Web界面使用

5.3.1 启动Web服务器

5.3.2 访问Web界面

6. 故障排除

6.1 常见问题

6.1.1 MCP服务器无法启动

6.1.2 设备连接失败

6.1.3 Claude Desktop连接问题

6.2 调试技巧

6.2.1 启用详细日志

6.2.2 测试工具连接

7. 扩展开发

7.1 添加自定义工具

7.2 添加自定义资源

7.3 添加自定义模板

8. 最佳实践

8.1 设备管理最佳实践

8.2 使用技巧

8.3 安全建议

9. 性能优化

9.1 连接优化

9.2 缓存优化

9.3 并发控制

10. 总结

18 KiB

Raw Permalink Blame History