pyATS
Interaja com dispositivos de rede usando as bibliotecas pyATS e Genie da Cisco para automação orientada a modelos.
Documentação
Servidor MCP pyATS
O Cisco pyATS e o Genie já sabem como conversar com uma rede — analisando comandos show, aplicando configuração, aprendendo o estado de funcionalidades, executando testes declarativos. O que faltava era uma forma de um agente de IA acionar qualquer um deles diretamente. Este servidor preenche essa lacuna: ele encapsula pyATS/Genie como um conjunto de ferramentas MCP estruturadas e protegidas que um agente como o Claude pode chamar contra um testbed real, por meio do transporte Streamable HTTP atual do Model Context Protocol.
Aponte um agente para ele e ele poderá consultar um dispositivo, executar e analisar um comando show, aplicar configuração com ponto de rollback, aprender e comparar o estado de uma funcionalidade antes e depois de uma alteração, distribuir um comando por uma frota — um pool de threads ou um processo por dispositivo — executar um teste declarativo Blitz ou Robot Framework, ou chamar a API REST/RESTCONF de um dispositivo diretamente. Todo caminho arriscado é protegido antes de chegar a um dispositivo, e cada chamada é registrada em um log de auditoria em memória que o agente pode revisar durante a sessão.
Resumo
- Transporte — Streamable HTTP (
mcp>=2.0.0), com ou sem estado, escolhido por uma variável de ambiente. STDIO foi removido. - 26 ferramentas em descoberta, comandos show, configuração, Genie learn/diff, Genie Clean, testes declarativos (Blitz, Robot Framework, AEtest), REST/RESTCONF genérico e Cisco XPresso.
- Duas formas de distribuir um comando por vários dispositivos — um pool de threads compartilhado para uso diário, ou um processo de SO por dispositivo (
pyats.async_.pcall) quando você quer isolamento real em escala. - Proteções, não sistemas de confiança — comandos perigosos são bloqueados antes de chegarem a um dispositivo, o Genie Clean nunca executa uma etapa que reinicie ou reimage um dispositivo, e ações destrutivas exigem uma frase de confirmação exata.
- Nada codificado — cada credencial e detalhe de dispositivo vive em
.env, puxado paratestbed.yamlem tempo de execução via substituição%ENV{}.
Pré-requisitos
- Python 3.10+
- Um
testbed.yamlpyATS apontado para dispositivos de rede reais ou virtuais — um laboratório físico, Cisco Modeling Labs / VIRL / GNS3, ou qualquer outra coisa que o Unicon consiga alcançar via SSH/Telnet. O pyATS MCP não simula uma rede; ele aciona uma. - Um cliente compatível com MCP para conversar com ele — veja Conecte Seu Agente abaixo.
Início Rápido
# 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
O endpoint MCP fica então acessível em http://<host>:<port>/mcp.
Configuração
Todos os detalhes de dispositivos e credenciais vivem em um arquivo .env — nada é codificado no repositório.
1. Copie o modelo
cp .env.example .env
2. Defina as variáveis do servidor
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 define stateless_http=True no transporte Streamable HTTP, para que nenhum estado de sessão no servidor seja mantido entre requisições de clientes que ainda negociam o protocolo mais antigo baseado em handshake. Clientes que falam o protocolo MCP atual (2026-07-28, SEP-2575) são livres de handshake por padrão, independentemente dessa configuração — isso vem do próprio SDK mcp>=2.0.0, não de nada configurado aqui.
3. Adicione um bloco para cada dispositivo
Cada dispositivo no seu testbed.yaml usa substituição %ENV{VAR}, então credenciais e detalhes de conexão são lidos de .env em tempo de execução.
Use a convenção de nomenclatura {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)
Se um grupo de dispositivos compartilha credenciais, defina variáveis de nível de grupo e referencie-as entre dispositivos:
SITE_A_USERNAME=netops
SITE_A_PASSWORD=s3cr3t
SITE_A_ENABLE_PASSWORD=s3cr3t
4. Referencie as variáveis no 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
Para dispositivos com SO desconhecido, defina
os: "%ENV{DEVICE_OS}"comDEVICE_OS=genericem.enve opcionalmente adicionelearn_os: truesobarguments:— o Unicon detectará e armazenará em cache o SO após a primeira conexão.
Docker
Build
docker build -t pyats-mcp-server .
Executar (passar .env diretamente)
docker run -p 8080:8080 --rm \
--env-file /absolute/path/to/.env \
-v /absolute/path/to/testbed.yaml:/app/testbed.yaml \
pyats-mcp-server
De qualquer forma, o servidor é um processo de longa duração que você inicia uma vez e aponta clientes para ele — não é algo que um agente inicia por sessão. Veja abaixo exatamente como cada cliente se conecta a ele.
Conecte Seu Agente
O servidor expõe uma coisa: um endpoint MCP em http://<host>:<port>/mcp (Streamable HTTP). Cada cliente abaixo só precisa dessa URL — sem command/args, sem processo local para o cliente gerenciar.
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"
Ou coloque diretamente em .mcp.json (escopo do projeto, commitado no repositório) ou ~/.claude.json (escopo do usuário):
{
"mcpServers": {
"pyats": { "type": "http", "url": "http://localhost:8080/mcp" }
}
}
VS Code (GitHub Copilot Chat)
Adicione um .vscode/mcp.json no workspace (ou execute MCP: Add Server na Paleta de Comandos):
{
"servers": {
"pyats": { "type": "http", "url": "http://localhost:8080/mcp" }
}
}
OpenAI Codex CLI
codex mcp add pyats --url http://localhost:8080/mcp
Ou em ~/.codex/config.toml:
[mcp_servers.pyats]
url = "http://localhost:8080/mcp"
Claude Desktop
O claude_desktop_config.json do Claude Desktop é somente STDIO — colocar um campo url nele não funciona (é um problema conhecido, não um caminho suportado). Servidores remotos/HTTP são adicionados em vez disso como um Custom Connector em Configurações → Conectores, e o Desktop se conecta a eles pela nuvem da Anthropic, não pela sua máquina local — então ele precisa de uma URL HTTPS real e publicamente acessível, não localhost.
Para apontar o Desktop para um servidor rodando na sua própria máquina mesmo assim, faça uma ponte via mcp-remote como um proxy STDIO local:
{
"mcpServers": {
"pyats": {
"command": "npx",
"args": ["-y", "mcp-remote", "http://localhost:8080/mcp", "--transport", "http-only"]
}
}
}
Python puro (LangGraph, agentes personalizados, qualquer outra coisa)
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"},
)
O Que Perguntar a Ele
Depois de conectado, fale com ele como falaria com alguém que já conhece a rede:
- "Quais dispositivos estão no testbed?" →
pyats_list_devices - "Mostre-me o resumo BGP no CORE1" →
pyats_run_show_command, analisado em JSON estruturado - "Capture o estado OSPF do CORE1, depois aplique esta configuração e mostre-me o que mudou" →
pyats_learn_feature(antes) →pyats_configure_with_diff→pyats_learn_feature(depois) →pyats_diff_learned_snapshots - "Execute
show ip interface briefem todos os switches" →pyats_run_show_command_multi(oupyats_pcall_show_commandpara isolamento de processo por dispositivo em escala real) - "Se essa mudança de configuração quebrar algo, reverta" →
pyats_rollback_config - "Execute este teste Blitz contra R1 e R2" / "Execute esta suíte Robot Framework" →
pyats_run_blitz/pyats_run_robot
O agente encadeia isso sozinho — você descreve o resultado, ele escolhe as ferramentas.
Ferramentas Disponíveis
26 ferramentas, agrupadas pelo que fazem.
Descoberta
| Ferramenta | Descrição |
|---|---|
pyats_list_devices | Lista todos os dispositivos no testbed |
pyats_search_devices | Busca difusa de dispositivos por nome ou alias |
Comandos show
| Ferramenta | Descrição |
|---|---|
pyats_run_show_command | Executa um comando show validado; retorna JSON analisado ou saída bruta |
pyats_run_show_command_multi | Executa um comando show em vários dispositivos simultaneamente (pool de threads) |
pyats_pcall_show_command | O mesmo, mas um processo de SO por dispositivo (pyats.async_.pcall) em vez de um pool de threads compartilhado |
pyats_show_running_config | Recupera a configuração running completa (texto bruto) |
pyats_show_logging | Recupera logs do sistema do dispositivo via show logging |
pyats_ping_from_network_device | Executa um ping a partir de um dispositivo de rede |
pyats_run_linux_command | Executa um comando em um host Linux |
Configuração
| Ferramenta | Descrição |
|---|---|
pyats_configure_device | Aplica comandos de configuração com proteções de segurança |
pyats_configure_devices_multi | Aplica configuração em vários dispositivos simultaneamente (pool de threads) |
pyats_pcall_configure_devices | O mesmo, mas um processo de SO por dispositivo |
pyats_configure_with_diff | Aplica configuração e retorna um diff antes/depois |
pyats_rollback_config | Reverte para o último snapshot de configuração salvo |
Estado e diagnóstico
| Ferramenta | Descrição |
|---|---|
pyats_device_health | Captura CPU, memória, interfaces e estado de roteamento |
pyats_get_neighbors | Recupera vizinhos CDP/LLDP |
pyats_find_interface_by_ip | Descobre qual interface possui um determinado endereço IP |
pyats_learn_feature | Genie device.learn() para uma funcionalidade inteira (interface, ospf, bgp, …), opcionalmente salvo como um snapshot nomeado |
pyats_diff_learned_snapshots | Compara dois snapshots salvos por pyats_learn_feature |
Testes e automação
| Ferramenta | Descrição |
|---|---|
pyats_clean_device | Genie Clean (Kleenex), restrito a etapas não destrutivas connect+execute_command; dry_run=True por padrão |
pyats_run_blitz | Executa um teste declarativo pyATS Blitz YAML |
pyats_run_robot | Executa uma suíte Robot Framework usando as bibliotecas de palavras-chave pyats.robot/genie.libs.robot |
pyats_run_dynamic_test | Executa um script pyATS AEtest em sandbox |
APIs
| Ferramenta | Descrição |
|---|---|
pyats_rest_request | Chamada REST/RESTCONF/NX-API genérica via rest.connector do pyATS (um tipo de conexão separado de CLI/SSH) |
pyats_xpresso_request | Chamada autenticada à API REST v2 do Cisco XPresso (solicitações de teste, jobs, testbeds, imagens, …) |
Sessão
| Ferramenta | Descrição |
|---|---|
pyats_get_operation_log | Recupera o log de operações em memória |
Segurança
- Comandos show são validados — pipes, redirecionamentos e palavras-chave perigosas são bloqueados.
- Mudanças de configuração são verificadas para
reload,erase,write erase,delete,format— a mesma verificação roda dentro depyats_clean_device,pyats_run_blitzepyats_run_robot. - Scripts de teste dinâmicos rodam em um sandbox restrito (imports proibidos:
os,sys,subprocess, etc.). pyats_clean_devicenunca executa uma etapa real do Genie Clean que reinicie, apague ou reimage um dispositivo — apenasconnect+execute_commandsão gerados — e o padrão édry_run=True; executar de verdade também exige uma frase de confirmação exata.- Todo cache global de processo (cache de conexão, cache de testbed, snapshots de config/learn, log de operações) é protegido por um lock, para que clientes HTTP concorrentes não corrompam estado compartilhado.
- Todas as credenciais vêm de
.env— nunca armazenadas no arquivo de testbed ou no código-fonte.
Estrutura do Projeto
.
├── 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
Desenvolvimento
# 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
Veja CONTRIBUTING.md para a configuração completa e o fluxo de PR.
Benchmark
benchmark/ compara STDIO (legado) contra Streamable HTTP nos modos com e sem estado, contra um testbed real. Veja benchmark/scenarios.py para a lista de cenários e benchmark/aggregate.py para gerar o relatório de comparação; benchmark/results/summary.md tem os números da execução mais recente.