server.py•16.1 kB
from mcp.server.fastmcp import FastMCP
from netbox_client import NetBoxRestClient
import os
# Mapping of simple object names to API endpoints
NETBOX_OBJECT_TYPES = {
# DCIM (Device and Infrastructure)
"cables": "dcim/cables",
"console-ports": "dcim/console-ports",
"console-server-ports": "dcim/console-server-ports",
"devices": "dcim/devices",
"device-bays": "dcim/device-bays",
"device-roles": "dcim/device-roles",
"device-types": "dcim/device-types",
"front-ports": "dcim/front-ports",
"interfaces": "dcim/interfaces",
"inventory-items": "dcim/inventory-items",
"locations": "dcim/locations",
"manufacturers": "dcim/manufacturers",
"modules": "dcim/modules",
"module-bays": "dcim/module-bays",
"module-types": "dcim/module-types",
"platforms": "dcim/platforms",
"power-feeds": "dcim/power-feeds",
"power-outlets": "dcim/power-outlets",
"power-panels": "dcim/power-panels",
"power-ports": "dcim/power-ports",
"racks": "dcim/racks",
"rack-reservations": "dcim/rack-reservations",
"rack-roles": "dcim/rack-roles",
"regions": "dcim/regions",
"sites": "dcim/sites",
"site-groups": "dcim/site-groups",
"virtual-chassis": "dcim/virtual-chassis",
# IPAM (IP Address Management)
"asns": "ipam/asns",
"asn-ranges": "ipam/asn-ranges",
"aggregates": "ipam/aggregates",
"fhrp-groups": "ipam/fhrp-groups",
"ip-addresses": "ipam/ip-addresses",
"ip-ranges": "ipam/ip-ranges",
"prefixes": "ipam/prefixes",
"rirs": "ipam/rirs",
"roles": "ipam/roles",
"route-targets": "ipam/route-targets",
"services": "ipam/services",
"vlans": "ipam/vlans",
"vlan-groups": "ipam/vlan-groups",
"vrfs": "ipam/vrfs",
# Circuits
"circuits": "circuits/circuits",
"circuit-types": "circuits/circuit-types",
"circuit-terminations": "circuits/circuit-terminations",
"providers": "circuits/providers",
"provider-networks": "circuits/provider-networks",
# Virtualization
"clusters": "virtualization/clusters",
"cluster-groups": "virtualization/cluster-groups",
"cluster-types": "virtualization/cluster-types",
"virtual-machines": "virtualization/virtual-machines",
"vm-interfaces": "virtualization/interfaces",
# Tenancy
"tenants": "tenancy/tenants",
"tenant-groups": "tenancy/tenant-groups",
"contacts": "tenancy/contacts",
"contact-groups": "tenancy/contact-groups",
"contact-roles": "tenancy/contact-roles",
# VPN
"ike-policies": "vpn/ike-policies",
"ike-proposals": "vpn/ike-proposals",
"ipsec-policies": "vpn/ipsec-policies",
"ipsec-profiles": "vpn/ipsec-profiles",
"ipsec-proposals": "vpn/ipsec-proposals",
"l2vpns": "vpn/l2vpns",
"tunnels": "vpn/tunnels",
"tunnel-groups": "vpn/tunnel-groups",
# Wireless
"wireless-lans": "wireless/wireless-lans",
"wireless-lan-groups": "wireless/wireless-lan-groups",
"wireless-links": "wireless/wireless-links",
# Extras
"config-contexts": "extras/config-contexts",
"custom-fields": "extras/custom-fields",
"export-templates": "extras/export-templates",
"image-attachments": "extras/image-attachments",
"jobs": "extras/jobs",
"saved-filters": "extras/saved-filters",
"scripts": "extras/scripts",
"tags": "extras/tags",
"webhooks": "extras/webhooks",
}
mcp = FastMCP("NetBox", log_level="DEBUG")
netbox = None
@mcp.tool()
def netbox_get_objects(object_type: str, filters: dict):
"""
Get objects from NetBox based on their type and filters
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
filters: dict of filters to apply to the API call based on the NetBox API filtering options
Valid object_type values:
DCIM (Device and Infrastructure):
- cables
- console-ports
- console-server-ports
- devices
- device-bays
- device-roles
- device-types
- front-ports
- interfaces
- inventory-items
- locations
- manufacturers
- modules
- module-bays
- module-types
- platforms
- power-feeds
- power-outlets
- power-panels
- power-ports
- racks
- rack-reservations
- rack-roles
- regions
- sites
- site-groups
- virtual-chassis
IPAM (IP Address Management):
- asns
- asn-ranges
- aggregates
- fhrp-groups
- ip-addresses
- ip-ranges
- prefixes
- rirs
- roles
- route-targets
- services
- vlans
- vlan-groups
- vrfs
Circuits:
- circuits
- circuit-types
- circuit-terminations
- providers
- provider-networks
Virtualization:
- clusters
- cluster-groups
- cluster-types
- virtual-machines
- vm-interfaces
Tenancy:
- tenants
- tenant-groups
- contacts
- contact-groups
- contact-roles
VPN:
- ike-policies
- ike-proposals
- ipsec-policies
- ipsec-profiles
- ipsec-proposals
- l2vpns
- tunnels
- tunnel-groups
Wireless:
- wireless-lans
- wireless-lan-groups
- wireless-links
See NetBox API documentation for filtering options for each object type.
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call
return netbox.get(endpoint, params=filters)
@mcp.tool()
def netbox_get_object_by_id(object_type: str, object_id: int):
"""
Get detailed information about a specific NetBox object by its ID.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
object_id: The numeric ID of the object
Returns:
Complete object details
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = f"{NETBOX_OBJECT_TYPES[object_type]}/{object_id}"
return netbox.get(endpoint)
@mcp.tool()
def netbox_get_changelogs(filters: dict):
"""
Get object change records (changelogs) from NetBox based on filters.
Args:
filters: dict of filters to apply to the API call based on the NetBox API filtering options
Returns:
List of changelog objects matching the specified filters
Filtering options include:
- user_id: Filter by user ID who made the change
- user: Filter by username who made the change
- changed_object_type_id: Filter by ContentType ID of the changed object
- changed_object_id: Filter by ID of the changed object
- object_repr: Filter by object representation (usually contains object name)
- action: Filter by action type (created, updated, deleted)
- time_before: Filter for changes made before a given time (ISO 8601 format)
- time_after: Filter for changes made after a given time (ISO 8601 format)
- q: Search term to filter by object representation
Example:
To find all changes made to a specific device with ID 123:
{"changed_object_type_id": "dcim.device", "changed_object_id": 123}
To find all deletions in the last 24 hours:
{"action": "delete", "time_after": "2023-01-01T00:00:00Z"}
Each changelog entry contains:
- id: The unique identifier of the changelog entry
- user: The user who made the change
- user_name: The username of the user who made the change
- request_id: The unique identifier of the request that made the change
- action: The type of action performed (created, updated, deleted)
- changed_object_type: The type of object that was changed
- changed_object_id: The ID of the object that was changed
- object_repr: String representation of the changed object
- object_data: The object's data after the change (null for deletions)
- object_data_v2: Enhanced data representation
- prechange_data: The object's data before the change (null for creations)
- postchange_data: The object's data after the change (null for deletions)
- time: The timestamp when the change was made
"""
endpoint = "core/object-changes"
# Make API call
return netbox.get(endpoint, params=filters)
@mcp.tool()
def netbox_create_object(object_type: str, data: dict):
"""
Create a new object in NetBox.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
data: Dict containing the object data to create
Returns:
The created object as a dict
Example:
To create a new site:
netbox_create_object("sites", {
"name": "New Site",
"slug": "new-site",
"status": "active"
})
To create a new device:
netbox_create_object("devices", {
"name": "new-device",
"device_type": 1, # ID of device type
"site": 1, # ID of site
"role": 1, # ID of device role
"status": "active"
})
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call
return netbox.create(endpoint, data)
@mcp.tool()
def netbox_update_object(object_type: str, object_id: int, data: dict):
"""
Update an existing object in NetBox.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
object_id: The numeric ID of the object to update
data: Dict containing the object data to update (only changed fields needed)
Returns:
The updated object as a dict
Example:
To update a site's description:
netbox_update_object("sites", 1, {"description": "Updated description"})
To change a device's status:
netbox_update_object("devices", 5, {"status": "offline"})
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call
return netbox.update(endpoint, object_id, data)
@mcp.tool()
def netbox_delete_object(object_type: str, object_id: int):
"""
Delete an object from NetBox.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
object_id: The numeric ID of the object to delete
Returns:
True if deletion was successful
WARNING: This permanently deletes the object and cannot be undone!
Example:
To delete a device:
netbox_delete_object("devices", 5)
To delete an IP address:
netbox_delete_object("ip-addresses", 123)
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call - this will raise an exception if it fails
success = netbox.delete(endpoint, object_id)
if success:
return {"success": True, "message": f"Successfully deleted {object_type} with ID {object_id}"}
else:
return {"success": False, "message": f"Failed to delete {object_type} with ID {object_id}"}
@mcp.tool()
def netbox_bulk_create_objects(object_type: str, data: list):
"""
Create multiple objects in NetBox in a single request.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
data: List of dicts containing the object data to create
Returns:
List of created objects
Example:
To create multiple sites:
netbox_bulk_create_objects("sites", [
{"name": "Site A", "slug": "site-a", "status": "active"},
{"name": "Site B", "slug": "site-b", "status": "active"}
])
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call
return netbox.bulk_create(endpoint, data)
@mcp.tool()
def netbox_bulk_update_objects(object_type: str, data: list):
"""
Update multiple objects in NetBox in a single request.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
data: List of dicts containing the object data to update (must include "id" field)
Returns:
List of updated objects
Example:
To update multiple devices:
netbox_bulk_update_objects("devices", [
{"id": 1, "status": "offline"},
{"id": 2, "status": "maintenance"}
])
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call
return netbox.bulk_update(endpoint, data)
@mcp.tool()
def netbox_bulk_delete_objects(object_type: str, object_ids: list):
"""
Delete multiple objects from NetBox in a single request.
Args:
object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses")
object_ids: List of numeric IDs to delete
Returns:
Success status
WARNING: This permanently deletes the objects and cannot be undone!
Example:
To delete multiple devices:
netbox_bulk_delete_objects("devices", [5, 6, 7])
"""
# Validate object_type exists in mapping
if object_type not in NETBOX_OBJECT_TYPES:
valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys()))
raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}")
# Get API endpoint from mapping
endpoint = NETBOX_OBJECT_TYPES[object_type]
# Make API call
success = netbox.bulk_delete(endpoint, object_ids)
if success:
return {"success": True, "message": f"Successfully deleted {len(object_ids)} {object_type} objects"}
else:
return {"success": False, "message": f"Failed to delete {object_type} objects"}
if __name__ == "__main__":
# Load NetBox configuration from environment variables
netbox_url = os.getenv("NETBOX_URL")
netbox_token = os.getenv("NETBOX_TOKEN")
if not netbox_url or not netbox_token:
raise ValueError("NETBOX_URL and NETBOX_TOKEN environment variables must be set")
# Initialize NetBox client
netbox = NetBoxRestClient(url=netbox_url, token=netbox_token)
mcp.run(transport="stdio")