Integration guide — wiring every supporting system with code

The architecture page's "Supporting systems" table names twenty-two 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 twenty-two 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; Section 19 (CyberArk) is where it stops being a stand-in and becomes a real implementation — a HashiCorp Vault dynamic secrets engine or a cloud Secrets Manager slots into the same function signature. 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 26'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()

Setup, from scratch:

1. Configure each UPS/rack PDU's Network Management Card (NMC3) with its own management-network address; for more than a handful of units, point Platform Ops at a central EcoStruxure IT / StruxureWare Data Center Expert instance instead of polling every card individually.
2. Create a dedicated local API account with the read-only "Device User" role for telemetry, and a separate account scoped to outlet-switching only for the narrow write path — not the card's own admin account reused everywhere.
3. Enable the NMC's REST API and set the session-token TTL short, matching the vault's own scoped-credential lifetime rather than a long-lived static token.
4. Route power-event traps (on-battery, low-runtime, overload) through the same alert —> ServiceNow webhook pattern as Zabbix/SCOM above, so a battery event opens an Incident the same way a disk or CPU threshold does — not a fourth, bespoke alert channel.
5. Document outlet-to-device mapping in NetBox (power feed as a cabled connection, same as a network port) before wiring up remote outlet control — switching the wrong outlet on a shared PDU is the one mistake this integration can't recover from with a rollback script.

API auth & a call (NMC3 REST, session token)

import httpx

NMC_BASE = "https://apc-ups-01.internal.example.com/rest"

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

def ups_status(token: str) -> dict:
    r = httpx.get(f"{NMC_BASE}/ups/status", headers={"Authorization": f"Bearer {token}"})
    r.raise_for_status()
    return r.json()  # battery charge %, runtime remaining, load %, on-battery flag

apc_power_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("platform-ops-power", system_class="apc")
NMC_BASE = "https://apc-ups-01.internal.example.com/rest"

def _token() -> str:
    cred = get_scoped_credential(system="apc", ttl_seconds=300)
    r = httpx.post(f"{NMC_BASE}/login", json={"username": cred.username, "password": cred.password})
    r.raise_for_status()
    return r.json()["token"]

@mcp.tool()
def ups_status() -> dict:
    """Tier 0 -- battery charge, runtime remaining, load, and on-battery flag
    for one UPS; the heartbeat Platform Ops watches alongside Zabbix/SCOM."""
    r = httpx.get(f"{NMC_BASE}/ups/status", headers={"Authorization": f"Bearer {_token()}"})
    r.raise_for_status()
    return r.json()

@mcp.tool()
def initiate_graceful_shutdown(ups_id: str, ticket_id: str) -> dict:
    """Tier 1 -- on-battery with runtime under threshold: signals the shutdown
    sequence the VMware/Windows/Linux Server agents carry out in dependency
    order (see the architecture page's self-healing table), ahead of a hard
    outage rather than reacting to one."""
    r = httpx.post(f"{NMC_BASE}/ups/{ups_id}/shutdown-signal",
                    headers={"Authorization": f"Bearer {_token()}"})
    r.raise_for_status()
    result = r.json()
    audit("initiate_graceful_shutdown", locals(), result, ticket_id)
    return result

@mcp.tool()
def switch_outlet(pdu_id: str, outlet_id: str, state: str, ticket_id: str) -> dict:
    """Tier 2 -- remote power-cycle of one PDU outlet (e.g. a hung device that
    doesn't respond to its own agent's restart); requires the NetBox
    outlet-to-device mapping confirmed in setup step 5, since the wrong
    outlet on a shared PDU has no rollback."""
    cred = get_scoped_credential(system="apc", ttl_seconds=300)
    r = httpx.post(f"{NMC_BASE}/pdu/{pdu_id}/outlets/{outlet_id}",
                    json={"state": state}, headers={"Authorization": f"Bearer {_token()}"})
    r.raise_for_status()
    result = r.json()
    audit("switch_outlet", locals(), result, ticket_id)
    return result

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

Every other section on this page assumes a call to get_scoped_credential(...). This is the section that makes that real: CyberArk is the vault those leases come from, so it is set up first, not last.

Setup, from scratch:

