Features

• Authenticated client and session handling for the Apstra REST API.

• Nested resource objects that map directly to REST paths.

• Generic payload generators for templates, rack types, logical devices, pools, configlets, policies, and telemetry configuration.

• Reference-design clients for Freeform and Two-Stage L3 CLOS blueprints.

• Utilities for async polling and cleanup in tests and automation.

Installation

aos-sdk-api requires Python 3.10 or newer.

pip install aos-sdk-api

Main Packages

• aos.sdk.api: common client classes and helpers. The top-level package exports Client, FreeformClient, TwoStageL3ClosClient, ClientError, Api, RestResource, RestResources, resources(), query_escape(), and with_async_state.

• aos.sdk.api.generator: payload generators for common Day 0 and Day 2 objects such as logical devices, rack types, templates, blueprints, pools, configlets, tags, and streaming configuration.

• aos.sdk.api.reference_design.freeform.client: Freeform blueprint APIs such as device profiles, systems, links, config templates, tags, and metadata.

• aos.sdk.api.reference_design.two_stage_l3clos.client: Two-Stage L3 CLOS blueprint APIs such as virtual networks, routing zones, racks, interface-map assignments, connectivity templates, telemetry helpers, and policy extensions.

• aos.sdk.api.reference_design.two_stage_l3clos.generator: helper generators for routing-zone and virtual-network workflows, including gen_vnet(), create_bound_to_payload(), and connectivity-template payload builders.

Quick Start

Use a base URL that includes the Apstra API prefix, typically /api. The examples below disable TLS verification because that is common in lab setups. In production, prefer trusted certificates and verify_certificates=True.

from aos.sdk.api import Client

client = Client(
    "https://apstra.example.com/api",
    verify_certificates=False,
)
client.login("admin", "admin")

print(client.version.get()["version"])

for blueprint in client.blueprints.list() or []:
    print(blueprint["label"], blueprint["id"])

Working With Resources

The SDK maps URL paths to nested Python objects. For example, client.blueprints[blueprint_id].nodes[node_id] maps to /blueprints/{blueprint_id}/nodes/{node_id}.

blueprint_id = "d7c4d9d4-..."
node_id = "leaf-01"

bp = client.blueprints[blueprint_id]

details = bp.get()
nodes = bp.nodes.list()
rendered_config = bp.nodes[node_id].get_config_rendering()

missing = bp.nodes["does-not-exist"].get()
assert missing is None

Useful behavior:

• Collection objects expose familiar list(), create(), options(), get(), update(), patch(), and delete() methods.

• Many collection list() calls return the unwrapped items list instead of the raw response body.

• Missing resources on GET and DELETE return None instead of raising ClientError for HTTP 404.

• Objects created through collection create() calls are registered for cleanup, so client.cleanup() is useful in tests and temporary lab automation.

Querying Blueprint Graphs

bp.query() executes QueryEngine DSL expressions against the blueprint graph. Use query_escape() when an expression includes runtime Python values. It renders values as Python literals and checks that the rendered result is a valid Python expression before the SDK sends it to Apstra. Leave replacement fields unquoted; query_escape() adds the required literal quoting.

The following query finds the virtual networks instantiated on a leaf for a local VLAN. This follows the Two-Stage L3 CLOS graph traversal used to locate a virtual network by system and VLAN ID.

from aos.sdk.api import query_escape

leaf_id = "<leaf-node-id>"
local_vlan_id = 120

virtual_network_query = query_escape(
    """match(
        node('system', id={leaf_id})
        .out('hosted_vn_instances')
        .node('vn_instance', vlan_id={local_vlan_id})
        .in_('instantiated_by')
        .node('virtual_network', name='virtual_network')
    )""",
    leaf_id=leaf_id,
    local_vlan_id=local_vlan_id,
)

virtual_networks = [
    path["virtual_network"]
    for path in bp.query(virtual_network_query)["items"]
]

This query finds the logical links that connect a system to a routing zone. The graph schema calls routing zones security_zone nodes.

system_id = "<leaf-node-id>"
routing_zone_id = "<routing-zone-id>"

