pyATS MCP Server

pyATS MCP 服务

Cisco pyATS 和 Genie 已经知道如何与网络通信——解析 show 命令、推送配置、学习功能状态、运行声明式测试。它们缺少的是让 AI Agent 直接驱动这些功能的方式。此服务填补了这一空白：它将 pyATS/Genie 封装为一组结构化且设有防护措施的 MCP 工具，Agent（例如 Claude）可通过 Model Context Protocol 当前的 Streamable HTTP 传输方式，针对真实测试环境调用这些工具。

将 Agent 指向此服务，它便可以查找设备、运行并解析 show 命令、应用配置并设置回滚点、学习并比较功能在变更前后的状态、将命令分发到整个设备集群（通过一个线程池或每台设备一个进程）、运行声明式 Blitz 或 Robot Framework 测试，或直接调用设备的 REST/RESTCONF API。所有有风险的操作在到达设备前都会受到防护，每次调用都会记录到内存中的审计日志，供 Agent 在会话期间查看。

---

一览

• 传输方式 — Streamable HTTP（mcp>=2.3.0），可选择有状态或无状态，通过一个环境变量进行设置。STDIO 已移除。
• 26 个工具，涵盖发现、show 命令、配置、Genie 学习/差异比较、Genie Clean、声明式测试（Blitz、Robot Framework、AEtest）、通用 REST/RESTCONF 和 Cisco XPresso。
• 两种分发命令的方式 — 在多台设备间分发命令：日常使用可采用共享线程池；需要大规模实现真正隔离时，可为每台设备启动一个 OS 进程（pyats.async_.pcall）。
• 防护措施，而非依赖自律 — 危险命令在到达设备前会被拦截；Genie Clean 绝不会运行会重启或重装设备的阶段；破坏性操作必须提供完全匹配的确认短语。
• 没有硬编码内容 — 所有凭据和设备详情都存放在 .env 中，并在运行时通过 %ENV{} 替换注入 testbed.yaml。

---

前置条件

• Python 3.10+
• 指向真实或虚拟网络设备的 pyATS testbed.yaml——可以是物理实验室、Cisco Modeling Labs / VIRL / GNS3，也可以是任何 Unicon 能通过 SSH/Telnet 访问的环境。pyATS MCP 不会模拟网络，而是直接驱动网络。
• 可与其通信的 MCP 客户端——请参阅下方的连接你的 Agent。

---

快速开始

# 1. Clone and install
git clone https://github.com/automateyournetwork/pyATS_MCP
cd pyATS_MCP
pip install -r requirements.txt

# 2. Configure your environment
cp .env.example .env
# Edit .env — see Configuration below

# 3. Run — starts a Streamable HTTP server on 0.0.0.0:8080 by default
python3 pyats_mcp_server.py

随后即可通过 http://<host>:<port>/mcp 访问 MCP 端点。

---

配置

所有设备详情和凭据都存放在 .env 文件中——仓库中没有任何硬编码内容。

1. 复制模板

cp .env.example .env

2. 设置服务变量

PYATS_TESTBED_PATH=/absolute/path/to/your/testbed.yaml
PYATS_MCP_ARTIFACTS_DIR=          # default: ~/.pyats-mcp/artifacts
PYATS_MCP_KEEP_ARTIFACTS=1        # 1 = keep, 0 = delete after each run
PYATS_MCP_TESTBED_CACHE_TTL=30    # seconds before testbed reloads from disk
PYATS_MCP_CONN_CACHE_TTL=0        # seconds to keep connections alive (0 = off)
PYATS_MCP_OP_LOG_MAX=500          # max entries in the in-memory operation log

# Transport (Streamable HTTP only — STDIO is not supported)
PYATS_MCP_TRANSPORT_MODE=stateful # stateful (default) | stateless
PYATS_MCP_HTTP_HOST=0.0.0.0
PYATS_MCP_HTTP_PORT=8080