1. Stand up the Vault, PVWA, CPM, and PSM components (or use CyberArk Privilege Cloud); onboard every target-system privileged account into a Safe named per domain agent (Safe-Network, Safe-Identity, …) so a Safe's membership is that agent's blast radius.
2. Enable the Central Credential Provider (CCP) or the Conjur-based Secrets Provider for application-to-vault retrieval — the orchestrator and agents authenticate to it with a short-lived platform identity, never a stored vault password.
3. Set each Safe's account to one-time-use / auto-rotate-on-retrieval with a check-in window equal to the longest expected task, so a lease that is not returned is force-rotated rather than left live.
4. Require PSM session isolation and recording for any interactive human break-glass checkout, and route PSM session-start / session-end and every credential retrieval event to Splunk (Section 21) and the audit log.
5. Add a CCP health probe to Platform Ops: if the vault is unreachable, agents must fail closed (refuse the action) rather than fall back to any cached or static credential.

API auth & a call (Central Credential Provider, client-cert auth)

import httpx

CCP_BASE = "https://ccp.internal.example.com/AIMWebService/api/Accounts"

def fetch_credential(app_id: str, safe: str, object_name: str) -> dict:
    r = httpx.get(
        CCP_BASE,
        params={"AppID": app_id, "Safe": safe, "Object": object_name},
        cert=("/etc/agent/ccp-client.pem", "/etc/agent/ccp-client.key"),
        verify="/etc/agent/ccp-ca.pem",
    )
    r.raise_for_status()
    return r.json()  # {"UserName": ..., "Content": ..., "PasswordChangeInProcess": ...}

shared/vault_client.py (the function every other section calls)

from dataclasses import dataclass
import httpx

CCP_BASE = "https://ccp.internal.example.com/AIMWebService/api/Accounts"
_SAFE_FOR = {"network": "Safe-Network", "identity": "Safe-Identity",
             "apc": "Safe-Facilities", "netapp": "Safe-Storage"}  # ... one per system class

@dataclass
class ScopedCredential:
    username: str
    password: str
    token: str = ""

def get_scoped_credential(system: str, ttl_seconds: int = 300) -> ScopedCredential:
    """Check out a one-time-use privileged credential for `system` from its
    CyberArk Safe. The account auto-rotates on retrieval, so `ttl_seconds`
    is the caller's promise to finish (and the check-in deadline), not a
    grant that outlives the task. Raises -- fails closed -- if the vault
    is unreachable; there is no static fallback."""
    r = httpx.get(CCP_BASE, params={"AppID": f"agent-{system}",
                                    "Safe": _SAFE_FOR[system],
                                    "Object": f"{system}-privileged"},
                  cert=("/etc/agent/ccp-client.pem", "/etc/agent/ccp-client.key"),
                  verify="/etc/agent/ccp-ca.pem", timeout=ttl_seconds)
    r.raise_for_status()
    body = r.json()
    return ScopedCredential(username=body["UserName"], password=body["Content"])

Setup, from scratch:

1. In the Falcon console, create an OAuth2 API client with only the scopes this design uses: Hosts:read, Detections:read, Alerts:read, and — only if host containment is in scope — Real Time Response (Admin) or Hosts:write for network-contain.
2. Send Falcon detections to ServiceNow through the streaming API or the CrowdStrike Store ServiceNow app, so a detection opens an Incident on the same intake path as a Zabbix trigger — not a separate SOC-only queue.
3. Tag the sensor grouping to match domain-agent ownership (servers vs. desktops vs. Linux) so a "sensor not reporting" alert routes to the right agent.
4. Mark contain_host as Tier 3 in the MCP server and require an explicit ServiceNow approval record; isolation cuts a host off the network and is disruptive even when correct.
5. Add sensor-count and sensor-health totals to Platform Ops so a silent drop in coverage is itself an alert.

API auth & a call (OAuth2 client credentials)

import httpx

FALCON_BASE = "https://api.crowdstrike.com"

def token(client_id: str, client_secret: str) -> str:
    r = httpx.post(f"{FALCON_BASE}/oauth2/token",
                   data={"client_id": client_id, "client_secret": client_secret})
    r.raise_for_status()
    return r.json()["access_token"]

def stale_sensors(bearer: str, days: int = 2) -> list[str]:
    r = httpx.get(f"{FALCON_BASE}/devices/queries/devices/v1",
                  params={"filter": f"last_seen:<'now-{days}d'"},
                  headers={"Authorization": f"Bearer {bearer}"})
    r.raise_for_status()
    return r.json()["resources"]

