Documentation

The nervous system
for your network.

Plexar is a unified, async-first Python SDK for network automation. Transport, parsing, intent, telemetry, topology, and AI — in one coherent platform.

Get Started → Core Concepts

terminal

$ pip install plexar

# or with all extras

$ pip install "plexar[all]"

01 Why Plexar?

The Python network automation ecosystem is fragmented. Netmiko for SSH, NAPALM for abstraction, Nornir for orchestration, TextFSM for parsing, pyATS for testing — each excellent at one thing, none solving the whole problem.

Plexar is designed from the ground up as a platform SDK — a single, layered system that handles every domain of network automation, with an async-first architecture and AI capabilities built in.

          API
          net.devices()
          device.get_bgp()
          intent.apply()
          ai.ask()
        

↕

INTELLIGENCE AI Engine Intent Engine Drift Detection

↕

OBSERVABILITY State Manager Topology Graph Telemetry

↕

ABSTRACTION Vendor-Neutral Models Parsers Config Engine

↕

TRANSPORT SSH NETCONF gNMI RESTCONF SNMP

↕

DEVICES Cisco Juniper Arista Palo Alto Nokia + more

02 Explore the Docs

⚡

Quickstart

Connect to your first device and run a query in under 5 minutes.

🎯

Intent Engine

Declare what you want. Plexar compiles it to vendor config automatically.

📡

Drift Detection

Continuously compare running state against desired state. Alert or auto-fix.

🤖

AI Engine

Natural language RCA, autonomous remediation, and LLM-assisted parsing.

🧪

Testing Framework

pytest-native network state assertions with a full mock driver for CI/CD.

🔌

Vendor Drivers

Supported platforms, transport matrix, and how to write your own driver.

Getting Started

Quickstart

Connect to a device, fetch structured data, and push a config change — in minutes.

01 Install

bash

pip install plexar

02 Connect to a Device

python

from plexar.core import Device, Credentials, Transport
import asyncio

async def main():
    device = Device(
        hostname="spine-01",
        management_ip="10.0.0.1",
        platform="arista_eos",
        transport=Transport.SSH,
        credentials=Credentials(
            username="admin",
            password_env="DEVICE_PASS"  # reads from $DEVICE_PASS
        )
    )

    await device.connect()

    # Fetch normalized BGP data — not raw CLI text
    bgp = await device.get_bgp_summary()

    for peer in bgp.peers:
        print(f"{peer.neighbor_ip} → {peer.state} ({peer.prefixes_received} prefixes)")

    await device.disconnect()

asyncio.run(main())
        

03 Run Across Many Devices

python

from plexar import Network

async def main():
    net = Network()
    net.inventory.load("yaml", path="./inventory.yaml")

    leafs = net.devices(role="leaf")

    # Concurrent across all leafs — async pool with rate limiting
    async with net.pool(max_concurrent=50) as pool:
        results = await pool.map(
            lambda d: d.get_bgp_summary(),
            devices=leafs
        )

    for device, bgp in results:
        down = [p for p in bgp.peers if p.state != "established"]
        if down:
            print(f"⚠ {device.hostname}: {len(down)} peers down")
        

💡

Async throughout

Plexar's pool handles concurrency, connection management, rate limiting, and retries automatically. You write business logic — not threading code.

04 Push Config with Rollback

python

async with device.transaction() as txn:
    await txn.push(config_block)

    ok = await txn.verify([
        ("bgp_peers", lambda r: r.peers_established >= 4),
        ("interfaces", lambda r: r.uplink.oper_state == "up"),
    ])

    if not ok:
        await txn.rollback()  # guaranteed — no partial state
        raise VerificationFailed("Post-push checks failed")
        

Core Modules

Intent Engine

Declare what you want. Plexar compiles it into vendor-specific configuration and verifies the outcome.

🎯

Intent-based networking

Instead of writing Cisco config for leaf-01, Arista config for leaf-02, and Juniper config for leaf-03 — you describe the desired state once and Plexar compiles it per vendor.

01 Basic Intent

python

from plexar.intent import Intent
from plexar.intent.primitives import BGPIntent, InterfaceIntent, VLANIntent