# Optional — only needed for pyats_xpresso_request
XPRESSO_URL=
XPRESSO_API_TOKEN=
XPRESSO_GROUP=

PYATS_MCP_TRANSPORT_MODE=stateless 会在 Streamable HTTP 传输中设置 stateless_http=True，因此对于仍在协商旧版、基于握手的协议的客户端，各请求之间不会保留服务端会话状态。无论此设置如何，使用当前 MCP 协议（2026-07-28，SEP-2575）的客户端默认都无需握手——这是 mcp>=2.3.0 SDK 本身的行为，并非此处配置的结果。

3. 为每台设备添加一个配置块

testbed.yaml 中的每台设备都会使用 %ENV{VAR} 替换，因此凭据和连接详情会在运行时从 .env 读取。

请使用 {DEVICENAME}_{FIELD} 命名约定：

# Supported os values: iosxe | iosxr | nxos | ios | eos | junos | panos | linux | windows
# Set os=generic and platform="" to let Unicon autodetect on first connect.

CORE1_IP=10.1.1.1
CORE1_PORT=22
CORE1_OS=iosxe
CORE1_PLATFORM=cat9k
CORE1_USERNAME=admin
CORE1_PASSWORD=s3cr3t
CORE1_ENABLE_PASSWORD=s3cr3t

FW1_IP=10.1.1.2
FW1_PORT=22
FW1_OS=panos
FW1_PLATFORM=
FW1_USERNAME=admin
FW1_PASSWORD=s3cr3t
# (no enable password for Palo Alto)

LINUX1_IP=10.1.1.3
LINUX1_PORT=22
LINUX1_OS=linux
LINUX1_PLATFORM=ubuntu
LINUX1_USERNAME=admin
LINUX1_PASSWORD=s3cr3t
# (no enable password for Linux)

如果一组设备共用凭据，请定义组级变量并在各设备间引用：

SITE_A_USERNAME=netops
SITE_A_PASSWORD=s3cr3t
SITE_A_ENABLE_PASSWORD=s3cr3t

4. 在 testbed.yaml 中引用这些变量

devices:
  CORE1:
    alias: "Core Switch 1"
    type: "switch"
    os: "%ENV{CORE1_OS}"
    platform: "%ENV{CORE1_PLATFORM}"
    credentials:
      default:
        username: "%ENV{CORE1_USERNAME}"
        password: "%ENV{CORE1_PASSWORD}"
      enable:
        password: "%ENV{CORE1_ENABLE_PASSWORD}"
    connections:
      cli:
        protocol: ssh
        ip: "%ENV{CORE1_IP}"
        port: "%ENV{CORE1_PORT}"
        arguments:
          connection_timeout: 360

对于 OS 未知的设备，请设置 os: "%ENV{DEVICE_OS}"，并在 .env 中设置 DEVICE_OS=generic 还可以选择在 arguments: 下添加 learn_os: true——Unicon 会在首次连接后检测并缓存 OS。

---

Docker

构建

docker build -t pyats-mcp-server .

运行（直接传入 .env）

docker run -p 8080:8080 --rm \
  --env-file /absolute/path/to/.env \
  -v /absolute/path/to/testbed.yaml:/app/testbed.yaml \
  pyats-mcp-server

无论哪种方式，服务都是一个长时间运行的进程，只需启动一次，然后让客户端连接到它——它不是由 Agent 为每个会话单独启动的进程。下文将准确说明各个客户端如何连接到它。

---

连接你的 Agent

服务只暴露一个接口：位于 http://<host>:<port>/mcp 的 MCP 端点（Streamable HTTP）。下方的每个客户端只需要这个 URL——无需 command/args，客户端也无需管理本地进程。

Claude Code

claude mcp add --transport http pyats http://localhost:8080/mcp

# Behind auth (e.g. a reverse proxy in front of the server)
claude mcp add --transport http pyats http://localhost:8080/mcp \
  --header "Authorization: Bearer your-token"