crowdstrike_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("security-edr", system_class="crowdstrike")
FALCON_BASE = "https://api.crowdstrike.com"

def _bearer() -> str:
    cred = get_scoped_credential(system="crowdstrike", ttl_seconds=300)
    r = httpx.post(f"{FALCON_BASE}/oauth2/token",
                   data={"client_id": cred.username, "client_secret": cred.password})
    r.raise_for_status()
    return r.json()["access_token"]

@mcp.tool()
def sensor_health(hostname: str) -> dict:
    """Tier 0 -- last-seen, sensor version, and reduced-functionality-mode
    flag for one host; the coverage signal Platform Ops watches."""
    r = httpx.get(f"{FALCON_BASE}/devices/queries/devices/v1",
                  params={"filter": f"hostname:'{hostname}'"},
                  headers={"Authorization": f"Bearer {_bearer()}"})
    r.raise_for_status()
    return r.json()

@mcp.tool()
def contain_host(device_id: str, ticket_id: str) -> dict:
    """Tier 3 -- network-isolate a host (keeps only the Falcon control
    channel). Disruptive; requires an approved ServiceNow record. Undo is
    lift_containment, so this is reversible but never automatic."""
    r = httpx.post(f"{FALCON_BASE}/devices/entities/devices-actions/v2",
                   params={"action_name": "contain"},
                   json={"ids": [device_id]},
                   headers={"Authorization": f"Bearer {_bearer()}"})
    r.raise_for_status()
    result = r.json()
    audit("contain_host", locals(), result, ticket_id)
    return result

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

Setup, from scratch:

1. Create a dedicated index for the framework's own audit log (index=agent_audit) with a retention and legal-hold policy at least as long as the change-record retention in the operations SOP, and make it write-once (no edit/delete role granted to any agent identity).
2. Forward the audit log, every CyberArk retrieval event, Falcon detections, and the domain agents' structured action logs into Splunk — this is where "immutable audit log" gets a query surface a human reviewer can actually use.
3. Create a service account with a role limited to search over the framework indexes only, and issue it a token (not a password) for the MCP server.
4. Build saved searches / correlation searches for the review cadences in the SOP (monthly Tier 3 approval audit, actions with no matching approval record, out-of-window changes) and wire alert actions to open a ServiceNow Incident.
5. Point Grafana (Section 6) at Splunk as a second data source so the health dashboards and the audit views sit side by side.

API auth & a call (REST, bearer token, one-shot search)

import httpx

SPLUNK_BASE = "https://splunk.internal.example.com:8089"

def search(token: str, spl: str) -> list[dict]:
    r = httpx.post(f"{SPLUNK_BASE}/services/search/jobs/export",
                   data={"search": f"search {spl}", "output_mode": "json", "earliest_time": "-24h"},
                   headers={"Authorization": f"Bearer {token}"})
    r.raise_for_status()
    return [line for line in r.text.splitlines() if line]

splunk_mcp_server.py

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

mcp = build_agent_server("platform-ops-siem", system_class="splunk")
SPLUNK_BASE = "https://splunk.internal.example.com:8089"

def _token() -> str:
    return get_scoped_credential(system="splunk", ttl_seconds=300).token

@mcp.tool()
def actions_without_approval(hours: int = 24) -> list[dict]:
    """Tier 0 -- correlation search: executed Tier >=1 actions in the audit
    index with no matching approval event in the window. Feeds the SOP's
    monthly approval audit; read-only, opens an Incident on any hit."""
    spl = (f'index=agent_audit action=execute tier>=1 earliest=-{hours}h '
           f'| join ticket_id [search index=agent_audit event=approval_granted] '
           f'| where isnull(approval_ts)')
    r = httpx.post(f"{SPLUNK_BASE}/services/search/jobs/export",
                   data={"search": f"search {spl}", "output_mode": "json"},
                   headers={"Authorization": f"Bearer {_token()}"})
    r.raise_for_status()
    return [json.loads(x) for x in r.text.splitlines() if x]

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

Setup, from scratch:

