Integration guide — wiring every supporting system with code

The architecture page's "Supporting systems" table names fourteen tools and what role each plays. This page is the "now actually build it" version of that table: for each one, from-scratch setup, how a domain agent authenticates to its API, and a small MCP server that exposes that system's operations as callable tools — the same shape for every system, so a domain agent's own code doesn't need fourteen different integration styles. Section 9 does this in the most depth for Cisco ISE specifically, since NAC/RADIUS integration tends to be the one people ask about by name.

Every code block below assumes the credential-scoping guardrail from the architecture page: nothing calls a vendor API with a static, standing credential. get_scoped_credential() appears throughout as a stand-in for a real vault client (HashiCorp Vault's dynamic secrets engine, or a cloud KMS/Secrets Manager equivalent) — it checks out a credential scoped to one system and one short TTL, and the lease expires whether or not the calling code remembers to revoke it.

Model Context Protocol (MCP) is the wire format each domain agent uses to call a supporting system: one small MCP server per system, one Python process, tools scoped to exactly what that system class needs and nothing else. The orchestrator never imports a vendor SDK directly — it holds an MCP client session per domain agent, and every tool call flows through the same request → plan → tier → (approval) → execute → verify lifecycle from the architecture page, just with the "execute" step now concretely a tools/call over MCP instead of an abstract box in a diagram.

shared/mcp_base.py — the scaffold every server below imports

from mcp.server.fastmcp import FastMCP
from vault_client import get_scoped_credential  # your vault/KMS client, not shown

def build_agent_server(name: str, system_class: str) -> FastMCP:
    """One MCP server per supporting system. name becomes the MCP server
    identity the orchestrator registers; system_class is what the vault
    policy uses to decide which credentials this process is even allowed
    to check out -- enforced server-side by the vault, not by this code."""
    server = FastMCP(name)
    server.system_class = system_class
    return server

def audit(tool: str, args: dict, result: dict, ticket_id: str) -> None:
    """Every tool call appends here before returning -- the same
    append-only audit log every other agent in the architecture writes
    to. Implementation is your log store's client, not shown."""
    ...

Three conventions hold across every server on this page, because they're what make the MCP layer trustworthy rather than just convenient:

• Tool docstrings state the tier. The orchestrator reads a tool's docstring to route it through the correct approval path — a Tier 0 tool never blocks on approval, a Tier 2–3 tool always does, and that fact lives next to the code that implements it, not in a separate spec that can drift out of sync.
• Every tool call is auditable independent of the MCP transport. audit() runs inside the tool, not as an MCP-level wrapper, so the audit trail survives even if a future transport replaces MCP for a given system.
• Mutating tools take a ticket_id argument and verify it server-side. An MCP tool that changes state without a linked, already-approved ticket is exactly the "second door around the gate" the architecture page rules out — the tool itself refuses to run without one, not just the orchestrator's calling convention.

Setup, from scratch:

1. Provision a ServiceNow instance (a Personal Developer Instance is enough to build and test this whole guide against, free, from ServiceNow's developer portal).
2. Create a dedicated integration service account — never the orchestrator running as an admin or as a named human user — and assign it the rest_api_explorer, itil, and approval_admin roles, no more.
3. Under System OAuth → Application Registry, register an OAuth inbound client for the orchestrator; prefer OAuth over Basic auth for anything beyond a lab instance, since it gives you token expiry and scoped revocation for free.
4. Enable the Table API and Import Set API plugins if not already active (System Definition → Plugins), then confirm /api/now/table/incident responds for the integration account before writing any agent code.
5. Build the Zabbix → ServiceNow webhook target: a scripted REST endpoint (or ServiceNow's built-in webhook inbound action) that creates an Incident from a Zabbix trigger payload, tagged with the host and metric — this is what lets a ticket originate from monitoring instead of a person, per the architecture page's walkthrough.

API auth & a scoped call (OAuth client-credentials grant)

import httpx

TOKEN_URL = "https://yourinstance.service-now.com/oauth_token.do"
API_BASE = "https://yourinstance.service-now.com/api/now"

def get_token(client_id: str, client_secret: str) -> str:
    resp = httpx.post(TOKEN_URL, data={
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
    })
    resp.raise_for_status()
    return resp.json()["access_token"]

def create_incident(token: str, short_description: str, cmdb_ci: str) -> dict:
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    body = {"short_description": short_description, "cmdb_ci": cmdb_ci, "category": "monitoring"}
    resp = httpx.post(f"{API_BASE}/table/incident", json=body, headers=headers)
    resp.raise_for_status()
    return resp.json()["result"]

servicenow_mcp_server.py

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("servicenow-intake", system_class="servicenow")
BASE = "https://yourinstance.service-now.com/api/now"

def _client() -> httpx.Client:
    cred = get_scoped_credential(system="servicenow", ttl_seconds=300)
    return httpx.Client(base_url=BASE, headers={"Authorization": f"Bearer {cred.token}"})

@mcp.tool()
def get_ticket(ticket_id: str) -> dict:
    """Tier 0 -- read a ticket's current fields, including CMDB CI and priority."""
    with _client() as c:
        r = c.get(f"/table/incident/{ticket_id}")
        r.raise_for_status()
        return r.json()["result"]

@mcp.tool()
def create_approval(ticket_id: str, plan_summary: str, dry_run_ref: str) -> dict:
    """Tier 2-3 -- opens a native ServiceNow Approval on the ticket, attaching the
    dry-run artifact reference. The orchestrator polls this record's state rather
    than building a bespoke approval UI."""
    with _client() as c:
        body = {"source_table": "incident", "sysapproval": ticket_id,
                 "comments": plan_summary, "u_dry_run_ref": dry_run_ref}
        r = c.post("/table/sysapproval_approver", json=body)
        r.raise_for_status()
        result = r.json()["result"]
        audit("create_approval", locals(), result, ticket_id)
        return result

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Deploy via the official netbox-community/netbox-docker compose file (Postgres + Redis + the NetBox app) — this is the supported path and avoids hand-rolling the app's own dependency stack.
2. On first login, create the site/region/rack hierarchy that matches your actual physical or logical topology before importing a single device — NetBox's data model expects that scaffolding to exist first.
3. Under Admin → API Tokens, mint a token for the orchestrator's read path and a separate, narrower one for the Network agent's write path (device/interface/IP updates only) — two tokens, two blast radii, same split as every other domain agent's credential.
4. Bulk-import existing inventory via NetBox's CSV import (Devices → Import) rather than API-scripting the initial load; use the API for ongoing sync once the baseline is in.

API auth & a call (pynetbox)

import pynetbox

nb = pynetbox.api("https://netbox.internal.example.com", token="REPLACE_WITH_VAULT_LEASE")

def get_device_by_name(name: str):
    return nb.dcim.devices.get(name=name)

def next_available_ip(prefix: str) -> str:
    p = nb.ipam.prefixes.get(prefix=prefix)
    return str(p.available_ips.create())

netbox_mcp_server.py

import pynetbox
from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential

mcp = build_agent_server("netbox-inventory", system_class="netbox")

def _nb():
    cred = get_scoped_credential(system="netbox", ttl_seconds=300)
    return pynetbox.api("https://netbox.internal.example.com", token=cred.token)

@mcp.tool()
def lookup_device(hostname: str) -> dict:
    """Tier 0 -- inventory lookup the Network agent reads before it writes,
    per the architecture page's supporting-systems table."""
    dev = _nb().dcim.devices.get(name=hostname)
    return dict(dev) if dev else {}

@mcp.tool()
def onboard_device(hostname: str, device_type: str, site: str, ticket_id: str) -> dict:
    """Tier 1 -- registers a new device record. Does not touch the device
    itself; this is inventory bookkeeping, not a config push, which is why
    it's Tier 1 rather than Tier 2 despite creating a record."""
    result = dict(_nb().dcim.devices.create(
        name=hostname, device_type=device_type, site=site, status="active"))
    audit("onboard_device", locals(), result, ticket_id)
    return result

if __name__ == "__main__":
    mcp.run()

device onboarding steps (per device)

1. Create the device-type record first if it doesn't exist (manufacturer, model, rack unit height) — devices can't be created against a type NetBox doesn't know.
2. Create the device against a site and rack position, then attach its interfaces (bulk-add is faster than one-by-one for a chassis with many ports).
3. Assign IPs from the correct IPAM prefix and mark the device's primary IP — this is the field the Network agent and Zabbix both key off of.
4. Tag the device with its risk-tier default and owning team, so Platform Ops's drift diff has that context without a second lookup.

Setup, from scratch:

1. Deploy the Zabbix server + web frontend + a MySQL/PostgreSQL backend (the official Docker Compose stack is the fastest path); Zabbix agents (or agentless SNMP/IPMI checks) go on every managed host afterward.
2. In the frontend, create an API-only user (Users → Users) with a role scoped to the host groups the orchestrator needs to read, not Super Admin.
3. Define host groups that mirror the domain-agent boundaries (Linux servers, Windows servers, network gear) so a trigger's host group alone is enough signal for the orchestrator's classifier to route it correctly.
4. Build the outbound webhook media type (Alerts → Media types → Webhook) that POSTs a trigger event to the ServiceNow scripted REST endpoint from Section 3 — this is the other half of the Zabbix → ServiceNow Incident path.

API auth & a call (JSON-RPC)

import httpx

ZBX = "https://zabbix.internal.example.com/api_jsonrpc.php"

def zbx_call(method: str, params: dict, auth: str = None) -> dict:
    payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
    if auth:
        payload["auth"] = auth
    r = httpx.post(ZBX, json=payload, headers={"Content-Type": "application/json-rpc"})
    r.raise_for_status()
    return r.json()["result"]

def login(user: str, password: str) -> str:
    return zbx_call("user.login", {"username": user, "password": password})

def active_problems(auth: str, host_group_id: str) -> list:
    return zbx_call("problem.get", {"groupids": [host_group_id], "recent": True}, auth)

zabbix_mcp_server.py

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("zabbix-monitoring", system_class="zabbix")
ZBX = "https://zabbix.internal.example.com/api_jsonrpc.php"

def _call(method: str, params: dict) -> dict:
    cred = get_scoped_credential(system="zabbix", ttl_seconds=300)
    payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1, "auth": cred.token}
    r = httpx.post(ZBX, json=payload)
    r.raise_for_status()
    return r.json()["result"]

@mcp.tool()
def host_problems(host: str) -> list:
    """Tier 0 -- Platform Ops's heartbeat/health source, per the
    architecture page's supporting-systems table."""
    hosts = _call("host.get", {"filter": {"host": [host]}})
    if not hosts:
        return []
    return _call("problem.get", {"hostids": [hosts[0]["hostid"]], "recent": True})

@mcp.tool()
def ack_problem(event_id: str, message: str, ticket_id: str) -> dict:
    """Tier 1 -- acknowledges an event with a note linking it to the
    orchestrator's ticket, so a human glancing at Zabbix sees it's owned."""
    result = _call("event.acknowledge", {"eventids": [event_id],
                     "message": message, "action": 6})
    audit("ack_problem", locals(), result, ticket_id)
    return result

if __name__ == "__main__":
    mcp.run()

host onboarding steps

1. Install and start the Zabbix agent (or configure the SNMP/IPMI check) on the target host.
2. Create the host record in the frontend (or via host.create), assign it to the correct host group, and link the item/trigger templates that match its role (Linux server, network device, hypervisor).
3. Confirm the host shows "Zabbix agent is available" (green) before relying on any of its triggers — a silently unreachable agent is worse than no monitoring, since Platform Ops will read its silence as "healthy" rather than "unknown."

Setup, from scratch:

1. Deploy Grafana OSS (Docker image or package) and add Zabbix as a data source via the community Zabbix plugin (alexanderzobnin-zabbix-app) using the API-only account created in Section 5.
2. Add a second data source pointed at the audit log store (a Postgres/Elasticsearch panel, depending what the log is backed by) — this is what turns the audit log from a compliance artifact into something a human approver actually looks at day to day.
3. Create a service account (Administration → Service accounts) with the Viewer role for any read-only integration and mint a scoped API token from it — Grafana deprecated standalone API keys in favor of this model.
4. Build the health dashboard the architecture page's mockup shows: per-agent heartbeat panel, open-ticket-by-tier panel, and the config-drift-vs-NetBox panel, each pointed at the data sources above.

API auth & a call

import httpx

GRAFANA = "https://grafana.internal.example.com"

def get_dashboard(uid: str, token: str) -> dict:
    r = httpx.get(f"{GRAFANA}/api/dashboards/uid/{uid}",
                   headers={"Authorization": f"Bearer {token}"})
    r.raise_for_status()
    return r.json()

grafana_mcp_server.py

from shared.mcp_base import build_agent_server
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("grafana-dashboards", system_class="grafana")
GRAFANA = "https://grafana.internal.example.com"

@mcp.tool()
def dashboard_snapshot_link(uid: str) -> str:
    """Tier 0 -- generates a shareable snapshot URL of a live dashboard, so an
    approval record can link the exact state a human saw when they approved,
    not just a static description of it."""
    cred = get_scoped_credential(system="grafana", ttl_seconds=300)
    headers = {"Authorization": f"Bearer {cred.token}", "Content-Type": "application/json"}
    r = httpx.post(f"{GRAFANA}/api/snapshots",
                    json={"dashboard": {"uid": uid}, "expires": 3600},
                    headers=headers)
    r.raise_for_status()
    return r.json()["url"]

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Run the batfish/allinone Docker image — it bundles the Batfish service the Python client below talks to over gRPC/REST internally.
2. Export current running configs for the devices in scope (via the Network agent's read-only config-pull path) into a snapshot directory laid out per Batfish's expected structure (configs/, optional hosts/, layer1_topology.json).
3. Initialize a snapshot from that directory and confirm it parses cleanly — Batfish will report unrecognized lines per-vendor, which is worth fixing before trusting any diff it produces.
4. Re-initialize a fresh snapshot on every proposed change rather than mutating one in place, so "before" and "after" are always two independently reproducible snapshots, not a moving target.

API auth & a call (pybatfish)

from pybatfish.client.session import Session
from pybatfish.question import bfq

bf = Session(host="batfish.internal.example.com")
bf.set_network("prod-campus")
bf.init_snapshot("/snapshots/2026-08-26-pre-change", name="pre_change", overwrite=True)

def acl_diff(pre: str, post: str, filter_name: str):
    bf.set_snapshot(pre)
    before = bfq.filterLineReachability(filters=filter_name).answer()
    bf.set_snapshot(post)
    after = bfq.filterLineReachability(filters=filter_name).answer()
    return before, after

batfish_mcp_server.py

from shared.mcp_base import build_agent_server, audit
from pybatfish.client.session import Session

mcp = build_agent_server("batfish-verify", system_class="batfish")
bf = Session(host="batfish.internal.example.com")

@mcp.tool()
def dry_run_config_diff(network: str, pre_snapshot: str, post_snapshot: str, ticket_id: str) -> dict:
    """Tier 0 -- this IS the mandatory dry-run for Tier >=2 network changes the
    architecture page requires: a modeled diff against a real topology copy,
    computed before anything touches a live device. Its output is what the
    human approver actually reviews."""
    bf.set_network(network)
    bf.set_snapshot(pre_snapshot)
    before_routes = bf.q.routes().answer().frame()
    bf.set_snapshot(post_snapshot)
    after_routes = bf.q.routes().answer().frame()
    diff = {"routes_added": len(after_routes) - len(before_routes)}
    audit("dry_run_config_diff", locals(), diff, ticket_id)
    return diff

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Stand up Ansible Automation Platform (or plain ansible-core plus ansible-runner for a lighter footprint) on the isolated execution segment the architecture page's deployment guide calls out — agents should never run playbooks from a box that also has flat network access.
2. Build the dynamic inventory source from NetBox (the official netbox.netbox inventory plugin) rather than a static file, so a device onboarded in Section 4 shows up here automatically.
3. Write playbooks idempotently and pair every mutating playbook with an explicit rollback playbook — this is the "paired undo step" the architecture page's guardrails require, made concrete.
4. If using AAP/Controller, create a service account with a Job Template launched only for a specific playbook and limited inventory scope; if using bare ansible-runner, the scoping happens at the vault-issued SSH cert/credential level instead.

running a playbook (ansible-runner, Python API)

import ansible_runner

def run_playbook(playbook: str, inventory: str, host_limit: str, extravars: dict):
    return ansible_runner.run(
        private_data_dir="/opt/agent-runspace",
        playbook=playbook,
        inventory=inventory,
        limit=host_limit,
        extravars=extravars,
        quiet=True,
    )

ansible_mcp_server.py

from shared.mcp_base import build_agent_server, audit
import ansible_runner

mcp = build_agent_server("ansible-exec", system_class="ansible")

@mcp.tool()
def check_mode_diff(playbook: str, host_limit: str, ticket_id: str) -> dict:
    """Tier 0 -- runs the playbook with --check --diff, Ansible's own dry-run
    mode. For Linux Server agent actions this is the Tier >=2 dry-run
    artifact, the same role Batfish's diff plays for the Network agent."""
    result = ansible_runner.run(private_data_dir="/opt/agent-runspace",
        playbook=playbook, limit=host_limit, cmdline="--check --diff", quiet=True)
    out = {"status": result.status, "rc": result.rc}
    audit("check_mode_diff", locals(), out, ticket_id)
    return out

@mcp.tool()
def execute_playbook(playbook: str, host_limit: str, ticket_id: str) -> dict:
    """Tier 1-2 depending on playbook -- the real run, only ever called after
    check_mode_diff has been reviewed (Tier 1) or approved (Tier >=2)."""
    result = ansible_runner.run(private_data_dir="/opt/agent-runspace",
        playbook=playbook, limit=host_limit, quiet=True)
    out = {"status": result.status, "rc": result.rc}
    audit("execute_playbook", locals(), out, ticket_id)
    return out

if __name__ == "__main__":
    mcp.run()

This is the system named directly as an example, so it gets the fullest treatment: every step from an empty rack to an MCP tool the Identity agent can call.

9.1 — Setup, from scratch:

1. Deploy ISE (physical appliance, Cisco-provided OVA, or AWS/Azure image) as at least a two-node deployment — Primary Administration Node and Policy Service Node, split even in a lab, since ISE's HA behavior differs meaningfully from a single-node install.
2. Under Administration → System → Settings → ERS Settings, enable the External RESTful Services (ERS) API — it's off by default. Note this is separate from ISE's newer Open API; ERS remains the broadest-coverage API for identity/endpoint operations as of this writing.
3. Create an ERS-admin API user under Administration → System → Admin Access → Administrators, assigned the ERS-Admin (or narrower ERS-Operator for read-only) group — not a shared admin login.
4. Add each network access device (switch, WLC) as a NAD under Administration → Network Resources → Network Devices, with its RADIUS shared secret pulled from vault, matching what's configured on the device side (radius-server host ... key ...).
5. Build the authentication and authorization policy sets (Policy → Policy Sets) that define what "Quarantine" actually restricts — typically a downloadable ACL (dACL) or a redirect ACL to a remediation portal — before wiring any automation that references that identity group by name.
6. If real-time session visibility is needed beyond what polling the ERS API gives you, additionally enable pxGrid (Administration → pxGrid Services) and register a pxGrid client certificate for the Identity agent — pxGrid pushes session and posture events rather than requiring the agent to poll.

9.2 — API authentication & a call (ERS API, basic auth over TLS + client cert):

import httpx

ISE_BASE = "https://ise01.internal.example.com:9060/ers/config"

def ise_client(username: str, password: str) -> httpx.Client:
    return httpx.Client(
        base_url=ISE_BASE,
        auth=(username, password),
        headers={"Accept": "application/json", "Content-Type": "application/json"},
        verify="/etc/pki/ise-ca-bundle.pem",
    )

def get_endpoint(client: httpx.Client, mac_address: str) -> dict:
    r = client.get("/endpoint", params={"filter": f"mac.EQ.{mac_address}"})
    r.raise_for_status()
    return r.json()

9.3 — MCP server (the Identity agent's ISE-facing tools):

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("cisco-ise-nac", system_class="cisco-ise")
ISE_BASE = "https://ise01.internal.example.com:9060/ers/config"

def _client() -> httpx.Client:
    cred = get_scoped_credential(system="cisco-ise", ttl_seconds=300)
    return httpx.Client(base_url=ISE_BASE, auth=(cred.username, cred.password),
                         verify="/etc/pki/ise-ca-bundle.pem")

@mcp.tool()
def endpoint_status(mac_address: str) -> dict:
    """Tier 0 -- read-only lookup of an endpoint's identity group and posture
    status. Extends the Identity agent from 'does this account exist' to 'is
    this device allowed on the network right now,' per the architecture
    page's supporting-systems table."""
    with _client() as c:
        r = c.get("/endpoint", params={"filter": f"mac.EQ.{mac_address}"})
        r.raise_for_status()
        return r.json()

@mcp.tool()
def quarantine_endpoint(mac_address: str, ticket_id: str) -> dict:
    """Tier 2 -- moves an endpoint into the Quarantine identity group via
    ISE's Adaptive Network Control (ANC) API, forcing session reauthorization
    against the redirect policy set up in 9.1. Requires an approved change;
    the tool checks the ticket's approval state before calling ISE, it does
    not trust the caller to have already checked."""
    with _client() as c:
        approved = c.get(f"/servicenow-relay/ticket/{ticket_id}/approved").json()
        if not approved.get("approved"):
            raise PermissionError(f"ticket {ticket_id} is not in an approved state")
        body = {"OperationAdditionalData": {"additionalData": [
            {"name": "macAddress", "value": mac_address},
            {"name": "ancPolicy", "value": "Quarantine"},
        ]}}
        r = c.put("/ancendpoint/apply", json=body)
        r.raise_for_status()
        result = r.json()
        audit("quarantine_endpoint", locals(), result, ticket_id)
        return result

@mcp.tool()
def unquarantine_endpoint(mac_address: str, ticket_id: str) -> dict:
    """Tier 1 -- clearing a quarantine is lower blast radius than imposing
    one (it can only restore prior access, not remove it), so it carries a
    lighter tier than quarantine_endpoint above."""
    with _client() as c:
        r = c.put("/ancendpoint/clear", json={"OperationAdditionalData":
            {"additionalData": [{"name": "macAddress", "value": mac_address}]}})
        r.raise_for_status()
        result = r.json()
        audit("unquarantine_endpoint", locals(), result, ticket_id)
        return result

if __name__ == "__main__":
    mcp.run()

9.4 — onboarding one more device once ISE itself is running:

1. Add the switch/WLC as a Network Device in ISE with its RADIUS shared secret.
2. On the device itself, point AAA at the ISE Policy Service Node(s) (radius server / aaa group server radius ise on IOS-XE) and enable 802.1X on the relevant interfaces or SSID.
3. Confirm a test endpoint authenticates and lands in the expected identity group in Operations → RADIUS → Live Logs before relying on any automation against it.
4. Only after that manual confirmation, register the device's identity-group membership expectations in the Identity agent's policy config, so endpoint_status above has a known-good baseline to diff against.

Setup, from scratch:

1. Deploy the Nexus Dashboard cluster (physical, virtual, or cloud form factor — a 3-node cluster is the supported minimum for production) and add it to the ACI/NX-OS fabric's management network.
2. Install the specific services this design needs on top of the platform — Nexus Dashboard Orchestrator for multi-fabric config/telemetry, Nexus Dashboard Insights for anomaly detection — since "Nexus Dashboard" alone is the platform, not a single monolithic API surface.
3. Under the platform's own admin console, create a local user or bind to your existing identity provider (SAML/OIDC), then generate a REST API bearer token scoped to the Network agent's DC-fabric role.
4. Onboard each fabric/site into the Orchestrator so it appears as a managed site before any automation references it by name.

API auth & a call (representative shape — confirm exact paths against your installed version's API docs, since these have moved across ND releases)

import httpx

ND_BASE = "https://nd.internal.example.com/api/v1"

def login(username: str, password: str) -> str:
    r = httpx.post(f"{ND_BASE}/login", json={"userName": username, "userPasswd": password})
    r.raise_for_status()
    return r.json()["token"]

def fabric_health(token: str, fabric_name: str) -> dict:
    r = httpx.get(f"{ND_BASE}/insights/fabric/{fabric_name}/health",
                   headers={"Authorization": f"Bearer {token}"})
    r.raise_for_status()
    return r.json()

nexus_dashboard_mcp_server.py

from shared.mcp_base import build_agent_server
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("nexus-dashboard-fabric", system_class="nexus-dashboard")
ND_BASE = "https://nd.internal.example.com/api/v1"

@mcp.tool()
def fabric_health(fabric_name: str) -> dict:
    """Tier 0 -- DC fabric telemetry, scoped to the Network agent, distinct
    from the box-by-box campus/branch config the base Network agent already
    handles, per the architecture page's supporting-systems table."""
    cred = get_scoped_credential(system="nexus-dashboard", ttl_seconds=300)
    r = httpx.get(f"{ND_BASE}/insights/fabric/{fabric_name}/health",
                   headers={"Authorization": f"Bearer {cred.token}"})
    r.raise_for_status()
    return r.json()

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Deploy FMC (physical, virtual, or cloud-delivered via Cisco Defense Orchestrator) and register every managed FTD/ASA device to it — a device left in "pending registration" won't accept a deployed policy.
2. Under System → Users, create a dedicated API-only user with the minimum role (Access Admin, or a custom role scoped to access-control-policy read/write) rather than the default Administrator.
3. Enable the REST API (System → Configuration → REST API Preferences) if it isn't already, and confirm it at /api/fmc_platform/v1/info before writing any agent code.
4. Give the Firewall agent its own dedicated access-control-policy layer for temporary rule adds, so its rollback is "remove this layer's rules," not untangling a shared policy shared with human-managed rules.

API auth & a call (FMC session-token auth)

import httpx

FMC_BASE = "https://fmc.internal.example.com/api"

def login(username: str, password: str) -> tuple[str, str]:
    r = httpx.post(f"{FMC_BASE}/fmc_platform/v1/auth/generatetoken", auth=(username, password))
    r.raise_for_status()
    return r.headers["X-auth-access-token"], r.headers["DOMAIN_UUID"]

def list_access_rules(token: str, domain_uuid: str, policy_id: str) -> dict:
    r = httpx.get(f"{FMC_BASE}/fmc_config/v1/domain/{domain_uuid}"
                   f"/policy/accesspolicies/{policy_id}/accessrules",
                   headers={"X-auth-access-token": token})
    r.raise_for_status()
    return r.json()

fmc_mcp_server.py

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("firewall-fmc", system_class="cisco-fmc")
FMC_BASE = "https://fmc.internal.example.com/api"

def _session() -> tuple[str, str]:
    cred = get_scoped_credential(system="cisco-fmc", ttl_seconds=300)
    r = httpx.post(f"{FMC_BASE}/fmc_platform/v1/auth/generatetoken",
                    auth=(cred.username, cred.password))
    r.raise_for_status()
    return r.headers["X-auth-access-token"], r.headers["DOMAIN_UUID"]

@mcp.tool()
def add_temporary_rule(policy_id: str, rule: dict, ticket_id: str) -> dict:
    """Tier 3 -- adds a rule to the Firewall agent's dedicated temporary-rule
    layer. Firewall changes default to Tier 3 on the architecture page's tier
    matrix, so this needs two-person approval and a dry-run diff before it's
    ever called, same as every other Tier 3 action in this design."""
    token, domain = _session()
    r = httpx.post(f"{FMC_BASE}/fmc_config/v1/domain/{domain}"
                     f"/policy/accesspolicies/{policy_id}/accessrules",
                     json=rule, headers={"X-auth-access-token": token})
    r.raise_for_status()
    result = r.json()
    audit("add_temporary_rule", locals(), result, ticket_id)
    return result

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Deploy the Catalyst Center appliance (formerly DNA Center) and complete first-run onboarding against your fabric's underlay before importing any device.
2. Add devices via Inventory → Add Device or auto-discovery so they show as "Managed" — an unmanaged device won't accept a template push.
3. Under System → Users & Roles, split a read-only (assurance/health) role from a provision (template deployment) role, matching the read/write split every other system in this guide uses.
4. Generate a token via POST /dna/system/api/v1/auth/token, which every subsequent Intent API call bearer-authenticates with.

API auth & a call

import httpx

DNAC_BASE = "https://catalyst-center.internal.example.com"

def get_token(username: str, password: str) -> str:
    r = httpx.post(f"{DNAC_BASE}/dna/system/api/v1/auth/token", auth=(username, password))
    r.raise_for_status()
    return r.json()["Token"]

def fabric_health(token: str, site_id: str) -> dict:
    r = httpx.get(f"{DNAC_BASE}/dna/intent/api/v1/network-health",
                   params={"siteId": site_id}, headers={"X-Auth-Token": token})
    r.raise_for_status()
    return r.json()

catalyst_center_mcp_server.py

from shared.mcp_base import build_agent_server
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("network-catalyst-center", system_class="catalyst-center")
DNAC_BASE = "https://catalyst-center.internal.example.com"

@mcp.tool()
def fabric_health(site_id: str) -> dict:
    """Tier 0 -- SD-Access fabric/site health, scoped to the Network agent;
    the campus/branch counterpart to Nexus Dashboard's DC-fabric role above."""
    cred = get_scoped_credential(system="catalyst-center", ttl_seconds=300)
    r = httpx.post(f"{DNAC_BASE}/dna/system/api/v1/auth/token", auth=(cred.username, cred.password))
    r.raise_for_status()
    token = r.json()["Token"]
    r = httpx.get(f"{DNAC_BASE}/dna/intent/api/v1/network-health",
                   params={"siteId": site_id}, headers={"X-Auth-Token": token})
    r.raise_for_status()
    return r.json()

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Deploy Aria Operations (formerly vRealize Operations) for read-only capacity/performance data, and Aria Automation only if the VMware agent will also trigger workflows — they're separate products under the Aria umbrella, not one appliance.
2. In Aria Operations, bind to your existing SSO (vCenter SSO/Workspace ONE Access) or create a local user scoped to the read-only "Content Consumer" role for the agent's query path.
3. Register the target vCenter(s) as a data source (Administration → Cloud Accounts) before querying any object's metrics.
4. In Aria Automation, publish only the specific catalog items the VMware agent is allowed to trigger (e.g. "resize VM"), scoped to a dedicated service-account entitlement — never expose the full workflow designer to an automated caller.

API auth & a call

import httpx

ARIA_OPS_BASE = "https://aria-ops.internal.example.com/suite-api/api"

def login(username: str, password: str) -> str:
    r = httpx.post(f"{ARIA_OPS_BASE}/auth/token/acquire",
                    json={"username": username, "password": password})
    r.raise_for_status()
    return r.json()["token"]

def resource_capacity(token: str, resource_id: str) -> dict:
    r = httpx.get(f"{ARIA_OPS_BASE}/resources/{resource_id}/stats",
                   headers={"Authorization": f"vRealizeOpsToken {token}"})
    r.raise_for_status()
    return r.json()

vmware_aria_mcp_server.py

from shared.mcp_base import build_agent_server
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("vmware-aria", system_class="vmware-aria")
ARIA_OPS_BASE = "https://aria-ops.internal.example.com/suite-api/api"

@mcp.tool()
def resource_capacity(resource_id: str) -> dict:
    """Tier 0 -- trend-based capacity/performance data for one VM or cluster,
    scoped to the VMware agent, above raw vCenter API's point-in-time state."""
    cred = get_scoped_credential(system="vmware-aria", ttl_seconds=300)
    r = httpx.post(f"{ARIA_OPS_BASE}/auth/token/acquire",
                    json={"username": cred.username, "password": cred.password})
    r.raise_for_status()
    token = r.json()["token"]
    r = httpx.get(f"{ARIA_OPS_BASE}/resources/{resource_id}/stats",
                   headers={"Authorization": f"vRealizeOpsToken {token}"})
    r.raise_for_status()
    return r.json()

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Deploy the SCOM management group (management server, operations database, reporting) and import only the management packs for the OS/app classes actually being monitored, not every pack SCOM ships.
2. Enable the SCOM REST API (built into SCOM 2019+; earlier versions need the separately-installed Web Console API) and confirm it at /OperationsManager/authenticate before writing agent code.
3. Create a SCOM Operator-role account for read access (alert/state queries) — never a SCOM Administrator account for an integration that only reads alerts.
4. Build the SCOM → ServiceNow alert connector via a Notification Channel webhook, mirroring the Zabbix → ServiceNow path from Section 3, so a SCOM alert opens an Incident the same way.

API auth & a call

import httpx

SCOM_BASE = "https://scom.internal.example.com/OperationsManager"

def authenticate(username: str, password: str) -> httpx.Cookies:
    r = httpx.post(f"{SCOM_BASE}/authenticate", auth=(username, password))
    r.raise_for_status()
    return r.cookies

def active_alerts(cookies: httpx.Cookies) -> dict:
    r = httpx.get(f"{SCOM_BASE}/data/alert", cookies=cookies,
                   params={"criteria": "ResolutionState = '0'"})
    r.raise_for_status()
    return r.json()

scom_mcp_server.py

from shared.mcp_base import build_agent_server
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("scom-monitoring", system_class="scom")
SCOM_BASE = "https://scom.internal.example.com/OperationsManager"

@mcp.tool()
def active_alerts(criteria: str = "ResolutionState = '0'") -> dict:
    """Tier 0 -- current unresolved SCOM alerts, feeding Platform Ops the
    same way a Zabbix trigger does for the mixed/legacy Microsoft estate."""
    cred = get_scoped_credential(system="scom", ttl_seconds=300)
    r = httpx.post(f"{SCOM_BASE}/authenticate", auth=(cred.username, cred.password))
    r.raise_for_status()
    r = httpx.get(f"{SCOM_BASE}/data/alert", cookies=r.cookies, params={"criteria": criteria})
    r.raise_for_status()
    return r.json()

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Register an app in Microsoft Entra ID for the integration and grant it the Monitoring Reader role, scoped to the specific subscription or resource group it needs, not the whole tenant.
2. If querying Log Analytics, also grant "Log Analytics Reader" on the specific workspace and note its workspace ID — the Log Analytics query API is separate from the Azure Resource Manager metrics API used below.
3. For hybrid (on-prem/other-cloud) hosts, onboard them via Azure Arc so the same Monitor pipeline covers them, instead of running a second monitoring stack for non-Azure resources.
4. Build the Azure Monitor → ServiceNow path via an Action Group calling a webhook action, the same pattern as Zabbix and SCOM above.

API auth & a call (OAuth client-credentials grant)

import httpx

TENANT_ID = "your-tenant-id"

def get_token(client_id: str, client_secret: str) -> str:
    r = httpx.post(f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",
        data={"grant_type": "client_credentials", "client_id": client_id,
              "client_secret": client_secret, "scope": "https://management.azure.com/.default"})
    r.raise_for_status()
    return r.json()["access_token"]

def resource_metrics(token: str, resource_id: str, metric: str) -> dict:
    r = httpx.get(f"https://management.azure.com{resource_id}/providers/microsoft.insights/metrics",
                   params={"api-version": "2018-01-01", "metricnames": metric},
                   headers={"Authorization": f"Bearer {token}"})
    r.raise_for_status()
    return r.json()

azure_monitor_mcp_server.py

from shared.mcp_base import build_agent_server
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("azure-monitor", system_class="azure-monitor")
TENANT_ID = "your-tenant-id"

def _token() -> str:
    cred = get_scoped_credential(system="azure-monitor", ttl_seconds=300)
    r = httpx.post(f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",
        data={"grant_type": "client_credentials", "client_id": cred.client_id,
              "client_secret": cred.client_secret, "scope": "https://management.azure.com/.default"})
    r.raise_for_status()
    return r.json()["access_token"]

@mcp.tool()
def resource_metrics(resource_id: str, metric: str) -> dict:
    """Tier 0 -- cloud-native metrics for one Azure or Arc-connected resource,
    the cloud counterpart to SCOM/Zabbix above for workloads that moved to Azure."""
    r = httpx.get(f"https://management.azure.com{resource_id}/providers/microsoft.insights/metrics",
                   params={"api-version": "2018-01-01", "metricnames": metric},
                   headers={"Authorization": f"Bearer {_token()}"})
    r.raise_for_status()
    return r.json()

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Install the MECM (Configuration Manager, formerly SCCM) admin console and confirm the site server's WMI provider is reachable — MECM's automation surface is WMI/PowerShell-based; the newer AdminService REST endpoint covers only a subset of operations, unlike the unified REST APIs above.
2. Confirm the AdminService REST endpoint is enabled (on by default in current builds) at /AdminService/wmi/ for the read/query operations it does cover.
3. Create a dedicated service account with the built-in "Read-only Analyst" role for query paths, and a separate, narrower custom role (scoped to software-update deployment only) for the Desktop agent's write path.
4. Build a dedicated collection the Desktop agent is allowed to target for patch/app deployment, rather than letting it query or act against "All Systems" by default.

API auth & a call (AdminService REST, Windows-integrated auth)

import httpx
from requests_negotiate_sspi import HttpNegotiateAuth  # Windows-integrated auth

MECM_BASE = "https://mecm.internal.example.com/AdminService/wmi"

def device_compliance(collection_id: str) -> dict:
    r = httpx.get(f"{MECM_BASE}/SMS_CollectionMember_a",
                   params={"$filter": f"CollectionID eq '{collection_id}'"},
                   auth=HttpNegotiateAuth())
    r.raise_for_status()
    return r.json()

mecm_mcp_server.py

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx
from requests_negotiate_sspi import HttpNegotiateAuth

mcp = build_agent_server("desktop-mecm", system_class="mecm")
MECM_BASE = "https://mecm.internal.example.com/AdminService/wmi"

@mcp.tool()
def device_compliance(collection_id: str) -> dict:
    """Tier 0 -- patch/app compliance state for a scoped device collection."""
    r = httpx.get(f"{MECM_BASE}/SMS_CollectionMember_a",
                   params={"$filter": f"CollectionID eq '{collection_id}'"},
                   auth=HttpNegotiateAuth())
    r.raise_for_status()
    return r.json()

@mcp.tool()
def deploy_update(collection_id: str, update_group_id: str, ticket_id: str) -> dict:
    """Tier 2 -- deploys a software update group to a scoped collection;
    covers the on-prem/co-managed half of the Windows fleet that Intune
    (Section 18's domain-agent table) doesn't reach on its own."""
    cred = get_scoped_credential(system="mecm", ttl_seconds=300)
    r = httpx.post(f"{MECM_BASE}/SMS_UpdateGroupAssignment",
                    json={"CollectionID": collection_id, "AssignedUpdateGroup": update_group_id},
                    auth=HttpNegotiateAuth())
    r.raise_for_status()
    result = r.json()
    audit("deploy_update", locals(), result, ticket_id)
    return result

if __name__ == "__main__":
    mcp.run()

Setup, from scratch:

1. Deploy Satellite Server and register it against your Red Hat subscription manifest before attaching any client host.
2. Register each RHEL host as a Satellite content host (subscription-manager register --org ... --activationkey ...) so it's managed by lifecycle environment/content view rather than a bare, unmanaged install.
3. Create an API-only user with a custom role scoped to "View hosts" and "Manage content views/lifecycle environments" separately from a full Satellite Administrator.
4. Build lifecycle environments (Library → Dev → Test → Prod) and content views so a patch promotion is "promote this content view," not a per-host yum update — this is what "staged in a canary group" means concretely for RHEL in the architecture page's self-healing table.
5. For single-host emergency access outside the fleet-wide flow, note that Red Hat's Cockpit web console runs per-host and doesn't go through Satellite at all — a break-glass path for a human, not something this agent automates against.

API auth & a call

import httpx

SAT_BASE = "https://satellite.internal.example.com/api/v2"

def list_content_hosts(username: str, password: str, org_id: int) -> dict:
    r = httpx.get(f"{SAT_BASE}/hosts", params={"organization_id": org_id}, auth=(username, password))
    r.raise_for_status()
    return r.json()

rh_satellite_mcp_server.py

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("linux-satellite", system_class="rhsatellite")
SAT_BASE = "https://satellite.internal.example.com/api/v2"

@mcp.tool()
def host_errata(host_id: int) -> dict:
    """Tier 0 -- outstanding errata (security/bugfix/enhancement) for one host."""
    cred = get_scoped_credential(system="rhsatellite", ttl_seconds=300)
    r = httpx.get(f"{SAT_BASE}/hosts/{host_id}/errata", auth=(cred.username, cred.password))
    r.raise_for_status()
    return r.json()

@mcp.tool()
def promote_content_view(content_view_id: int, environment_id: int, ticket_id: str) -> dict:
    """Tier 2 -- promotes a content view to the next lifecycle environment,
    the fleet-wide patch-staging step the Linux Server agent's mandatory
    dry-run wraps around before this runs against anything but Library."""
    cred = get_scoped_credential(system="rhsatellite", ttl_seconds=300)
    r = httpx.post(f"{SAT_BASE}/content_view_versions/{content_view_id}/promote",
                    json={"environment_id": environment_id},
                    auth=(cred.username, cred.password))
    r.raise_for_status()
    result = r.json()
    audit("promote_content_view", locals(), result, ticket_id)
    return result

if __name__ == "__main__":
    mcp.run()

The fourteen systems above are the supporting cast; the actual write targets are the nine domain agents' own systems. Each follows the identical MCP pattern from Section 2 — one server, credential checkout per call, docstring-declared tier — so rather than repeat the full pattern nine more times, here's the API surface and the one that's most commonly asked about (Identity/AD) worked in the same depth as ISE above.

identity_ad_mcp_server.py — the Identity agent, worked in full since it's the most common "how do I actually wire AD" question

from shared.mcp_base import build_agent_server, audit
from shared.vault_client import get_scoped_credential
import httpx

mcp = build_agent_server("identity-ad", system_class="entra-ad")
GRAPH = "https://graph.microsoft.com/v1.0"

def _token() -> str:
    cred = get_scoped_credential(system="entra-ad", ttl_seconds=300)
    r = httpx.post(f"https://login.microsoftonline.com/{cred.tenant_id}/oauth2/v2.0/token",
        data={"grant_type": "client_credentials", "client_id": cred.client_id,
              "client_secret": cred.client_secret, "scope": "https://graph.microsoft.com/.default"})
    r.raise_for_status()
    return r.json()["access_token"]

@mcp.tool()
def get_account_status(user_principal_name: str) -> dict:
    """Tier 0 -- read account state (locked, disabled, last sign-in)."""
    headers = {"Authorization": f"Bearer {_token()}"}
    r = httpx.get(f"{GRAPH}/users/{user_principal_name}"
                   "?$select=accountEnabled,userPrincipalName,signInActivity", headers=headers)
    r.raise_for_status()
    return r.json()

@mcp.tool()
def reset_password(user_principal_name: str, ticket_id: str) -> dict:
    """Tier 1 -- forces a temporary password and sign-in-required flag. High
    volume, low blast radius, clean rollback (re-reset), which is why the
    architecture page's deployment guide picks this as the first domain
    agent to build end to end."""
    headers = {"Authorization": f"Bearer {_token()}", "Content-Type": "application/json"}
    body = {"passwordProfile": {"forceChangePasswordNextSignIn": True,
                                  "password": get_scoped_credential(system="temp-pw-gen").token}}
    r = httpx.patch(f"{GRAPH}/users/{user_principal_name}", json=body, headers=headers)
    r.raise_for_status()
    result = {"status": r.status_code}
    audit("reset_password", locals(), result, ticket_id)
    return result

if __name__ == "__main__":
    mcp.run()

This page is the "how" for the "what" in the architecture page's phased deployment guide — use that page's phase table for sequencing (audit log and monitoring first, one Tier 1 agent before the rest, approval mechanism before any Tier 2 agent goes live) and this page for the actual setup/API/MCP steps each phase's numbered build step refers to. Concretely: Phase 0 build steps 1–3 map to Sections 1–3, 5, and 6 above; Phase 1 step 7 maps to Section 18's Identity agent; Phase 2 steps 12 and 14 map to Sections 7 and 9. Sections 11–17 (Firepower Management Center, Catalyst Center, Aria, SCOM, Azure Monitor, MECM, and Satellite) don't map to a numbered phase step on their own — each slots into the phase of whichever domain agent or supporting-monitoring role it extends above, onboarded with the same burn-in discipline as everything else in that phase, not a shortcut for arriving later. Nothing here changes the tier assignments, the approval gate, or the deny-list from the architecture page — this is implementation detail underneath decisions that page already made.

• Download this guide as a PDF — every section and code block above, formatted to print or attach to an email.
• Back to the architecture blueprint — the diagrams, risk tiers, and phased rollout this guide's code implements.