Python examples¶
Generated from eve-iac OpenAPI/semantics. Do not edit manually.
Task-oriented examples using the current public SDK. These snippets are generated with the API reference.
Working with responses¶
Methods return typed models (PlanData, BulkResult, …). Read fields as attributes (plan.plan_identity, result.started).
Failures raise AgentError with message, code, status, and optional data. Key off code, not the diagnostic string.
from eveiac import AgentError, is_auth_error
try:
plan = client.plan_project(PlanProjectRequest(**packed.payload))
except AgentError as exc:
if is_auth_error(exc):
...
if exc.code == "lab_busy":
...
CLI --json on failure writes the same error shape to stdout:
{
"error": "unauthenticated",
"code": "unauthenticated",
"status": 401
}
Import versus from_eve¶
- Import (
importProject/import_project/ImportProject) reads an existing EVE source lab into a new IaC project. It writes intent and does not deploy. It must not mutatesource_lab. - from_eve is
reconcileProjectwithdirection: "from_eve". It updates files in an existing project from that project's managed lab. It is not Import. It reads startup-config blobs already stored on the lab, not NOS memory. - exportProjectConfig copies the running NOS configuration into an existing startup-config set. It does not write
topology.ymlorconfigs/**. A later from_eve adopts that blob.
Create a client¶
from eveiac import EveIacClient
client = EveIacClient(
url="https://eve.example:8787",
token="",
ca_pem=open("ca.pem", encoding="utf-8").read(),
timeout_ms=15000,
)
Authenticate / login¶
from eveiac import EveIacClient, LoginRequest
client = EveIacClient(url=url, ca_pem=ca_pem)
session = client.login(LoginRequest(username="admin", password="secret"))
client = EveIacClient(url=url, token=session.token, ca_pem=ca_pem)
Get session¶
session = client.get_session()
List templates¶
from eveiac import ListTemplatesQuery
templates = client.list_templates()
provisioned = client.list_templates(ListTemplatesQuery(provisioned="true"))
Create LOAD project locally (no agent)¶
from eveiac import init_load, pack_lab
root = init_load("/Customers/OSPF.unl", "./IaC/ospf")
packed = pack_lab(root) # payload is manifest + lab only
Import an existing EVE-NG lab¶
from eveiac import ImportProjectRequest, pack_lab
packed = pack_lab("./imported-lab")
imported = client.import_project(ImportProjectRequest(**packed.payload))
# imported.managed_lab is the later deploy target. Import does not create it. Never mutate source_lab.
Validate¶
from eveiac import ValidateProjectRequest, pack_lab
packed = pack_lab("./my-lab")
result = client.validate_project(ValidateProjectRequest(**packed.payload))
Plan¶
from eveiac import PlanProjectRequest, pack_lab
packed = pack_lab("./my-lab")
plan = client.plan_project(PlanProjectRequest(**packed.payload))
# applicable is a summary of immediate full convergence; do not gate reconcile on it.
n_actions = len(plan.actions or plan.operations or [])
identity = plan.plan_identity # copy into later confirm=true reconcile with the same direction
Deploy¶
from eveiac import DeployProjectRequest, pack_lab
packed = pack_lab("./my-lab")
deployed = client.deploy_project(DeployProjectRequest(**packed.payload))
Reconcile to_eve¶
from eveiac import ReconcileProjectRequest, pack_lab
packed = pack_lab("./my-lab")
body = dict(packed.payload)
body["direction"] = "to_eve"
applied = client.reconcile_project(ReconcileProjectRequest(**body))
# Omitted confirm is live-safe. Remaining requires_stop: confirm=True AND matching plan_identity.
# confirm does not invent identity. Mismatch → 409 plan_stale, zero StopNode.
if getattr(applied, "status", None) == "pending_stop":
body["confirm"] = True
body["plan_identity"] = applied.plan_identity
applied = client.reconcile_project(ReconcileProjectRequest(**body))
Reconcile from_eve¶
from eveiac import ReconcileProjectRequest, pack_lab, commit_pulled, should_persist_from_eve
packed = pack_lab("./my-lab")
body = dict(packed.payload)
body["direction"] = "from_eve"
pulled = client.reconcile_project(ReconcileProjectRequest(**body))
if should_persist_from_eve(pulled):
commit_pulled(packed.root, pulled)
Export running config¶
from eveiac import pack_lab
packed = pack_lab("./my-lab")
body = dict(packed.payload)
body["configset"] = "default"
body["nodes"] = ["n_1"]
saved = client.export_project_config(body)
# Does not write topology.yml or configs/**. A later from_eve adopts the blob.
Status¶
from eveiac import GetProjectStatusRequest, pack_lab
packed = pack_lab("./my-lab")
status = client.get_project_status(GetProjectStatusRequest(**packed.payload))
live = getattr(status, "live", None)
if getattr(live, "available", None) is True:
for node in getattr(live, "nodes", None) or []:
print(f"{getattr(node, 'key', '')}: {getattr(node, 'state_label', '')}")
Inspect¶
from eveiac import InspectProjectRequest, pack_lab
packed = pack_lab("./my-lab")
observed = client.inspect_project(InspectProjectRequest(**{**packed.payload, "refresh": True}))
# Structural snapshot. Poll live node state with get_project_status.
Start nodes¶
from eveiac import ExecProjectRequest, pack_lab
packed = pack_lab("./my-lab")
result = client.exec_project(ExecProjectRequest(**{**packed.payload, "action": "start"}))
Stop nodes¶
from eveiac import ExecProjectRequest, pack_lab
packed = pack_lab("./my-lab")
result = client.exec_project(ExecProjectRequest(**{**packed.payload, "action": "stop"}))
Wipe nodes¶
from eveiac import ExecProjectRequest, pack_lab
packed = pack_lab("./my-lab")
result = client.exec_project(ExecProjectRequest(**{**packed.payload, "action": "wipe"}))
List consoles¶
from eveiac import ListProjectConsolesRequest, pack_lab
packed = pack_lab("./my-lab")
consoles = client.list_project_consoles(ListProjectConsolesRequest(**packed.payload))
Execute a console command¶
from eveiac import ExecConsoleRequest, pack_lab
packed = pack_lab("./my-lab")
data = client.exec_console(ExecConsoleRequest(**{**packed.payload, "node": "n_1", "command": "show version"}))
Stream console¶
from eveiac import AttachConsoleRequest, pack_lab
packed = pack_lab("./my-lab")
attached = client.attach_console(AttachConsoleRequest(**{**packed.payload, "node": "n_1"}))
sock = client.stream_console(attached.stream.split("/")[-1])
Error handling¶
from eveiac import AgentError, is_auth_error
try:
client.get_session()
except AgentError as exc:
if is_auth_error(exc):
raise SystemExit(f"auth failed: {exc.code}") from exc
raise