logical_link_query = query_escape(
    """match(
        node('system', id={system_id})
        .out('hosted_interfaces')
        .node('interface', name='interface', if_type='port_channel')
        .out('composed_of')
        .node('interface', if_type='subinterface', name='subinterface'),

        node('security_zone', id={routing_zone_id})
        .out('instantiated_by')
        .node('sz_instance')
        .out('member_interfaces')
        .node('interface', name='subinterface')
        .out('link')
        .node('link', link_type='logical_link', name='logical_link')
    )""",
    system_id=system_id,
    routing_zone_id=routing_zone_id,
)

logical_links = [
    path["logical_link"]
    for path in bp.query(logical_link_query)["items"]
]

Creating Payloads With Generators

Generator functions build ordinary Python dictionaries and lists that are ready to pass to create(), update(), or patch(). They do not make API calls by themselves.

Common generator entry points include gen_logical_device(), gen_rack_type(), gen_rack_based_template(), gen_blueprint(), gen_ip_pool(), gen_asn_pool(), gen_security_zone(), gen_configlet(), make_configlet_generator(), and gen_streaming_config().

from aos.sdk.api import Client
from aos.sdk.api import generator as aos_gen

client = Client(
    "https://apstra.example.com/api",
    verify_certificates=False,
)
client.login("admin", "admin")

rack_type = aos_gen.gen_rack_type(
    display_name="SDK Rack",
    leafs=[
        aos_gen.gen_rack_type_leaf(
            label="leaf",
            logical_device=aos_gen.AOS_48x10_6x40,
            link_per_spine_count=1,
            link_per_spine_speed=40,
        )
    ],
    generic_systems=[],
)

template_payload = aos_gen.gen_rack_based_template(
    display_name="SDK Template",
    spine=aos_gen.gen_spine_type(
        logical_device=aos_gen.AOS_32x40,
        count=2,
    ),
    rack_types=[rack_type],
    rack_type_counts={rack_type["id"]: 2},
    virtual_network_policy=aos_gen.gen_virtual_network_policy(
        overlay_control_protocol="evpn",
    ),
)

template = client.templates.create(data=template_payload)

blueprint = client.blueprints.create(
    data=aos_gen.gen_blueprint(
        name="SDK Blueprint",
        template_id=template["id"],
    )
)

print(template["id"], blueprint["id"])

Reference Design Example

Use TwoStageL3ClosClient when you want blueprint methods that are specific to the Two-Stage L3 CLOS reference design. The reference-design generator module adds helpers for payloads such as virtual networks.

from aos.sdk.api import TwoStageL3ClosClient
from aos.sdk.api import generator as aos_gen
from aos.sdk.api.reference_design.two_stage_l3clos.generator import (
    create_bound_to_payload,
    gen_vnet,
)

client = TwoStageL3ClosClient(
    "https://apstra.example.com/api",
    verify_certificates=False,
)
client.login("admin", "admin")

blueprint = next(
    bp for bp in client.blueprints.list() or []
    if bp["label"] == "prod-fabric"
)
bp = client.blueprints[blueprint["id"]]

routing_zone = bp.security_zones.create(
    data=aos_gen.gen_security_zone(
        label="prod",
        vrf_name="prod",
        sz_type="evpn",
    )
)

vn_payload = gen_vnet(
    vn_type="vxlan",
    label="web",
    security_zone_id=routing_zone["id"],
    bound_to=[
        create_bound_to_payload("<leaf-or-rg-node-id-1>", 120),
        create_bound_to_payload("<leaf-or-rg-node-id-2>", 120),
    ],
    ipv4_enabled=True,
    ipv4_subnet="192.168.120.0/24",
    virtual_gateway_ipv4="192.168.120.1",
)

bp.virtual_networks.create(data=vn_payload)

The bound_to payload expects leaf or redundancy-group node IDs from the blueprint graph. These IDs are typically obtained from blueprint queries or list endpoints.

Waiting For Async State

Some workflows are eventually consistent. with_async_state retries a function with exponential backoff until it succeeds or the timeout expires.

from aos.sdk.api import with_async_state

def wait_for_policy():
    assert bp.endpoint_policies[policy_id].get() is not None

with_async_state(wait_for_policy, timeout=30)