pyATS MCP Server

pyATS MCP Server

Cisco pyATS と Genie は、show コマンドの解析、設定の適用、機能の状態の学習、宣言的テストの実行など、ネットワークと通信する方法をすでに備えています。しかし、AI エージェントがこれらを直接操作する手段はありませんでした。このサーバーはそのギャップを埋めます。pyATS/Genie を、Claude のようなエージェントが実際のテストベッドに対して呼び出せる、構造化され、ガード付きの MCP ツール群としてラップします。通信には、Model Context Protocol の現在の Streamable HTTP トランスポートを使用します。

エージェントを接続すれば、デバイスの検索、show コマンドの実行と解析、ロールバックポイントを設定した構成の適用、変更前後における機能状態の学習と差分比較、スレッドプールまたはデバイスごとに 1 つのプロセスを使ったコマンドのフリート全体への一括実行、宣言的な Blitz または Robot Framework テストの実行、あるいはデバイスの REST/RESTCONF API の直接呼び出しが可能になります。リスクのある処理はすべてデバイスに到達する前にガードされ、すべての呼び出しはエージェントがセッション中に確認できるインメモリ監査ログに記録されます。

---

概要

• トランスポート — Streamable HTTP（mcp>=2.3.0）。環境変数 1 つでステートフルまたはステートレスを選択できます。STDIO は廃止されています。
• 検出、show コマンド、構成、Genie による学習と差分比較、Genie Clean、宣言的テスト（Blitz、Robot Framework、AEtest）、汎用 REST/RESTCONF、Cisco XPresso にわたる 26 個のツール。
• 多数のデバイスにコマンドを一括実行する 2 つの方法 — 日常的な用途に適した共有スレッドプール、または大規模環境で確実に分離したい場合に使う、デバイスごとに 1 つの 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 対応クライアント — 以下のエージェントを接続するを参照してください。

---

クイックスタート

# 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

MCP エンドポイントには http://<host>:<port>/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 が不明なデバイスでは、.env に DEVICE_OS=generic を設定し、os: "%ENV{DEVICE_OS}" を指定してください。 また、必要に応じて 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

いずれの場合も、サーバーは一度起動してクライアントから接続する、長時間実行されるプロセスです。エージェントがセッションごとに起動するものではありません。各クライアントからの接続方法については、以下をご覧ください。

---

エージェントを接続する

サーバーが公開するのは、http://<host>:<port>/mcp（Streamable HTTP）の MCP エンドポイントのみです。以下の各クライアントにはこの 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 のクラウドから接続するため、localhost ではなく、実際にパブリックアクセス可能な HTTPS URL が必要です。

それでも自分のマシンで実行中のサーバーに Desktop を接続するには、mcp-remote をローカルの stdio プロキシとして経由させます。

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

Python を直接使用（LangGraph、カスタムエージェント、その他）

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をご覧ください。

クライアントがリクエストでこの拡張機能を通知すると、デバイス I/O、リモートリクエスト、テスト実行に関する 22 個のツールすべてでタスクが使用されます。デバイスの一覧表示／検索、スナップショットの差分、操作履歴は、引き続き即時に返されます。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 が含まれます。同じメタデータを付けて、{"taskId": "..."} で tasks/get をポーリングしてください。そのレスポンスには resultType: "complete" が含まれます。作業が完了したかどうかを判断するには、status を確認してください。完了したタスクには、元の MCP ツールレスポンスが result に含まれます。プロトコルエラーの場合は 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 コマンドの安全な停止、設定変更の取り消し、テストのサブプロセスの強制終了はできないため、await をキャンセルしてもこれらを安全に停止できません。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 が設定され、この TTL には保持期間も含まれます。期限切れの結果は、後続のタスク送信時に削除されます。保持タスク数の上限に達すると、新たなタスクは受け付けられません。再起動時に未完了のタスクは、結果が不明であることを示すメッセージとともに失敗状態になります。自動的に再実行されることはありません。変更を再試行する前に、デバイスの状態を確認してください。正常なシャットダウンでは、キュー内のタスクをキャンセルし、実行中の操作が完了するまで待機します。

タスクデータベースごとにサーバープロセスを1つだけ実行してください。ファイルロックによってこれが強制されます。Docker では、永続ディレクトリをマウントし、PYATS_MCP_TASK_DB の保存先をそのディレクトリ内に設定してください。水平スケーリングを行う場合は、ストアを分け、各タスク ID をそれを所有するレプリカにルーティングする必要があります。この実装は分散タスクキューではありません。

データベースにはツールの結果が保存され、設定情報が含まれる場合もあります。また、所有者のみがアクセスできる権限で作成されます。認証済みのタスクは、検証済みのクライアント／発行者／サブジェクトの ID に紐付けられます。このプロジェクトのデフォルトである未認証サーバーでは、推測困難なタスク 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
• 「この Blitz テストを R1 と R2 に対して実行してください」／「この Robot Framework テストスイートを実行してください」 → pyats_run_blitz／pyats_run_robot

エージェントがこれらを自動で連携します。実現したい結果を説明すると、使用するツールを選択します。

---

利用可能なツール

26 個のツールを、用途別に分類しています。

検出

ツール	説明
pyats_list_devices	テストベッド内のすべてのデバイスを一覧表示します
pyats_search_devices	名前またはエイリアスでデバイスをあいまい検索します

show コマンド

ツール	説明
pyats_run_show_command	検証済みの show コマンドを実行し、解析済みの JSON または生の出力を返します
pyats_run_show_command_multi	複数のデバイスに対して show コマンドをスレッドプールで並行実行します
pyats_pcall_show_command	同様ですが、共有スレッドプールの代わりにデバイスごとに 1 つの OS プロセス（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	同様ですが、デバイスごとに 1 つの OS プロセスを使用します
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	機能全体（インターフェイス、ospf、bgp など）に対して Genie の device.learn() を実行します。名前付きスナップショットとして保存することもできます
pyats_diff_learned_snapshots	pyats_learn_feature で保存した 2 つのスナップショットの差分を表示します

テストと自動化

ツール	説明
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