intent = Intent(devices=net.devices(role="leaf"))

# BGP — compiles to correct syntax per vendor
intent.ensure(BGPIntent(
    asn=65001,
    neighbors=["10.0.0.1", "10.0.0.2"],
    address_family="evpn",
    bfd=True
))

# Interface — normalized across all platforms
intent.ensure(InterfaceIntent(
    name="Ethernet1",
    mtu=9214,
    admin_state="up",
    description="uplink-to-spine-01"
))

# Preview what will change — per device, per vendor
plan = await intent.compile()
print(plan.diff())

# Apply and verify
result = await intent.apply()
report = await intent.verify()
print(report.compliant)  # True
        

02 Available Primitives

Primitive	Description	Status
BGPIntent	BGP peer, ASN, address families, timers, BFD	✓ stable
InterfaceIntent	Admin state, MTU, description, IP, speed	✓ stable
VLANIntent	VLAN ID, name, trunk/access assignment	✓ stable
OSPFIntent	OSPF area, process ID, passive interfaces	⚡ in progress
ACLIntent	Access list rules, application to interfaces	⚡ in progress
MPLSIntent	LDP, RSVP, segment routing	○ roadmap

Reference

Vendor Drivers

Supported platforms, transport matrix, and how to write a custom driver.

01 Support Matrix

Platform	SSH	NETCONF	RESTCONF	gNMI
cisco_ios	✓	✓	✓	⚡
cisco_nxos	✓	✓	✓	⚡
cisco_xr	✓	✓	✓	✓
arista_eos	✓	✓	✓	✓
juniper_junos	✓	✓	⚡	⚡
paloalto_panos	✓	✓	✓	—
fortinet_fortios	✓	⚡	✓	—
nokia_sros	⚡	✓	⚡	✓

✓ Stable ⚡ In Progress — Roadmap

02 Writing a Custom Driver

Drivers implement a single abstract base class. New vendor = new class. Zero core changes.

python

from plexar.drivers.base import BaseDriver
from plexar.models import Interface, BGPSummary, RoutingTable

class MyVendorDriver(BaseDriver):
    platform = "myvendor_os"

    async def connect(self) -> None:
        self._conn = await my_ssh_connect(self.device)

    async def get_interfaces(self) -> list[Interface]:
        raw = await self._conn.run("show interfaces")
        return self._parse_interfaces(raw)

    async def get_bgp_summary(self) -> BGPSummary:
        raw = await self._conn.run("show bgp summary")
        return self._parse_bgp(raw)

    # ... implement remaining abstract methods
        

📦

Register your driver

Add entry_points in your pyproject.toml under plexar.drivers and Plexar will auto-discover it at import time.

AI & Simulation

AI Engine

Natural language queries, root cause analysis, autonomous remediation, and LLM-assisted parsing for unknown CLI output.

⚡

Coming in v0.3

The AI Engine is under active development. API design is stable; implementation ships in v0.3.

01 Natural Language RCA

python

from plexar.ai import NetworkAI

ai = NetworkAI(net, llm_provider="openai")

rca = await ai.ask("Why is traffic slow between dc1 and dc2?")

print(rca.root_cause)       # "BGP prefix limit on leaf-03"
print(rca.affected_devices)  # ["leaf-03", "spine-01"]
print(rca.evidence)          # telemetry + route data used
print(rca.recommendation)    # "Increase prefix-limit or add filter"
        

02 Autonomous Remediation

python

# With human approval gate
plan = await ai.remediate(rca, require_approval=True)

print(plan.summary())   # shows exactly what it will do
await plan.approve()    # human-in-the-loop
await plan.execute()    # applies + verifies
        

03 LLM-Assisted Parser

For commands with no existing template, the AI parser extracts structured data using an LLM — returning a validated Pydantic model.

python

raw    = await device.run("show platform hardware qos queue-stats")
parsed = await ai.parse(raw, hint="QoS queue statistics")

print(parsed.queues[0].dropped_packets)  # fully typed

Testing

Testing Framework

pytest-native network state assertions. Run in CI/CD without real devices using the built-in mock driver.

01 Writing Network Tests

python

# test_network.py
import pytest
from plexar.testing import net_fixture