1. Enable the ONTAP REST API on each cluster management LIF and create an SVM-scoped role: readonly for capacity/snapshot telemetry, plus a separate role limited to snapshot create and volume modify (size) if provisioning is in scope for the Storage-consuming agents.
2. Onboard both credential roles into the CyberArk Safe-Storage so the VMware / Database / Windows / Linux agents check them out per call rather than holding a standing storage login.
3. Register volume-to-consumer mapping in NetBox / the CMDB (which datastore, which DB, which host) so a capacity alert routes to the owning domain agent, not a generic storage queue.
4. Mark volume delete, snapshot delete, and any SnapMirror break as deny-list / Tier 3 — destructive storage actions never auto-execute, matching the backup deny-list rule.
5. Send EMS events and capacity thresholds to ServiceNow via the same alert → Incident webhook as the monitoring stack.

API auth & a call (ONTAP REST, basic auth over TLS)

import httpx

ONTAP_BASE = "https://netapp-cluster-01.internal.example.com/api"

def volume_space(user: str, password: str, volume_uuid: str) -> dict:
    r = httpx.get(f"{ONTAP_BASE}/storage/volumes/{volume_uuid}",
                  params={"fields": "space,svm,state"},
                  auth=(user, password))
    r.raise_for_status()
    return r.json()

netapp_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("storage-netapp", system_class="netapp")
ONTAP_BASE = "https://netapp-cluster-01.internal.example.com/api"

def _auth():
    c = get_scoped_credential(system="netapp", ttl_seconds=300)
    return (c.username, c.password)

@mcp.tool()
def volume_space(volume_uuid: str) -> dict:
    """Tier 0 -- used/available/snapshot-reserve for one volume; the
    capacity signal Platform Ops watches and the owning agent reads
    before it provisions."""
    r = httpx.get(f"{ONTAP_BASE}/storage/volumes/{volume_uuid}",
                  params={"fields": "space,svm,state"}, auth=_auth())
    r.raise_for_status()
    return r.json()

@mcp.tool()
def create_snapshot(volume_uuid: str, name: str, ticket_id: str) -> dict:
    """Tier 2 -- point-in-time snapshot before a risky change on the
    consuming system (DB migration, in-place upgrade). Reversible;
    snapshot *deletion* is deny-list, not offered here."""
    r = httpx.post(f"{ONTAP_BASE}/storage/volumes/{volume_uuid}/snapshots",
                   json={"name": name}, auth=_auth())
    r.raise_for_status()
    result = r.json()
    audit("create_snapshot", locals(), result, ticket_id)
    return result

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

Same role as NetApp ONTAP, different vendor API. Environments that run both keep one MCP server per array family; the design does not assume a single storage vendor.

Setup, from scratch:

1. Enable the array's REST API (PowerStore REST, Unisphere for PowerMax REST, or Unity REST) and create a monitor-role account for telemetry plus a storage-admin account scoped to snapshot/volume-resize only, both onboarded into CyberArk Safe-Storage.
2. For PowerMax, front the array with an on-network Unisphere / REST endpoint; for PowerStore, the cluster management IP serves the API directly.
3. Map volumes/LUNs to consumers in the CMDB exactly as for ONTAP so alerts route by owner.
4. Deny-list volume/LUN delete, snapshot delete, and replication-session delete; Tier 3 for create/resize.
5. Forward alerts to ServiceNow via the common webhook.

API auth & a call (PowerStore REST, token from basic auth)

import httpx

PS_BASE = "https://powerstore-01.internal.example.com/api/rest"

def login(user: str, password: str) -> str:
    r = httpx.get(f"{PS_BASE}/login_session", auth=(user, password))
    r.raise_for_status()
    return r.headers["DELL-EMC-TOKEN"]

def volume_metrics(token: str, volume_id: str) -> dict:
    r = httpx.get(f"{PS_BASE}/volume/{volume_id}",
                  params={"select": "name,size,state,logical_used"},
                  headers={"DELL-EMC-TOKEN": token})
    r.raise_for_status()
    return r.json()

dell_storage_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("storage-dell", system_class="dell-storage")
PS_BASE = "https://powerstore-01.internal.example.com/api/rest"

def _token() -> str:
    c = get_scoped_credential(system="dell-storage", ttl_seconds=300)
    r = httpx.get(f"{PS_BASE}/login_session", auth=(c.username, c.password))
    r.raise_for_status()
    return r.headers["DELL-EMC-TOKEN"]