或者直接将其放入 .mcp.json（项目范围，提交到代码仓库）或 ~/.claude.json（用户范围）：

{
  "mcpServers": {
    "pyats": { "type": "http", "url": "http://localhost:8080/mcp" }
  }
}

VS Code（GitHub Copilot Chat）

在工作区中添加 .vscode/mcp.json（或者从命令面板运行 MCP: Add Server）：

{
  "servers": {
    "pyats": { "type": "http", "url": "http://localhost:8080/mcp" }
  }
}

OpenAI Codex CLI

codex mcp add pyats --url http://localhost:8080/mcp

或者写入 ~/.codex/config.toml：

[mcp_servers.pyats]
url = "http://localhost:8080/mcp"

Claude Desktop

Claude Desktop 的 claude_desktop_config.json 仅支持 stdio——在其中填写 url 字段无效（这是一个已知问题，并非受支持的方式）。远程/HTTP 服务应通过“设置 → 连接器”下的自定义连接器添加；Desktop 会从 Anthropic 云端连接到该服务，而不是从你的本地计算机连接——因此需要一个真实且可公开访问的 HTTPS URL，而不能是 localhost。

如果仍要让 Desktop 连接到运行在你自己计算机上的服务，可以通过 mcp-remote 将其桥接为本地 stdio 代理：

{
  "mcpServers": {
    "pyats": {
      "command": "npx",
      "args": ["-y", "mcp-remote", "http://localhost:8080/mcp", "--transport", "http-only"]
    }
  }
}

原始 Python（LangGraph、自定义 Agent 及其他场景）

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    async with streamablehttp_client("http://localhost:8080/mcp") as (read, write, _session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            result = await session.call_tool(
                "pyats_run_show_command",
                arguments={"device_name": "CORE1", "command": "show version"},
            )

---

异步 MCP 任务

MCP 支持立即返回任务句柄，并在稍后获取结果。 本服务通过 Python SDK 的扩展钩子（mcp>=2.3.0）实现了 2026-07-28 io.modelcontextprotocol/tasks 扩展。它未实现较早的实验性 2025-11-25 tasks/result / tasks/list 协议。 请参阅 Tasks 规范 和 Python SDK 扩展 API。

当客户端在请求中声明支持此扩展时，所有 22 个设备 I/O、远程请求和测试执行工具都会使用任务。设备清单/搜索、快照差异和操作历史仍会立即返回。未声明支持 Tasks 的现有客户端仍会获得普通的工具结果。客户端必须支持此扩展；仅支持较早的实验性 Tasks API 并不够。

初始的 tools/call 请求包含以下元数据（以及 name 和 arguments）：

{
  "_meta": {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientCapabilities": {
      "extensions": {"io.modelcontextprotocol/tasks": {}}
    }
  }
}

响应包含 resultType: "task"、taskId、status: "working"、时间戳和 pollIntervalMs。使用相同的元数据，调用 tasks/get 并传入 {"taskId": "..."} 进行轮询。其响应包含 resultType: "complete"；请检查 status 以确定工作是否已完成。已完成的任务会在 result 中包含原始 MCP 工具响应。协议失败时，status 为 "failed"，并包含 error；工具错误仍会使用 status: "completed"，并保留工具的错误负载。HTTP 请求使用 Mcp-Method 和 Mcp-Name 标头；对于任务方法，Mcp-Name 是任务 ID。

运行轮询客户端示例：

python examples/task_client.py http://localhost:8080/mcp pyats_run_show_command \
  '{"device_name":"router-1","command":"show version"}'

tasks/cancel 会确认已收到取消意图。排队中的工作会在执行前取消。工具一旦开始运行，就会完成并保留其结果：取消等待无法安全地停止 SSH 命令、还原配置更改或终止测试子进程。tasks/update 会确认并忽略输入响应键；这些工具不会请求交互式输入。通知尚未实现；客户端使用轮询，并遵守 pollIntervalMs（1 秒）。