@pytest.mark.asyncio
async def test_all_bgp_peers_established(net):
    async for device in net.devices(role="leaf"):
        bgp = await device.get_bgp_summary()
        assert all(p.state == "established" for p in bgp.peers), \
            f"{device.hostname}: BGP peers not all established"

async def test_no_config_drift(net):
    report = await net.drift_report()
    assert report.is_clean, report.summary()

async def test_default_route_exists(net):
    core = await net.device("core-sw-01")
    await net.assert_route("0.0.0.0/0", device=core)
        

02 Mock Driver for CI/CD

Test your automation logic without any real devices. The mock driver returns configurable fixture data.

python

from plexar.drivers.mock import MockDriver
from plexar.models import BGPSummary, BGPPeer

mock = MockDriver()
mock.set_response("get_bgp_summary", BGPSummary(
    peers=[
        BGPPeer(neighbor_ip="10.0.0.1", state="established", prefixes_received=150),
        BGPPeer(neighbor_ip="10.0.0.2", state="established", prefixes_received=148),
    ]
))

# Your tests run against mock — identical API surface
device = Device(..., driver=mock)
bgp = await device.get_bgp_summary()
        

Observability

Drift Detection

Continuously compare running network state against desired state. Alert your team or auto-remediate.

01 How It Works

Define desired state — from NetBox, YAML, Git, or any inventory source
Collect running state — Plexar polls devices on your schedule
Compute diff — semantic diff engine compares field-by-field
Fire callbacks — alert, log, or auto-remediate on any drift event

02 Setup

python

from plexar.state import DriftMonitor

monitor = DriftMonitor(
    inventory=net.inventory,
    desired_state_source="netbox",
    interval_seconds=300  # check every 5 min
)

@monitor.on_drift
async def handle_drift(event):
    await slack_alert(
        channel="#network-ops",
        message=f"Drift on {event.device.hostname}:\n{event.diff.summary()}"
    )

    # Optional: auto-remediate low-risk drift
    if event.risk_score < 20:
        await event.remediate()

await monitor.start()
        

Reference

Contributing

Plexar is open source. The most impactful contribution is a new vendor driver.

01 Setup Dev Environment

bash

$ git clone https://github.com/plexar/plexar
$ cd plexar
$ pip install -e ".[dev]"
$ pytest                    # run full test suite
$ pytest tests/unit/        # unit tests only (no devices)
        

02 What We Need Most

🔌

Vendor Drivers

Nokia SR-OS, Mikrotik, Cumulus, SONiC, Huawei VRP — all need drivers.

📝

Parser Templates

TTP templates for new platforms and commands expand our coverage.

🧪

Test Cases

Real device output samples for existing parsers to catch edge cases.

📚

Documentation

Examples, tutorials, and corrections — all welcome.

Reference

Changelog

What's new in Plexar.

v0.1.0-alpha — Current

🚀

Initial Release

Core device model, async SSH transport, Cisco IOS/NX-OS, Arista EOS, and Juniper JunOS drivers. Normalized Interface, BGP, and RoutingTable models. Mock driver for testing.

v0.2.0 — Planned

NETCONF driver, config diff engine, transactional push, drift detection v1, intent engine primitives (BGP, Interface, VLAN).

v0.3.0 — Planned

AI Engine (RCA, parser, remediation), gNMI telemetry, topology graph, LLDP discovery.

The nervous systemfor your network.

01 Why Plexar?

02 Explore the Docs

Quickstart

01 Install

02 Connect to a Device

03 Run Across Many Devices

04 Push Config with Rollback

Intent Engine

01 Basic Intent

02 Available Primitives

Vendor Drivers

01 Support Matrix

02 Writing a Custom Driver

AI Engine

01 Natural Language RCA

02 Autonomous Remediation

03 LLM-Assisted Parser

Testing Framework

01 Writing Network Tests

02 Mock Driver for CI/CD

Drift Detection

01 How It Works

02 Setup

Contributing

01 Setup Dev Environment

02 What We Need Most

Changelog

v0.1.0-alpha — Current

v0.2.0 — Planned

v0.3.0 — Planned

The nervous system
for your network.