lmiranda
a31447e28f
Comprehensive MCP server covering the entire NetBox REST API: - DCIM: sites, racks, devices, interfaces, cables, power - IPAM: prefixes, IP addresses, VLANs, VRFs, ASNs - Circuits: providers, circuits, terminations - Virtualization: clusters, VMs, VM interfaces - Tenancy: tenants, contacts, contact assignments - VPN: tunnels, IKE/IPSec policies, L2VPN - Wireless: WLANs, links, groups - Extras: tags, custom fields, webhooks, audit log Features: - 100+ MCP tools for full CRUD operations - Auto-pagination handling - Hybrid config (system + project level) - Available prefix/IP allocation support 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
561 lines
22 KiB
Python
561 lines
22 KiB
Python
"""
|
|
Extras tools for NetBox MCP Server.
|
|
|
|
Covers: Tags, Custom Fields, Custom Links, Webhooks, Journal Entries, and more.
|
|
"""
|
|
import logging
|
|
from typing import List, Dict, Optional, Any
|
|
from ..netbox_client import NetBoxClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ExtrasTools:
|
|
"""Tools for Extras operations in NetBox"""
|
|
|
|
def __init__(self, client: NetBoxClient):
|
|
self.client = client
|
|
self.base_endpoint = 'extras'
|
|
|
|
# ==================== Tags ====================
|
|
|
|
async def list_tags(
|
|
self,
|
|
name: Optional[str] = None,
|
|
slug: Optional[str] = None,
|
|
color: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all tags with optional filtering."""
|
|
params = {k: v for k, v in {
|
|
'name': name, 'slug': slug, 'color': color, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/tags', params=params)
|
|
|
|
async def get_tag(self, id: int) -> Dict:
|
|
"""Get a specific tag by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/tags', id)
|
|
|
|
async def create_tag(
|
|
self,
|
|
name: str,
|
|
slug: str,
|
|
color: str = '9e9e9e',
|
|
description: Optional[str] = None,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new tag."""
|
|
data = {'name': name, 'slug': slug, 'color': color, **kwargs}
|
|
if description:
|
|
data['description'] = description
|
|
return self.client.create(f'{self.base_endpoint}/tags', data)
|
|
|
|
async def update_tag(self, id: int, **kwargs) -> Dict:
|
|
"""Update a tag."""
|
|
return self.client.patch(f'{self.base_endpoint}/tags', id, kwargs)
|
|
|
|
async def delete_tag(self, id: int) -> None:
|
|
"""Delete a tag."""
|
|
self.client.delete(f'{self.base_endpoint}/tags', id)
|
|
|
|
# ==================== Custom Fields ====================
|
|
|
|
async def list_custom_fields(
|
|
self,
|
|
name: Optional[str] = None,
|
|
type: Optional[str] = None,
|
|
content_types: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all custom fields."""
|
|
params = {k: v for k, v in {
|
|
'name': name, 'type': type, 'content_types': content_types, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/custom-fields', params=params)
|
|
|
|
async def get_custom_field(self, id: int) -> Dict:
|
|
"""Get a specific custom field by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/custom-fields', id)
|
|
|
|
async def create_custom_field(
|
|
self,
|
|
name: str,
|
|
content_types: List[str],
|
|
type: str = 'text',
|
|
label: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
required: bool = False,
|
|
filter_logic: str = 'loose',
|
|
default: Optional[Any] = None,
|
|
weight: int = 100,
|
|
validation_minimum: Optional[int] = None,
|
|
validation_maximum: Optional[int] = None,
|
|
validation_regex: Optional[str] = None,
|
|
choice_set: Optional[int] = None,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new custom field."""
|
|
data = {
|
|
'name': name, 'content_types': content_types, 'type': type,
|
|
'required': required, 'filter_logic': filter_logic, 'weight': weight, **kwargs
|
|
}
|
|
for key, val in [
|
|
('label', label), ('description', description), ('default', default),
|
|
('validation_minimum', validation_minimum), ('validation_maximum', validation_maximum),
|
|
('validation_regex', validation_regex), ('choice_set', choice_set)
|
|
]:
|
|
if val is not None:
|
|
data[key] = val
|
|
return self.client.create(f'{self.base_endpoint}/custom-fields', data)
|
|
|
|
async def update_custom_field(self, id: int, **kwargs) -> Dict:
|
|
"""Update a custom field."""
|
|
return self.client.patch(f'{self.base_endpoint}/custom-fields', id, kwargs)
|
|
|
|
async def delete_custom_field(self, id: int) -> None:
|
|
"""Delete a custom field."""
|
|
self.client.delete(f'{self.base_endpoint}/custom-fields', id)
|
|
|
|
# ==================== Custom Field Choice Sets ====================
|
|
|
|
async def list_custom_field_choice_sets(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
|
|
"""List all custom field choice sets."""
|
|
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/custom-field-choice-sets', params=params)
|
|
|
|
async def get_custom_field_choice_set(self, id: int) -> Dict:
|
|
"""Get a specific custom field choice set by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/custom-field-choice-sets', id)
|
|
|
|
async def create_custom_field_choice_set(
|
|
self,
|
|
name: str,
|
|
extra_choices: List[List[str]],
|
|
description: Optional[str] = None,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new custom field choice set."""
|
|
data = {'name': name, 'extra_choices': extra_choices, **kwargs}
|
|
if description:
|
|
data['description'] = description
|
|
return self.client.create(f'{self.base_endpoint}/custom-field-choice-sets', data)
|
|
|
|
async def update_custom_field_choice_set(self, id: int, **kwargs) -> Dict:
|
|
"""Update a custom field choice set."""
|
|
return self.client.patch(f'{self.base_endpoint}/custom-field-choice-sets', id, kwargs)
|
|
|
|
async def delete_custom_field_choice_set(self, id: int) -> None:
|
|
"""Delete a custom field choice set."""
|
|
self.client.delete(f'{self.base_endpoint}/custom-field-choice-sets', id)
|
|
|
|
# ==================== Custom Links ====================
|
|
|
|
async def list_custom_links(
|
|
self,
|
|
name: Optional[str] = None,
|
|
content_types: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all custom links."""
|
|
params = {k: v for k, v in {
|
|
'name': name, 'content_types': content_types, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/custom-links', params=params)
|
|
|
|
async def get_custom_link(self, id: int) -> Dict:
|
|
"""Get a specific custom link by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/custom-links', id)
|
|
|
|
async def create_custom_link(
|
|
self,
|
|
name: str,
|
|
content_types: List[str],
|
|
link_text: str,
|
|
link_url: str,
|
|
enabled: bool = True,
|
|
new_window: bool = False,
|
|
weight: int = 100,
|
|
group_name: Optional[str] = None,
|
|
button_class: str = 'outline-dark',
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new custom link."""
|
|
data = {
|
|
'name': name, 'content_types': content_types,
|
|
'link_text': link_text, 'link_url': link_url,
|
|
'enabled': enabled, 'new_window': new_window,
|
|
'weight': weight, 'button_class': button_class, **kwargs
|
|
}
|
|
if group_name:
|
|
data['group_name'] = group_name
|
|
return self.client.create(f'{self.base_endpoint}/custom-links', data)
|
|
|
|
async def update_custom_link(self, id: int, **kwargs) -> Dict:
|
|
"""Update a custom link."""
|
|
return self.client.patch(f'{self.base_endpoint}/custom-links', id, kwargs)
|
|
|
|
async def delete_custom_link(self, id: int) -> None:
|
|
"""Delete a custom link."""
|
|
self.client.delete(f'{self.base_endpoint}/custom-links', id)
|
|
|
|
# ==================== Webhooks ====================
|
|
|
|
async def list_webhooks(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
|
|
"""List all webhooks."""
|
|
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/webhooks', params=params)
|
|
|
|
async def get_webhook(self, id: int) -> Dict:
|
|
"""Get a specific webhook by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/webhooks', id)
|
|
|
|
async def create_webhook(
|
|
self,
|
|
name: str,
|
|
payload_url: str,
|
|
content_types: List[str],
|
|
type_create: bool = True,
|
|
type_update: bool = True,
|
|
type_delete: bool = True,
|
|
type_job_start: bool = False,
|
|
type_job_end: bool = False,
|
|
enabled: bool = True,
|
|
http_method: str = 'POST',
|
|
http_content_type: str = 'application/json',
|
|
additional_headers: Optional[str] = None,
|
|
body_template: Optional[str] = None,
|
|
secret: Optional[str] = None,
|
|
ssl_verification: bool = True,
|
|
ca_file_path: Optional[str] = None,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new webhook."""
|
|
data = {
|
|
'name': name, 'payload_url': payload_url, 'content_types': content_types,
|
|
'type_create': type_create, 'type_update': type_update, 'type_delete': type_delete,
|
|
'type_job_start': type_job_start, 'type_job_end': type_job_end,
|
|
'enabled': enabled, 'http_method': http_method,
|
|
'http_content_type': http_content_type, 'ssl_verification': ssl_verification, **kwargs
|
|
}
|
|
for key, val in [
|
|
('additional_headers', additional_headers), ('body_template', body_template),
|
|
('secret', secret), ('ca_file_path', ca_file_path)
|
|
]:
|
|
if val is not None:
|
|
data[key] = val
|
|
return self.client.create(f'{self.base_endpoint}/webhooks', data)
|
|
|
|
async def update_webhook(self, id: int, **kwargs) -> Dict:
|
|
"""Update a webhook."""
|
|
return self.client.patch(f'{self.base_endpoint}/webhooks', id, kwargs)
|
|
|
|
async def delete_webhook(self, id: int) -> None:
|
|
"""Delete a webhook."""
|
|
self.client.delete(f'{self.base_endpoint}/webhooks', id)
|
|
|
|
# ==================== Journal Entries ====================
|
|
|
|
async def list_journal_entries(
|
|
self,
|
|
assigned_object_type: Optional[str] = None,
|
|
assigned_object_id: Optional[int] = None,
|
|
kind: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all journal entries."""
|
|
params = {k: v for k, v in {
|
|
'assigned_object_type': assigned_object_type,
|
|
'assigned_object_id': assigned_object_id, 'kind': kind, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/journal-entries', params=params)
|
|
|
|
async def get_journal_entry(self, id: int) -> Dict:
|
|
"""Get a specific journal entry by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/journal-entries', id)
|
|
|
|
async def create_journal_entry(
|
|
self,
|
|
assigned_object_type: str,
|
|
assigned_object_id: int,
|
|
comments: str,
|
|
kind: str = 'info',
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new journal entry."""
|
|
data = {
|
|
'assigned_object_type': assigned_object_type,
|
|
'assigned_object_id': assigned_object_id,
|
|
'comments': comments, 'kind': kind, **kwargs
|
|
}
|
|
return self.client.create(f'{self.base_endpoint}/journal-entries', data)
|
|
|
|
async def update_journal_entry(self, id: int, **kwargs) -> Dict:
|
|
"""Update a journal entry."""
|
|
return self.client.patch(f'{self.base_endpoint}/journal-entries', id, kwargs)
|
|
|
|
async def delete_journal_entry(self, id: int) -> None:
|
|
"""Delete a journal entry."""
|
|
self.client.delete(f'{self.base_endpoint}/journal-entries', id)
|
|
|
|
# ==================== Config Contexts ====================
|
|
|
|
async def list_config_contexts(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
|
|
"""List all config contexts."""
|
|
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/config-contexts', params=params)
|
|
|
|
async def get_config_context(self, id: int) -> Dict:
|
|
"""Get a specific config context by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/config-contexts', id)
|
|
|
|
async def create_config_context(
|
|
self,
|
|
name: str,
|
|
data: Dict[str, Any],
|
|
weight: int = 1000,
|
|
description: Optional[str] = None,
|
|
is_active: bool = True,
|
|
regions: Optional[List[int]] = None,
|
|
site_groups: Optional[List[int]] = None,
|
|
sites: Optional[List[int]] = None,
|
|
locations: Optional[List[int]] = None,
|
|
device_types: Optional[List[int]] = None,
|
|
roles: Optional[List[int]] = None,
|
|
platforms: Optional[List[int]] = None,
|
|
cluster_types: Optional[List[int]] = None,
|
|
cluster_groups: Optional[List[int]] = None,
|
|
clusters: Optional[List[int]] = None,
|
|
tenant_groups: Optional[List[int]] = None,
|
|
tenants: Optional[List[int]] = None,
|
|
tags: Optional[List[str]] = None,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new config context."""
|
|
context_data = {
|
|
'name': name, 'data': data, 'weight': weight, 'is_active': is_active, **kwargs
|
|
}
|
|
for key, val in [
|
|
('description', description), ('regions', regions),
|
|
('site_groups', site_groups), ('sites', sites),
|
|
('locations', locations), ('device_types', device_types),
|
|
('roles', roles), ('platforms', platforms),
|
|
('cluster_types', cluster_types), ('cluster_groups', cluster_groups),
|
|
('clusters', clusters), ('tenant_groups', tenant_groups),
|
|
('tenants', tenants), ('tags', tags)
|
|
]:
|
|
if val is not None:
|
|
context_data[key] = val
|
|
return self.client.create(f'{self.base_endpoint}/config-contexts', context_data)
|
|
|
|
async def update_config_context(self, id: int, **kwargs) -> Dict:
|
|
"""Update a config context."""
|
|
return self.client.patch(f'{self.base_endpoint}/config-contexts', id, kwargs)
|
|
|
|
async def delete_config_context(self, id: int) -> None:
|
|
"""Delete a config context."""
|
|
self.client.delete(f'{self.base_endpoint}/config-contexts', id)
|
|
|
|
# ==================== Config Templates ====================
|
|
|
|
async def list_config_templates(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
|
|
"""List all config templates."""
|
|
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/config-templates', params=params)
|
|
|
|
async def get_config_template(self, id: int) -> Dict:
|
|
"""Get a specific config template by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/config-templates', id)
|
|
|
|
async def create_config_template(
|
|
self,
|
|
name: str,
|
|
template_code: str,
|
|
description: Optional[str] = None,
|
|
environment_params: Optional[Dict[str, Any]] = None,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new config template."""
|
|
data = {'name': name, 'template_code': template_code, **kwargs}
|
|
if description:
|
|
data['description'] = description
|
|
if environment_params:
|
|
data['environment_params'] = environment_params
|
|
return self.client.create(f'{self.base_endpoint}/config-templates', data)
|
|
|
|
async def update_config_template(self, id: int, **kwargs) -> Dict:
|
|
"""Update a config template."""
|
|
return self.client.patch(f'{self.base_endpoint}/config-templates', id, kwargs)
|
|
|
|
async def delete_config_template(self, id: int) -> None:
|
|
"""Delete a config template."""
|
|
self.client.delete(f'{self.base_endpoint}/config-templates', id)
|
|
|
|
# ==================== Export Templates ====================
|
|
|
|
async def list_export_templates(
|
|
self,
|
|
name: Optional[str] = None,
|
|
content_types: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all export templates."""
|
|
params = {k: v for k, v in {
|
|
'name': name, 'content_types': content_types, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/export-templates', params=params)
|
|
|
|
async def get_export_template(self, id: int) -> Dict:
|
|
"""Get a specific export template by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/export-templates', id)
|
|
|
|
async def create_export_template(
|
|
self,
|
|
name: str,
|
|
content_types: List[str],
|
|
template_code: str,
|
|
description: Optional[str] = None,
|
|
mime_type: str = 'text/plain',
|
|
file_extension: Optional[str] = None,
|
|
as_attachment: bool = True,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new export template."""
|
|
data = {
|
|
'name': name, 'content_types': content_types,
|
|
'template_code': template_code, 'mime_type': mime_type,
|
|
'as_attachment': as_attachment, **kwargs
|
|
}
|
|
if description:
|
|
data['description'] = description
|
|
if file_extension:
|
|
data['file_extension'] = file_extension
|
|
return self.client.create(f'{self.base_endpoint}/export-templates', data)
|
|
|
|
async def update_export_template(self, id: int, **kwargs) -> Dict:
|
|
"""Update an export template."""
|
|
return self.client.patch(f'{self.base_endpoint}/export-templates', id, kwargs)
|
|
|
|
async def delete_export_template(self, id: int) -> None:
|
|
"""Delete an export template."""
|
|
self.client.delete(f'{self.base_endpoint}/export-templates', id)
|
|
|
|
# ==================== Saved Filters ====================
|
|
|
|
async def list_saved_filters(
|
|
self,
|
|
name: Optional[str] = None,
|
|
content_types: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all saved filters."""
|
|
params = {k: v for k, v in {
|
|
'name': name, 'content_types': content_types, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/saved-filters', params=params)
|
|
|
|
async def get_saved_filter(self, id: int) -> Dict:
|
|
"""Get a specific saved filter by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/saved-filters', id)
|
|
|
|
async def create_saved_filter(
|
|
self,
|
|
name: str,
|
|
slug: str,
|
|
content_types: List[str],
|
|
parameters: Dict[str, Any],
|
|
description: Optional[str] = None,
|
|
weight: int = 100,
|
|
enabled: bool = True,
|
|
shared: bool = True,
|
|
**kwargs
|
|
) -> Dict:
|
|
"""Create a new saved filter."""
|
|
data = {
|
|
'name': name, 'slug': slug, 'content_types': content_types,
|
|
'parameters': parameters, 'weight': weight,
|
|
'enabled': enabled, 'shared': shared, **kwargs
|
|
}
|
|
if description:
|
|
data['description'] = description
|
|
return self.client.create(f'{self.base_endpoint}/saved-filters', data)
|
|
|
|
async def update_saved_filter(self, id: int, **kwargs) -> Dict:
|
|
"""Update a saved filter."""
|
|
return self.client.patch(f'{self.base_endpoint}/saved-filters', id, kwargs)
|
|
|
|
async def delete_saved_filter(self, id: int) -> None:
|
|
"""Delete a saved filter."""
|
|
self.client.delete(f'{self.base_endpoint}/saved-filters', id)
|
|
|
|
# ==================== Image Attachments ====================
|
|
|
|
async def list_image_attachments(
|
|
self,
|
|
object_type: Optional[str] = None,
|
|
object_id: Optional[int] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all image attachments."""
|
|
params = {k: v for k, v in {
|
|
'object_type': object_type, 'object_id': object_id, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/image-attachments', params=params)
|
|
|
|
async def get_image_attachment(self, id: int) -> Dict:
|
|
"""Get a specific image attachment by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/image-attachments', id)
|
|
|
|
async def delete_image_attachment(self, id: int) -> None:
|
|
"""Delete an image attachment."""
|
|
self.client.delete(f'{self.base_endpoint}/image-attachments', id)
|
|
|
|
# ==================== Object Changes (Audit Log) ====================
|
|
|
|
async def list_object_changes(
|
|
self,
|
|
user_id: Optional[int] = None,
|
|
changed_object_type: Optional[str] = None,
|
|
changed_object_id: Optional[int] = None,
|
|
action: Optional[str] = None,
|
|
**kwargs
|
|
) -> List[Dict]:
|
|
"""List all object changes (audit log)."""
|
|
params = {k: v for k, v in {
|
|
'user_id': user_id, 'changed_object_type': changed_object_type,
|
|
'changed_object_id': changed_object_id, 'action': action, **kwargs
|
|
}.items() if v is not None}
|
|
return self.client.list(f'{self.base_endpoint}/object-changes', params=params)
|
|
|
|
async def get_object_change(self, id: int) -> Dict:
|
|
"""Get a specific object change by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/object-changes', id)
|
|
|
|
# ==================== Scripts ====================
|
|
|
|
async def list_scripts(self, **kwargs) -> List[Dict]:
|
|
"""List all available scripts."""
|
|
return self.client.list(f'{self.base_endpoint}/scripts', params=kwargs)
|
|
|
|
async def get_script(self, id: str) -> Dict:
|
|
"""Get a specific script by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/scripts', id)
|
|
|
|
async def run_script(self, id: str, data: Dict[str, Any], commit: bool = True) -> Dict:
|
|
"""Run a script with the provided data."""
|
|
payload = {'data': data, 'commit': commit}
|
|
return self.client.create(f'{self.base_endpoint}/scripts/{id}', payload)
|
|
|
|
# ==================== Reports ====================
|
|
|
|
async def list_reports(self, **kwargs) -> List[Dict]:
|
|
"""List all available reports."""
|
|
return self.client.list(f'{self.base_endpoint}/reports', params=kwargs)
|
|
|
|
async def get_report(self, id: str) -> Dict:
|
|
"""Get a specific report by ID."""
|
|
return self.client.get(f'{self.base_endpoint}/reports', id)
|
|
|
|
async def run_report(self, id: str) -> Dict:
|
|
"""Run a report."""
|
|
return self.client.create(f'{self.base_endpoint}/reports/{id}', {})
|