变量	默认值	含义
PYATS_MCP_TASK_DB	<artifacts directory>/tasks.sqlite3	持久化任务／结果存储
PYATS_MCP_TASK_WORKERS	4	长时间运行的工具调用的最大并发数
PYATS_MCP_TASK_MAX	1000	保留任务的最大数量，包括排队中／运行中的任务
PYATS_MCP_TASK_RETENTION	86400	终止状态结果的保留秒数

工作线程限制也适用于普通调用。多设备工具会保留其内部扇出行为。针对同一设备的调用会在整个工具操作期间串行执行，包括配置快照和差异对比；针对不同设备的调用可以并发执行。动态 Python／Robot 测试会锁定测试床中的所有设备。连接到所提供测试床之外设备的脚本仍需自行负责并发控制。连接 TTL 驱逐会在下次使用该设备时发生。

无论处于有状态还是无状态模式，任务都能在 HTTP 断开后继续运行。SQLite 会在返回任务句柄前提交该句柄，并在重启后保留终止状态的结果。运行中的任务具有 ttlMs: null；任务完成后会设置 TTL，从创建时间起算，其中包括保留期。过期结果会在后续提交任务时删除。达到保留任务数量上限时，将拒绝接收新任务。 重启时，未完成的任务会变为失败，并显示结果未知的消息；绝不会自动重放这些任务。重试更改前，请先检查设备状态。正常关闭时，会取消排队中的任务，并等待活动操作完成。

每个任务数据库仅运行一个服务进程；文件锁会强制执行此限制。对于 Docker，请挂载持久化目录，并将 PYATS_MCP_TASK_DB 指向该目录。水平副本需要使用独立存储，并将每个任务 ID 路由回其所属副本。此实现不是分布式任务队列。

数据库包含工具结果，可能包括配置，并且创建时会设置仅所有者可访问的权限。经过身份验证的任务会绑定到已验证客户端／发行方／主体身份。在本项目默认的未认证服务中，不可猜测的任务 ID 充当持有者凭证；请妥善保护它和数据库。

可以问它什么

连接后，可以像跟熟悉网络的人交谈一样向它提问：

• “测试床里有哪些设备？” → pyats_list_devices
• “显示 CORE1 上的 BGP 摘要” → pyats_run_show_command，并解析为结构化 JSON
• “先获取 CORE1 的 OSPF 状态快照，再应用此配置，并告诉我有哪些变化” → pyats_learn_feature（之前） → pyats_configure_with_diff → pyats_learn_feature（之后） → pyats_diff_learned_snapshots
• “在每台交换机上运行 show ip interface brief” → pyats_run_show_command_multi（或者使用 pyats_pcall_show_command，以便在大规模场景中按进程隔离每台设备）
• “如果配置更改导致问题，就回滚” → pyats_rollback_config
• “针对 R1 和 R2 运行此 Blitz 测试”／“运行此 Robot Framework 套件” → pyats_run_blitz／pyats_run_robot

Agent 会自行串联这些工具——你描述想要的结果，它会选择合适的工具。

---

可用工具

共 26 个工具，按用途分组。

发现

工具	描述
pyats_list_devices	列出测试平台中的所有设备
pyats_search_devices	按名称或别名模糊搜索设备

Show 命令

工具	描述
pyats_run_show_command	运行经过验证的 show 命令；返回解析后的 JSON 或原始输出
pyats_run_show_command_multi	通过线程池并发在多台设备上运行 show 命令
pyats_pcall_show_command	功能相同，但为每台设备启动一个操作系统进程（pyats.async_.pcall），而非使用共享线程池
pyats_show_running_config	获取完整的运行配置（原始文本）
pyats_show_logging	通过 show logging 获取设备系统日志
pyats_ping_from_network_device	从网络设备执行 ping
pyats_run_linux_command	在 Linux 主机上运行命令