@mcp.tool()
def volume_metrics(volume_id: str) -> dict:
    """Tier 0 -- size, logical-used, and state for one volume; the capacity
    signal Platform Ops watches, vendor-parallel to netapp.volume_space."""
    r = httpx.get(f"{PS_BASE}/volume/{volume_id}",
                  params={"select": "name,size,state,logical_used"},
                  headers={"DELL-EMC-TOKEN": _token()})
    r.raise_for_status()
    return r.json()

@mcp.tool()
def create_snapshot(volume_id: str, name: str, ticket_id: str) -> dict:
    """Tier 2 -- snapshot before a risky change on the consuming system.
    Deletion is deny-list, not offered here."""
    r = httpx.post(f"{PS_BASE}/volume/{volume_id}/snapshot",
                   json={"name": name}, headers={"DELL-EMC-TOKEN": _token()})
    r.raise_for_status()
    result = r.json()
    audit("create_snapshot", locals(), result, ticket_id)
    return result

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

An extension of the Network agent (Section 26), not a new agent. Dell OS10 leaf/ToR switches sit alongside Cisco IOS/NX-OS in the same rack rows.

Setup, from scratch:

1. Enable the OS10 RESTCONF API (or manage via Ansible's dellemc.os10 collection) on each switch's management interface; create a role-scoped account onboarded into CyberArk Safe-Network.
2. Add the OS10 devices to NetBox with the same platform/role fields as the Cisco gear so inventory and the pre-change Batfish model cover them uniformly.
3. Import OS10 configs into Batfish (Section 7) so a proposed ToR change gets the same offline verification as an NX-OS change — this is what "mandatory dry-run" means for the ToR layer.
4. If SmartFabric Services (fabric mode) is in use, target the fabric's REST endpoint for fabric-wide changes rather than switch-by-switch, same principle as Nexus Dashboard for ACI.
5. Keep port toggles at Tier 2 and any VLAN/routing change at Tier 3, identical to the base Network agent's tiers.

API auth & a call (OS10 RESTCONF, basic auth over TLS)

import httpx

OS10_BASE = "https://tor-leaf-01.internal.example.com/restconf/data"

def interface_status(user: str, password: str, ifname: str) -> dict:
    r = httpx.get(f"{OS10_BASE}/dell-interface:interfaces/interface={ifname}",
                  auth=(user, password), headers={"Accept": "application/yang-data+json"})
    r.raise_for_status()
    return r.json()

dell_powerswitch_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("network-tor-os10", system_class="dell-powerswitch")
OS10_BASE = "https://tor-leaf-01.internal.example.com/restconf/data"

def _auth():
    c = get_scoped_credential(system="dell-powerswitch", ttl_seconds=300)
    return (c.username, c.password)

@mcp.tool()
def interface_status(ifname: str) -> dict:
    """Tier 0 -- admin/oper state, counters, and description for one ToR
    port; read the same way the Network agent reads a Cisco interface."""
    r = httpx.get(f"{OS10_BASE}/dell-interface:interfaces/interface={ifname}",
                  auth=_auth(), headers={"Accept": "application/yang-data+json"})
    r.raise_for_status()
    return r.json()

@mcp.tool()
def set_interface_admin_state(ifname: str, enabled: bool, ticket_id: str) -> dict:
    """Tier 2 -- shut/no-shut one access port (e.g. isolating a noisy host).
    Batfish must have modeled the change first; rollback is the inverse call."""
    r = httpx.patch(f"{OS10_BASE}/dell-interface:interfaces/interface={ifname}",
                    json={"interface": {"enabled": enabled}}, auth=_auth(),
                    headers={"Content-Type": "application/yang-data+json"})
    r.raise_for_status()
    result = {"status": r.status_code}
    audit("set_interface_admin_state", locals(), result, ticket_id)
    return result

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

The one integration where the deny-list matters most: no agent, at any tier, ever deletes a backup or shortens its retention. This server is read-heavy by design — job status and restore verification — with restore itself gated at Tier 2.

Setup, from scratch:

1. Create a restricted API account (Veeam RESTful API, Commvault REST, Rubrik/Cohesity REST) with permission to read job sessions and start restores, and no permission to modify jobs, retention, or backup repositories. Onboard it into CyberArk Safe-Backup.
2. Enable immutability / air-gap on the repository at the storage layer (hardened Linux repo, object-lock bucket, or appliance immutability) so retention cannot be shortened even by a compromised backup-console admin, let alone an agent.
3. Forward job success/failure and SLA-compliance events to ServiceNow so a failed backup opens an Incident automatically.
4. Register a scheduled restore-test (mount / instant-recovery to an isolated network, checksum, tear down) and treat a failed verification as a Tier 1 self-healing condition: re-queue once, then page.
5. Map every protected workload to its owning domain agent so a restore request is planned and approved by that agent, not a generic backup queue.

API auth & a call (Veeam RESTful API, OAuth2 password grant)

import httpx

VBR_BASE = "https://veeam-01.internal.example.com:9419/api/v1"

def token(user: str, password: str) -> str:
    r = httpx.post(f"{VBR_BASE}/token",
                   data={"grant_type": "password", "username": user, "password": password},
                   headers={"x-api-version": "1.1-rev1"})
    r.raise_for_status()
    return r.json()["access_token"]

def last_session(bearer: str, job_id: str) -> dict:
    r = httpx.get(f"{VBR_BASE}/sessions",
                  params={"jobIdFilter": job_id, "limit": 1, "orderColumn": "CreationTime", "orderAsc": "false"},
                  headers={"Authorization": f"Bearer {bearer}", "x-api-version": "1.1-rev1"})
    r.raise_for_status()
    return r.json()

backup_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("platform-ops-backup", system_class="backup")
VBR_BASE = "https://veeam-01.internal.example.com:9419/api/v1"
HDRS = {"x-api-version": "1.1-rev1"}

def _bearer() -> str:
    c = get_scoped_credential(system="backup", ttl_seconds=300)
    r = httpx.post(f"{VBR_BASE}/token",
                   data={"grant_type": "password", "username": c.username, "password": c.password},
                   headers=HDRS)
    r.raise_for_status()
    return r.json()["access_token"]

@mcp.tool()
def job_status(job_id: str) -> dict:
    """Tier 0 -- result and timing of the most recent session for one
    backup job; the signal Platform Ops watches for a failed/missed backup."""
    r = httpx.get(f"{VBR_BASE}/sessions",
                  params={"jobIdFilter": job_id, "limit": 1,
                          "orderColumn": "CreationTime", "orderAsc": "false"},
                  headers={"Authorization": f"Bearer {_bearer()}", **HDRS})
    r.raise_for_status()
    return r.json()

@mcp.tool()
def start_restore(restore_point_id: str, target: str, ticket_id: str) -> dict:
    """Tier 2 -- start a restore from an existing recovery point to `target`.
    Never deletes or overwrites a backup. Deleting/re-timing retention is
    deny-list and has no tool here at all."""
    r = httpx.post(f"{VBR_BASE}/restore",
                   json={"restorePointId": restore_point_id, "target": target},
                   headers={"Authorization": f"Bearer {_bearer()}", **HDRS})
    r.raise_for_status()
    result = r.json()
    audit("start_restore", locals(), result, ticket_id)
    return result

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

The twenty-two 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 26's Identity agent; Phase 2 steps 12 and 14 map to Sections 7 and 9. Section 19 (CyberArk) is a Phase 0 prerequisite — the vault every later section's get_scoped_credential() call depends on. Sections 11–18 and 20–25 (Firepower Management Center, Catalyst Center, Aria, SCOM, Azure Monitor, MECM, Satellite, APC power management, CrowdStrike, Splunk, NetApp, Dell storage, Dell PowerSwitch, and backup/recovery) 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.

• Paid full edition — the portable PDF (39 pages, every section and code block above, formatted to print or attach to an email).
• Service desk readiness review — have your own integration and credential-scoping design assessed directly. $349, fixed price, written report.
• Back to the architecture blueprint — the diagrams, risk tiers, and phased rollout this guide's code implements.
• Operations model & SOP — once this is built, how it actually gets run: on-call, incident response, change windows, and review cadence.
• Agent-to-agent coordination protocol — the message envelope and delivery guarantees these integrations exchange: supporting-system events arrive as protocol messages on the same bus.
• SOC & incident-response architecture — the security-operations counterpart, reusing the EDR, SIEM, secrets vault, and backup integrations documented here for detection and response.