配置

工具	描述
pyats_configure_device	应用配置命令，并提供安全保护措施
pyats_configure_devices_multi	通过线程池并发在多台设备上应用配置
pyats_pcall_configure_devices	功能相同，但为每台设备启动一个操作系统进程
pyats_configure_with_diff	应用配置并返回前后差异
pyats_rollback_config	回滚到上次保存的配置快照

状态与诊断

工具	描述
pyats_device_health	获取 CPU、内存、接口和路由状态快照
pyats_get_neighbors	获取 CDP/LLDP 邻居
pyats_find_interface_by_ip	查找拥有指定 IP 地址的接口
pyats_learn_feature	对整个功能执行 Genie device.learn()（接口、ospf、bgp 等），可选择保存为命名快照
pyats_diff_learned_snapshots	对比由 pyats_learn_feature 保存的两个快照

测试与自动化

工具	描述
pyats_clean_device	Genie Clean（Kleenex），仅允许非破坏性的 connect+execute_command 阶段；默认 dry_run=True
pyats_run_blitz	运行声明式的 pyATS Blitz YAML 测试
pyats_run_robot	使用 pyats.robot/genie.libs.robot 关键字库运行 Robot Framework 套件
pyats_run_dynamic_test	执行沙箱隔离的 pyATS AEtest 脚本

API

工具	描述
pyats_rest_request	通过 pyATS 的 rest.connector 发起通用 REST/RESTCONF/NX-API 调用（与 CLI/SSH 不同的连接类型）
pyats_xpresso_request	通过身份验证调用 Cisco XPresso REST API v2（测试请求、作业、测试平台、镜像等）

会话

工具	描述
pyats_get_operation_log	获取内存中的操作日志

---

安全性

• Show 命令均经过验证——管道符、重定向符和危险关键字都会被拦截。
• 配置变更会检查 reload、erase、write erase、delete、format——pyats_clean_device、pyats_run_blitz 和 pyats_run_robot 内部也会执行相同检查。
• 动态测试脚本在受限沙箱中运行（禁止导入 os、sys、subprocess 等模块）。
• pyats_clean_device 绝不会运行会重启、擦除或重建映像设备的真实 Genie Clean 阶段——只会生成 connect+execute_command；默认使用 dry_run=True，实际运行还需要提供完全匹配的确认短语。
• 所有进程级全局缓存（连接缓存、测试平台缓存、配置/学习快照、操作日志）都受锁保护，因此并发 HTTP 客户端无法破坏共享状态。
• 所有凭据均来自 .env——绝不会存储在测试平台文件或源代码中。

---

项目结构

.
├── pyats_mcp_server.py      # MCP server
├── test_pyats_mcp_server.py # Unit tests (119 tests)
├── benchmark/               # Pre/post, stateful/stateless transport benchmark
├── Dockerfile               # Container definition
├── requirements.txt         # Pinned runtime dependencies
├── requirements-dev.txt     # Dev/test dependencies
├── pyproject.toml           # Tool config (black, isort, pytest, mypy)
├── .env.example             # Configuration template — copy to .env
├── .gitignore
├── LICENSE
└── CONTRIBUTING.md

---

开发

# Install dev dependencies with uv
uv venv .venv && uv pip install -r requirements-dev.txt

# Run tests
.venv/bin/python -m pytest

# Lint and format
.venv/bin/black .
.venv/bin/isort .
.venv/bin/flake8 . --max-line-length=100

完整的环境设置和 PR 工作流请参阅 CONTRIBUTING.md。

---

基准测试

benchmark/ 将 STDIO（旧版）与有状态和无状态模式下的 Streamable HTTP 进行比较，测试环境为真实测试床。场景列表请参见 benchmark/scenarios.py，比较报告的生成方式请参见 benchmark/aggregate.py；benchmark/results/summary.md 列出了最近一次运行的数据。

---

许可证

MIT