feat: add NetBox MCP server for infrastructure management

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>
This commit is contained in:
2025-12-03 12:15:17 -05:00
parent c9d8a91c83
commit a31447e28f
16 changed files with 5849 additions and 0 deletions

View File

@@ -0,0 +1 @@
"""NetBox MCP Server for Claude Code integration."""

View File

@@ -0,0 +1,108 @@
"""
Configuration loader for NetBox MCP Server.
Implements hybrid configuration system:
- System-level: ~/.config/claude/netbox.env (credentials)
- Project-level: .env (optional overrides)
"""
from pathlib import Path
from dotenv import load_dotenv
import os
import logging
from typing import Dict, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class NetBoxConfig:
"""Configuration loader for NetBox MCP Server"""
def __init__(self):
self.api_url: Optional[str] = None
self.api_token: Optional[str] = None
self.verify_ssl: bool = True
self.timeout: int = 30
def load(self) -> Dict[str, any]:
"""
Load configuration from system and project levels.
Project-level configuration overrides system-level.
Returns:
Dict containing api_url, api_token, verify_ssl, timeout
Raises:
FileNotFoundError: If system config is missing
ValueError: If required configuration is missing
"""
# Load system config
system_config = Path.home() / '.config' / 'claude' / 'netbox.env'
if system_config.exists():
load_dotenv(system_config)
logger.info(f"Loaded system configuration from {system_config}")
else:
raise FileNotFoundError(
f"System config not found: {system_config}\n"
"Create it with:\n"
" mkdir -p ~/.config/claude\n"
" cat > ~/.config/claude/netbox.env << EOF\n"
" NETBOX_API_URL=https://your-netbox-instance/api\n"
" NETBOX_API_TOKEN=your-api-token\n"
" EOF"
)
# Load project config (overrides system)
project_config = Path.cwd() / '.env'
if project_config.exists():
load_dotenv(project_config, override=True)
logger.info(f"Loaded project configuration from {project_config}")
# Extract values
self.api_url = os.getenv('NETBOX_API_URL')
self.api_token = os.getenv('NETBOX_API_TOKEN')
# Optional settings with defaults
verify_ssl_str = os.getenv('NETBOX_VERIFY_SSL', 'true').lower()
self.verify_ssl = verify_ssl_str in ('true', '1', 'yes')
timeout_str = os.getenv('NETBOX_TIMEOUT', '30')
try:
self.timeout = int(timeout_str)
except ValueError:
self.timeout = 30
logger.warning(f"Invalid NETBOX_TIMEOUT value '{timeout_str}', using default 30")
# Validate required variables
self._validate()
# Normalize API URL (remove trailing slash)
if self.api_url and self.api_url.endswith('/'):
self.api_url = self.api_url.rstrip('/')
return {
'api_url': self.api_url,
'api_token': self.api_token,
'verify_ssl': self.verify_ssl,
'timeout': self.timeout
}
def _validate(self) -> None:
"""
Validate that required configuration is present.
Raises:
ValueError: If required configuration is missing
"""
required = {
'NETBOX_API_URL': self.api_url,
'NETBOX_API_TOKEN': self.api_token
}
missing = [key for key, value in required.items() if not value]
if missing:
raise ValueError(
f"Missing required configuration: {', '.join(missing)}\n"
"Check your ~/.config/claude/netbox.env file"
)

View File

@@ -0,0 +1,294 @@
"""
NetBox API client for interacting with NetBox REST API.
Provides a generic HTTP client with methods for all standard REST operations.
Individual tool modules use this client for their specific endpoints.
"""
import requests
import logging
from typing import List, Dict, Optional, Any, Union
from urllib.parse import urljoin
from .config import NetBoxConfig
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class NetBoxClient:
"""Generic client for interacting with NetBox REST API"""
def __init__(self):
"""Initialize NetBox client with configuration"""
config = NetBoxConfig()
config_dict = config.load()
self.base_url = config_dict['api_url']
self.token = config_dict['api_token']
self.verify_ssl = config_dict['verify_ssl']
self.timeout = config_dict['timeout']
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Token {self.token}',
'Content-Type': 'application/json',
'Accept': 'application/json'
})
self.session.verify = self.verify_ssl
logger.info(f"NetBox client initialized for {self.base_url}")
def _build_url(self, endpoint: str) -> str:
"""
Build full URL for API endpoint.
Args:
endpoint: API endpoint path (e.g., 'dcim/devices/')
Returns:
Full URL
"""
# Ensure endpoint starts with /
if not endpoint.startswith('/'):
endpoint = '/' + endpoint
# Ensure endpoint ends with /
if not endpoint.endswith('/'):
endpoint = endpoint + '/'
return f"{self.base_url}{endpoint}"
def _handle_response(self, response: requests.Response) -> Any:
"""
Handle API response and raise appropriate errors.
Args:
response: requests Response object
Returns:
Parsed JSON response
Raises:
requests.HTTPError: If request failed
"""
try:
response.raise_for_status()
except requests.HTTPError as e:
# Try to get error details from response
try:
error_detail = response.json()
logger.error(f"API error: {error_detail}")
except Exception:
logger.error(f"API error: {response.text}")
raise e
# Handle empty responses (e.g., DELETE)
if response.status_code == 204 or not response.content:
return None
return response.json()
def list(
self,
endpoint: str,
params: Optional[Dict[str, Any]] = None,
paginate: bool = True,
limit: int = 50
) -> List[Dict]:
"""
List objects from an endpoint with optional pagination.
Args:
endpoint: API endpoint path
params: Query parameters for filtering
paginate: Whether to handle pagination automatically
limit: Number of results per page
Returns:
List of objects
"""
url = self._build_url(endpoint)
params = params or {}
params['limit'] = limit
logger.info(f"Listing objects from {endpoint}")
if not paginate:
response = self.session.get(url, params=params, timeout=self.timeout)
result = self._handle_response(response)
return result.get('results', []) if isinstance(result, dict) else result
# Handle pagination
all_results = []
while url:
response = self.session.get(url, params=params, timeout=self.timeout)
result = self._handle_response(response)
if isinstance(result, dict):
all_results.extend(result.get('results', []))
url = result.get('next')
params = {} # Next URL already contains params
else:
all_results.extend(result)
break
return all_results
def get(self, endpoint: str, id: Union[int, str]) -> Dict:
"""
Get a single object by ID.
Args:
endpoint: API endpoint path
id: Object ID
Returns:
Object dictionary
"""
url = self._build_url(f"{endpoint}/{id}")
logger.info(f"Getting object {id} from {endpoint}")
response = self.session.get(url, timeout=self.timeout)
return self._handle_response(response)
def create(self, endpoint: str, data: Dict) -> Dict:
"""
Create a new object.
Args:
endpoint: API endpoint path
data: Object data
Returns:
Created object dictionary
"""
url = self._build_url(endpoint)
logger.info(f"Creating object in {endpoint}")
response = self.session.post(url, json=data, timeout=self.timeout)
return self._handle_response(response)
def create_bulk(self, endpoint: str, data: List[Dict]) -> List[Dict]:
"""
Create multiple objects in bulk.
Args:
endpoint: API endpoint path
data: List of object data
Returns:
List of created objects
"""
url = self._build_url(endpoint)
logger.info(f"Bulk creating {len(data)} objects in {endpoint}")
response = self.session.post(url, json=data, timeout=self.timeout)
return self._handle_response(response)
def update(self, endpoint: str, id: Union[int, str], data: Dict) -> Dict:
"""
Update an existing object (full update).
Args:
endpoint: API endpoint path
id: Object ID
data: Updated object data
Returns:
Updated object dictionary
"""
url = self._build_url(f"{endpoint}/{id}")
logger.info(f"Updating object {id} in {endpoint}")
response = self.session.put(url, json=data, timeout=self.timeout)
return self._handle_response(response)
def patch(self, endpoint: str, id: Union[int, str], data: Dict) -> Dict:
"""
Partially update an existing object.
Args:
endpoint: API endpoint path
id: Object ID
data: Fields to update
Returns:
Updated object dictionary
"""
url = self._build_url(f"{endpoint}/{id}")
logger.info(f"Patching object {id} in {endpoint}")
response = self.session.patch(url, json=data, timeout=self.timeout)
return self._handle_response(response)
def delete(self, endpoint: str, id: Union[int, str]) -> None:
"""
Delete an object.
Args:
endpoint: API endpoint path
id: Object ID
"""
url = self._build_url(f"{endpoint}/{id}")
logger.info(f"Deleting object {id} from {endpoint}")
response = self.session.delete(url, timeout=self.timeout)
self._handle_response(response)
def delete_bulk(self, endpoint: str, ids: List[Union[int, str]]) -> None:
"""
Delete multiple objects in bulk.
Args:
endpoint: API endpoint path
ids: List of object IDs
"""
url = self._build_url(endpoint)
data = [{'id': id} for id in ids]
logger.info(f"Bulk deleting {len(ids)} objects from {endpoint}")
response = self.session.delete(url, json=data, timeout=self.timeout)
self._handle_response(response)
def options(self, endpoint: str) -> Dict:
"""
Get available options for an endpoint (schema info).
Args:
endpoint: API endpoint path
Returns:
Options/schema dictionary
"""
url = self._build_url(endpoint)
logger.info(f"Getting options for {endpoint}")
response = self.session.options(url, timeout=self.timeout)
return self._handle_response(response)
def search(
self,
endpoint: str,
query: str,
params: Optional[Dict[str, Any]] = None
) -> List[Dict]:
"""
Search objects using the 'q' parameter.
Args:
endpoint: API endpoint path
query: Search query string
params: Additional filter parameters
Returns:
List of matching objects
"""
params = params or {}
params['q'] = query
return self.list(endpoint, params=params)
def filter(
self,
endpoint: str,
**filters
) -> List[Dict]:
"""
Filter objects by various fields.
Args:
endpoint: API endpoint path
**filters: Filter parameters (field=value)
Returns:
List of matching objects
"""
return self.list(endpoint, params=filters)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,20 @@
"""NetBox MCP tools package."""
from .dcim import DCIMTools
from .ipam import IPAMTools
from .circuits import CircuitsTools
from .virtualization import VirtualizationTools
from .tenancy import TenancyTools
from .vpn import VPNTools
from .wireless import WirelessTools
from .extras import ExtrasTools
__all__ = [
'DCIMTools',
'IPAMTools',
'CircuitsTools',
'VirtualizationTools',
'TenancyTools',
'VPNTools',
'WirelessTools',
'ExtrasTools',
]

View File

@@ -0,0 +1,373 @@
"""
Circuits tools for NetBox MCP Server.
Covers: Providers, Circuits, Circuit Types, Circuit Terminations, and related models.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class CircuitsTools:
"""Tools for Circuits operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'circuits'
# ==================== Providers ====================
async def list_providers(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all circuit providers."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/providers', params=params)
async def get_provider(self, id: int) -> Dict:
"""Get a specific provider by ID."""
return self.client.get(f'{self.base_endpoint}/providers', id)
async def create_provider(
self,
name: str,
slug: str,
asns: Optional[List[int]] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new provider."""
data = {'name': name, 'slug': slug, **kwargs}
if asns:
data['asns'] = asns
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/providers', data)
async def update_provider(self, id: int, **kwargs) -> Dict:
"""Update a provider."""
return self.client.patch(f'{self.base_endpoint}/providers', id, kwargs)
async def delete_provider(self, id: int) -> None:
"""Delete a provider."""
self.client.delete(f'{self.base_endpoint}/providers', id)
# ==================== Provider Accounts ====================
async def list_provider_accounts(
self,
provider_id: Optional[int] = None,
name: Optional[str] = None,
account: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all provider accounts."""
params = {k: v for k, v in {
'provider_id': provider_id, 'name': name, 'account': account, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/provider-accounts', params=params)
async def get_provider_account(self, id: int) -> Dict:
"""Get a specific provider account by ID."""
return self.client.get(f'{self.base_endpoint}/provider-accounts', id)
async def create_provider_account(
self,
provider: int,
account: str,
name: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new provider account."""
data = {'provider': provider, 'account': account, **kwargs}
if name:
data['name'] = name
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/provider-accounts', data)
async def update_provider_account(self, id: int, **kwargs) -> Dict:
"""Update a provider account."""
return self.client.patch(f'{self.base_endpoint}/provider-accounts', id, kwargs)
async def delete_provider_account(self, id: int) -> None:
"""Delete a provider account."""
self.client.delete(f'{self.base_endpoint}/provider-accounts', id)
# ==================== Provider Networks ====================
async def list_provider_networks(
self,
provider_id: Optional[int] = None,
name: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all provider networks."""
params = {k: v for k, v in {
'provider_id': provider_id, 'name': name, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/provider-networks', params=params)
async def get_provider_network(self, id: int) -> Dict:
"""Get a specific provider network by ID."""
return self.client.get(f'{self.base_endpoint}/provider-networks', id)
async def create_provider_network(
self,
provider: int,
name: str,
service_id: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new provider network."""
data = {'provider': provider, 'name': name, **kwargs}
if service_id:
data['service_id'] = service_id
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/provider-networks', data)
async def update_provider_network(self, id: int, **kwargs) -> Dict:
"""Update a provider network."""
return self.client.patch(f'{self.base_endpoint}/provider-networks', id, kwargs)
async def delete_provider_network(self, id: int) -> None:
"""Delete a provider network."""
self.client.delete(f'{self.base_endpoint}/provider-networks', id)
# ==================== Circuit Types ====================
async def list_circuit_types(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all circuit types."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/circuit-types', params=params)
async def get_circuit_type(self, id: int) -> Dict:
"""Get a specific circuit type by ID."""
return self.client.get(f'{self.base_endpoint}/circuit-types', id)
async def create_circuit_type(
self,
name: str,
slug: str,
color: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new circuit type."""
data = {'name': name, 'slug': slug, **kwargs}
if color:
data['color'] = color
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/circuit-types', data)
async def update_circuit_type(self, id: int, **kwargs) -> Dict:
"""Update a circuit type."""
return self.client.patch(f'{self.base_endpoint}/circuit-types', id, kwargs)
async def delete_circuit_type(self, id: int) -> None:
"""Delete a circuit type."""
self.client.delete(f'{self.base_endpoint}/circuit-types', id)
# ==================== Circuit Groups ====================
async def list_circuit_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all circuit groups."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/circuit-groups', params=params)
async def get_circuit_group(self, id: int) -> Dict:
"""Get a specific circuit group by ID."""
return self.client.get(f'{self.base_endpoint}/circuit-groups', id)
async def create_circuit_group(
self,
name: str,
slug: str,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new circuit group."""
data = {'name': name, 'slug': slug, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/circuit-groups', data)
async def update_circuit_group(self, id: int, **kwargs) -> Dict:
"""Update a circuit group."""
return self.client.patch(f'{self.base_endpoint}/circuit-groups', id, kwargs)
async def delete_circuit_group(self, id: int) -> None:
"""Delete a circuit group."""
self.client.delete(f'{self.base_endpoint}/circuit-groups', id)
# ==================== Circuit Group Assignments ====================
async def list_circuit_group_assignments(
self,
group_id: Optional[int] = None,
circuit_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all circuit group assignments."""
params = {k: v for k, v in {
'group_id': group_id, 'circuit_id': circuit_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/circuit-group-assignments', params=params)
async def get_circuit_group_assignment(self, id: int) -> Dict:
"""Get a specific circuit group assignment by ID."""
return self.client.get(f'{self.base_endpoint}/circuit-group-assignments', id)
async def create_circuit_group_assignment(
self,
group: int,
circuit: int,
priority: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new circuit group assignment."""
data = {'group': group, 'circuit': circuit, **kwargs}
if priority:
data['priority'] = priority
return self.client.create(f'{self.base_endpoint}/circuit-group-assignments', data)
async def update_circuit_group_assignment(self, id: int, **kwargs) -> Dict:
"""Update a circuit group assignment."""
return self.client.patch(f'{self.base_endpoint}/circuit-group-assignments', id, kwargs)
async def delete_circuit_group_assignment(self, id: int) -> None:
"""Delete a circuit group assignment."""
self.client.delete(f'{self.base_endpoint}/circuit-group-assignments', id)
# ==================== Circuits ====================
async def list_circuits(
self,
cid: Optional[str] = None,
provider_id: Optional[int] = None,
provider_account_id: Optional[int] = None,
type_id: Optional[int] = None,
status: Optional[str] = None,
tenant_id: Optional[int] = None,
site_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all circuits with optional filtering."""
params = {k: v for k, v in {
'cid': cid, 'provider_id': provider_id, 'provider_account_id': provider_account_id,
'type_id': type_id, 'status': status, 'tenant_id': tenant_id, 'site_id': site_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/circuits', params=params)
async def get_circuit(self, id: int) -> Dict:
"""Get a specific circuit by ID."""
return self.client.get(f'{self.base_endpoint}/circuits', id)
async def create_circuit(
self,
cid: str,
provider: int,
type: int,
status: str = 'active',
provider_account: Optional[int] = None,
tenant: Optional[int] = None,
install_date: Optional[str] = None,
termination_date: Optional[str] = None,
commit_rate: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new circuit."""
data = {'cid': cid, 'provider': provider, 'type': type, 'status': status, **kwargs}
for key, val in [
('provider_account', provider_account), ('tenant', tenant),
('install_date', install_date), ('termination_date', termination_date),
('commit_rate', commit_rate), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/circuits', data)
async def update_circuit(self, id: int, **kwargs) -> Dict:
"""Update a circuit."""
return self.client.patch(f'{self.base_endpoint}/circuits', id, kwargs)
async def delete_circuit(self, id: int) -> None:
"""Delete a circuit."""
self.client.delete(f'{self.base_endpoint}/circuits', id)
# ==================== Circuit Terminations ====================
async def list_circuit_terminations(
self,
circuit_id: Optional[int] = None,
site_id: Optional[int] = None,
provider_network_id: Optional[int] = None,
term_side: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all circuit terminations."""
params = {k: v for k, v in {
'circuit_id': circuit_id, 'site_id': site_id,
'provider_network_id': provider_network_id, 'term_side': term_side, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/circuit-terminations', params=params)
async def get_circuit_termination(self, id: int) -> Dict:
"""Get a specific circuit termination by ID."""
return self.client.get(f'{self.base_endpoint}/circuit-terminations', id)
async def create_circuit_termination(
self,
circuit: int,
term_side: str,
site: Optional[int] = None,
provider_network: Optional[int] = None,
port_speed: Optional[int] = None,
upstream_speed: Optional[int] = None,
xconnect_id: Optional[str] = None,
pp_info: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new circuit termination."""
data = {'circuit': circuit, 'term_side': term_side, **kwargs}
for key, val in [
('site', site), ('provider_network', provider_network),
('port_speed', port_speed), ('upstream_speed', upstream_speed),
('xconnect_id', xconnect_id), ('pp_info', pp_info), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/circuit-terminations', data)
async def update_circuit_termination(self, id: int, **kwargs) -> Dict:
"""Update a circuit termination."""
return self.client.patch(f'{self.base_endpoint}/circuit-terminations', id, kwargs)
async def delete_circuit_termination(self, id: int) -> None:
"""Delete a circuit termination."""
self.client.delete(f'{self.base_endpoint}/circuit-terminations', id)
async def get_circuit_termination_paths(self, id: int) -> Dict:
"""Get cable paths for a circuit termination."""
return self.client.get(f'{self.base_endpoint}/circuit-terminations', f'{id}/paths')

View File

@@ -0,0 +1,935 @@
"""
DCIM (Data Center Infrastructure Management) tools for NetBox MCP Server.
Covers: Sites, Locations, Racks, Devices, Cables, Interfaces, and related models.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class DCIMTools:
"""Tools for DCIM operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'dcim'
# ==================== Regions ====================
async def list_regions(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
parent_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all regions with optional filtering."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'parent_id': parent_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/regions', params=params)
async def get_region(self, id: int) -> Dict:
"""Get a specific region by ID."""
return self.client.get(f'{self.base_endpoint}/regions', id)
async def create_region(self, name: str, slug: str, parent: Optional[int] = None, **kwargs) -> Dict:
"""Create a new region."""
data = {'name': name, 'slug': slug, **kwargs}
if parent:
data['parent'] = parent
return self.client.create(f'{self.base_endpoint}/regions', data)
async def update_region(self, id: int, **kwargs) -> Dict:
"""Update a region."""
return self.client.patch(f'{self.base_endpoint}/regions', id, kwargs)
async def delete_region(self, id: int) -> None:
"""Delete a region."""
self.client.delete(f'{self.base_endpoint}/regions', id)
# ==================== Site Groups ====================
async def list_site_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
parent_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all site groups with optional filtering."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'parent_id': parent_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/site-groups', params=params)
async def get_site_group(self, id: int) -> Dict:
"""Get a specific site group by ID."""
return self.client.get(f'{self.base_endpoint}/site-groups', id)
async def create_site_group(self, name: str, slug: str, parent: Optional[int] = None, **kwargs) -> Dict:
"""Create a new site group."""
data = {'name': name, 'slug': slug, **kwargs}
if parent:
data['parent'] = parent
return self.client.create(f'{self.base_endpoint}/site-groups', data)
async def update_site_group(self, id: int, **kwargs) -> Dict:
"""Update a site group."""
return self.client.patch(f'{self.base_endpoint}/site-groups', id, kwargs)
async def delete_site_group(self, id: int) -> None:
"""Delete a site group."""
self.client.delete(f'{self.base_endpoint}/site-groups', id)
# ==================== Sites ====================
async def list_sites(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
status: Optional[str] = None,
region_id: Optional[int] = None,
group_id: Optional[int] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all sites with optional filtering."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'status': status,
'region_id': region_id, 'group_id': group_id, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/sites', params=params)
async def get_site(self, id: int) -> Dict:
"""Get a specific site by ID."""
return self.client.get(f'{self.base_endpoint}/sites', id)
async def create_site(
self,
name: str,
slug: str,
status: str = 'active',
region: Optional[int] = None,
group: Optional[int] = None,
tenant: Optional[int] = None,
facility: Optional[str] = None,
time_zone: Optional[str] = None,
description: Optional[str] = None,
physical_address: Optional[str] = None,
shipping_address: Optional[str] = None,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
**kwargs
) -> Dict:
"""Create a new site."""
data = {'name': name, 'slug': slug, 'status': status, **kwargs}
for key, val in [
('region', region), ('group', group), ('tenant', tenant),
('facility', facility), ('time_zone', time_zone),
('description', description), ('physical_address', physical_address),
('shipping_address', shipping_address), ('latitude', latitude),
('longitude', longitude)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/sites', data)
async def update_site(self, id: int, **kwargs) -> Dict:
"""Update a site."""
return self.client.patch(f'{self.base_endpoint}/sites', id, kwargs)
async def delete_site(self, id: int) -> None:
"""Delete a site."""
self.client.delete(f'{self.base_endpoint}/sites', id)
# ==================== Locations ====================
async def list_locations(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
site_id: Optional[int] = None,
parent_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all locations with optional filtering."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'site_id': site_id, 'parent_id': parent_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/locations', params=params)
async def get_location(self, id: int) -> Dict:
"""Get a specific location by ID."""
return self.client.get(f'{self.base_endpoint}/locations', id)
async def create_location(
self,
name: str,
slug: str,
site: int,
parent: Optional[int] = None,
**kwargs
) -> Dict:
"""Create a new location."""
data = {'name': name, 'slug': slug, 'site': site, **kwargs}
if parent:
data['parent'] = parent
return self.client.create(f'{self.base_endpoint}/locations', data)
async def update_location(self, id: int, **kwargs) -> Dict:
"""Update a location."""
return self.client.patch(f'{self.base_endpoint}/locations', id, kwargs)
async def delete_location(self, id: int) -> None:
"""Delete a location."""
self.client.delete(f'{self.base_endpoint}/locations', id)
# ==================== Rack Roles ====================
async def list_rack_roles(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all rack roles."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/rack-roles', params=params)
async def get_rack_role(self, id: int) -> Dict:
"""Get a specific rack role by ID."""
return self.client.get(f'{self.base_endpoint}/rack-roles', id)
async def create_rack_role(self, name: str, slug: str, color: str = '9e9e9e', **kwargs) -> Dict:
"""Create a new rack role."""
data = {'name': name, 'slug': slug, 'color': color, **kwargs}
return self.client.create(f'{self.base_endpoint}/rack-roles', data)
async def update_rack_role(self, id: int, **kwargs) -> Dict:
"""Update a rack role."""
return self.client.patch(f'{self.base_endpoint}/rack-roles', id, kwargs)
async def delete_rack_role(self, id: int) -> None:
"""Delete a rack role."""
self.client.delete(f'{self.base_endpoint}/rack-roles', id)
# ==================== Rack Types ====================
async def list_rack_types(self, manufacturer_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all rack types."""
params = {k: v for k, v in {'manufacturer_id': manufacturer_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/rack-types', params=params)
async def get_rack_type(self, id: int) -> Dict:
"""Get a specific rack type by ID."""
return self.client.get(f'{self.base_endpoint}/rack-types', id)
async def create_rack_type(
self,
manufacturer: int,
model: str,
slug: str,
form_factor: str = '4-post-frame',
width: int = 19,
u_height: int = 42,
**kwargs
) -> Dict:
"""Create a new rack type."""
data = {
'manufacturer': manufacturer, 'model': model, 'slug': slug,
'form_factor': form_factor, 'width': width, 'u_height': u_height, **kwargs
}
return self.client.create(f'{self.base_endpoint}/rack-types', data)
async def update_rack_type(self, id: int, **kwargs) -> Dict:
"""Update a rack type."""
return self.client.patch(f'{self.base_endpoint}/rack-types', id, kwargs)
async def delete_rack_type(self, id: int) -> None:
"""Delete a rack type."""
self.client.delete(f'{self.base_endpoint}/rack-types', id)
# ==================== Racks ====================
async def list_racks(
self,
name: Optional[str] = None,
site_id: Optional[int] = None,
location_id: Optional[int] = None,
status: Optional[str] = None,
role_id: Optional[int] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all racks with optional filtering."""
params = {k: v for k, v in {
'name': name, 'site_id': site_id, 'location_id': location_id,
'status': status, 'role_id': role_id, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/racks', params=params)
async def get_rack(self, id: int) -> Dict:
"""Get a specific rack by ID."""
return self.client.get(f'{self.base_endpoint}/racks', id)
async def create_rack(
self,
name: str,
site: int,
status: str = 'active',
location: Optional[int] = None,
role: Optional[int] = None,
tenant: Optional[int] = None,
rack_type: Optional[int] = None,
width: int = 19,
u_height: int = 42,
**kwargs
) -> Dict:
"""Create a new rack."""
data = {'name': name, 'site': site, 'status': status, 'width': width, 'u_height': u_height, **kwargs}
for key, val in [('location', location), ('role', role), ('tenant', tenant), ('rack_type', rack_type)]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/racks', data)
async def update_rack(self, id: int, **kwargs) -> Dict:
"""Update a rack."""
return self.client.patch(f'{self.base_endpoint}/racks', id, kwargs)
async def delete_rack(self, id: int) -> None:
"""Delete a rack."""
self.client.delete(f'{self.base_endpoint}/racks', id)
# ==================== Rack Reservations ====================
async def list_rack_reservations(
self,
rack_id: Optional[int] = None,
site_id: Optional[int] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all rack reservations."""
params = {k: v for k, v in {
'rack_id': rack_id, 'site_id': site_id, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/rack-reservations', params=params)
async def get_rack_reservation(self, id: int) -> Dict:
"""Get a specific rack reservation by ID."""
return self.client.get(f'{self.base_endpoint}/rack-reservations', id)
async def create_rack_reservation(
self,
rack: int,
units: List[int],
user: int,
description: str,
tenant: Optional[int] = None,
**kwargs
) -> Dict:
"""Create a new rack reservation."""
data = {'rack': rack, 'units': units, 'user': user, 'description': description, **kwargs}
if tenant:
data['tenant'] = tenant
return self.client.create(f'{self.base_endpoint}/rack-reservations', data)
async def update_rack_reservation(self, id: int, **kwargs) -> Dict:
"""Update a rack reservation."""
return self.client.patch(f'{self.base_endpoint}/rack-reservations', id, kwargs)
async def delete_rack_reservation(self, id: int) -> None:
"""Delete a rack reservation."""
self.client.delete(f'{self.base_endpoint}/rack-reservations', id)
# ==================== Manufacturers ====================
async def list_manufacturers(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all manufacturers."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/manufacturers', params=params)
async def get_manufacturer(self, id: int) -> Dict:
"""Get a specific manufacturer by ID."""
return self.client.get(f'{self.base_endpoint}/manufacturers', id)
async def create_manufacturer(self, name: str, slug: str, **kwargs) -> Dict:
"""Create a new manufacturer."""
data = {'name': name, 'slug': slug, **kwargs}
return self.client.create(f'{self.base_endpoint}/manufacturers', data)
async def update_manufacturer(self, id: int, **kwargs) -> Dict:
"""Update a manufacturer."""
return self.client.patch(f'{self.base_endpoint}/manufacturers', id, kwargs)
async def delete_manufacturer(self, id: int) -> None:
"""Delete a manufacturer."""
self.client.delete(f'{self.base_endpoint}/manufacturers', id)
# ==================== Device Types ====================
async def list_device_types(
self,
manufacturer_id: Optional[int] = None,
model: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all device types."""
params = {k: v for k, v in {
'manufacturer_id': manufacturer_id, 'model': model, 'slug': slug, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/device-types', params=params)
async def get_device_type(self, id: int) -> Dict:
"""Get a specific device type by ID."""
return self.client.get(f'{self.base_endpoint}/device-types', id)
async def create_device_type(
self,
manufacturer: int,
model: str,
slug: str,
u_height: float = 1.0,
is_full_depth: bool = True,
**kwargs
) -> Dict:
"""Create a new device type."""
data = {
'manufacturer': manufacturer, 'model': model, 'slug': slug,
'u_height': u_height, 'is_full_depth': is_full_depth, **kwargs
}
return self.client.create(f'{self.base_endpoint}/device-types', data)
async def update_device_type(self, id: int, **kwargs) -> Dict:
"""Update a device type."""
return self.client.patch(f'{self.base_endpoint}/device-types', id, kwargs)
async def delete_device_type(self, id: int) -> None:
"""Delete a device type."""
self.client.delete(f'{self.base_endpoint}/device-types', id)
# ==================== Module Types ====================
async def list_module_types(self, manufacturer_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all module types."""
params = {k: v for k, v in {'manufacturer_id': manufacturer_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/module-types', params=params)
async def get_module_type(self, id: int) -> Dict:
"""Get a specific module type by ID."""
return self.client.get(f'{self.base_endpoint}/module-types', id)
async def create_module_type(self, manufacturer: int, model: str, **kwargs) -> Dict:
"""Create a new module type."""
data = {'manufacturer': manufacturer, 'model': model, **kwargs}
return self.client.create(f'{self.base_endpoint}/module-types', data)
async def update_module_type(self, id: int, **kwargs) -> Dict:
"""Update a module type."""
return self.client.patch(f'{self.base_endpoint}/module-types', id, kwargs)
async def delete_module_type(self, id: int) -> None:
"""Delete a module type."""
self.client.delete(f'{self.base_endpoint}/module-types', id)
# ==================== Device Roles ====================
async def list_device_roles(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all device roles."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/device-roles', params=params)
async def get_device_role(self, id: int) -> Dict:
"""Get a specific device role by ID."""
return self.client.get(f'{self.base_endpoint}/device-roles', id)
async def create_device_role(
self,
name: str,
slug: str,
color: str = '9e9e9e',
vm_role: bool = False,
**kwargs
) -> Dict:
"""Create a new device role."""
data = {'name': name, 'slug': slug, 'color': color, 'vm_role': vm_role, **kwargs}
return self.client.create(f'{self.base_endpoint}/device-roles', data)
async def update_device_role(self, id: int, **kwargs) -> Dict:
"""Update a device role."""
return self.client.patch(f'{self.base_endpoint}/device-roles', id, kwargs)
async def delete_device_role(self, id: int) -> None:
"""Delete a device role."""
self.client.delete(f'{self.base_endpoint}/device-roles', id)
# ==================== Platforms ====================
async def list_platforms(self, name: Optional[str] = None, manufacturer_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all platforms."""
params = {k: v for k, v in {'name': name, 'manufacturer_id': manufacturer_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/platforms', params=params)
async def get_platform(self, id: int) -> Dict:
"""Get a specific platform by ID."""
return self.client.get(f'{self.base_endpoint}/platforms', id)
async def create_platform(
self,
name: str,
slug: str,
manufacturer: Optional[int] = None,
**kwargs
) -> Dict:
"""Create a new platform."""
data = {'name': name, 'slug': slug, **kwargs}
if manufacturer:
data['manufacturer'] = manufacturer
return self.client.create(f'{self.base_endpoint}/platforms', data)
async def update_platform(self, id: int, **kwargs) -> Dict:
"""Update a platform."""
return self.client.patch(f'{self.base_endpoint}/platforms', id, kwargs)
async def delete_platform(self, id: int) -> None:
"""Delete a platform."""
self.client.delete(f'{self.base_endpoint}/platforms', id)
# ==================== Devices ====================
async def list_devices(
self,
name: Optional[str] = None,
site_id: Optional[int] = None,
location_id: Optional[int] = None,
rack_id: Optional[int] = None,
status: Optional[str] = None,
role_id: Optional[int] = None,
device_type_id: Optional[int] = None,
manufacturer_id: Optional[int] = None,
platform_id: Optional[int] = None,
tenant_id: Optional[int] = None,
serial: Optional[str] = None,
asset_tag: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all devices with optional filtering."""
params = {k: v for k, v in {
'name': name, 'site_id': site_id, 'location_id': location_id,
'rack_id': rack_id, 'status': status, 'role_id': role_id,
'device_type_id': device_type_id, 'manufacturer_id': manufacturer_id,
'platform_id': platform_id, 'tenant_id': tenant_id,
'serial': serial, 'asset_tag': asset_tag, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/devices', params=params)
async def get_device(self, id: int) -> Dict:
"""Get a specific device by ID."""
return self.client.get(f'{self.base_endpoint}/devices', id)
async def create_device(
self,
name: str,
device_type: int,
role: int,
site: int,
status: str = 'active',
location: Optional[int] = None,
rack: Optional[int] = None,
position: Optional[float] = None,
face: Optional[str] = None,
platform: Optional[int] = None,
tenant: Optional[int] = None,
serial: Optional[str] = None,
asset_tag: Optional[str] = None,
primary_ip4: Optional[int] = None,
primary_ip6: Optional[int] = None,
**kwargs
) -> Dict:
"""Create a new device."""
data = {
'name': name, 'device_type': device_type, 'role': role,
'site': site, 'status': status, **kwargs
}
for key, val in [
('location', location), ('rack', rack), ('position', position),
('face', face), ('platform', platform), ('tenant', tenant),
('serial', serial), ('asset_tag', asset_tag),
('primary_ip4', primary_ip4), ('primary_ip6', primary_ip6)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/devices', data)
async def update_device(self, id: int, **kwargs) -> Dict:
"""Update a device."""
return self.client.patch(f'{self.base_endpoint}/devices', id, kwargs)
async def delete_device(self, id: int) -> None:
"""Delete a device."""
self.client.delete(f'{self.base_endpoint}/devices', id)
# ==================== Modules ====================
async def list_modules(self, device_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all modules."""
params = {k: v for k, v in {'device_id': device_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/modules', params=params)
async def get_module(self, id: int) -> Dict:
"""Get a specific module by ID."""
return self.client.get(f'{self.base_endpoint}/modules', id)
async def create_module(self, device: int, module_bay: int, module_type: int, **kwargs) -> Dict:
"""Create a new module."""
data = {'device': device, 'module_bay': module_bay, 'module_type': module_type, **kwargs}
return self.client.create(f'{self.base_endpoint}/modules', data)
async def update_module(self, id: int, **kwargs) -> Dict:
"""Update a module."""
return self.client.patch(f'{self.base_endpoint}/modules', id, kwargs)
async def delete_module(self, id: int) -> None:
"""Delete a module."""
self.client.delete(f'{self.base_endpoint}/modules', id)
# ==================== Interfaces ====================
async def list_interfaces(
self,
device_id: Optional[int] = None,
name: Optional[str] = None,
type: Optional[str] = None,
enabled: Optional[bool] = None,
**kwargs
) -> List[Dict]:
"""List all interfaces."""
params = {k: v for k, v in {
'device_id': device_id, 'name': name, 'type': type, 'enabled': enabled, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/interfaces', params=params)
async def get_interface(self, id: int) -> Dict:
"""Get a specific interface by ID."""
return self.client.get(f'{self.base_endpoint}/interfaces', id)
async def create_interface(
self,
device: int,
name: str,
type: str,
enabled: bool = True,
mtu: Optional[int] = None,
mac_address: Optional[str] = None,
description: Optional[str] = None,
mode: Optional[str] = None,
untagged_vlan: Optional[int] = None,
tagged_vlans: Optional[List[int]] = None,
**kwargs
) -> Dict:
"""Create a new interface."""
data = {'device': device, 'name': name, 'type': type, 'enabled': enabled, **kwargs}
for key, val in [
('mtu', mtu), ('mac_address', mac_address), ('description', description),
('mode', mode), ('untagged_vlan', untagged_vlan), ('tagged_vlans', tagged_vlans)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/interfaces', data)
async def update_interface(self, id: int, **kwargs) -> Dict:
"""Update an interface."""
return self.client.patch(f'{self.base_endpoint}/interfaces', id, kwargs)
async def delete_interface(self, id: int) -> None:
"""Delete an interface."""
self.client.delete(f'{self.base_endpoint}/interfaces', id)
# ==================== Console Ports ====================
async def list_console_ports(self, device_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all console ports."""
params = {k: v for k, v in {'device_id': device_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/console-ports', params=params)
async def get_console_port(self, id: int) -> Dict:
"""Get a specific console port by ID."""
return self.client.get(f'{self.base_endpoint}/console-ports', id)
async def create_console_port(self, device: int, name: str, **kwargs) -> Dict:
"""Create a new console port."""
data = {'device': device, 'name': name, **kwargs}
return self.client.create(f'{self.base_endpoint}/console-ports', data)
async def update_console_port(self, id: int, **kwargs) -> Dict:
"""Update a console port."""
return self.client.patch(f'{self.base_endpoint}/console-ports', id, kwargs)
async def delete_console_port(self, id: int) -> None:
"""Delete a console port."""
self.client.delete(f'{self.base_endpoint}/console-ports', id)
# ==================== Console Server Ports ====================
async def list_console_server_ports(self, device_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all console server ports."""
params = {k: v for k, v in {'device_id': device_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/console-server-ports', params=params)
async def get_console_server_port(self, id: int) -> Dict:
"""Get a specific console server port by ID."""
return self.client.get(f'{self.base_endpoint}/console-server-ports', id)
async def create_console_server_port(self, device: int, name: str, **kwargs) -> Dict:
"""Create a new console server port."""
data = {'device': device, 'name': name, **kwargs}
return self.client.create(f'{self.base_endpoint}/console-server-ports', data)
async def update_console_server_port(self, id: int, **kwargs) -> Dict:
"""Update a console server port."""
return self.client.patch(f'{self.base_endpoint}/console-server-ports', id, kwargs)
async def delete_console_server_port(self, id: int) -> None:
"""Delete a console server port."""
self.client.delete(f'{self.base_endpoint}/console-server-ports', id)
# ==================== Power Ports ====================
async def list_power_ports(self, device_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all power ports."""
params = {k: v for k, v in {'device_id': device_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/power-ports', params=params)
async def get_power_port(self, id: int) -> Dict:
"""Get a specific power port by ID."""
return self.client.get(f'{self.base_endpoint}/power-ports', id)
async def create_power_port(self, device: int, name: str, **kwargs) -> Dict:
"""Create a new power port."""
data = {'device': device, 'name': name, **kwargs}
return self.client.create(f'{self.base_endpoint}/power-ports', data)
async def update_power_port(self, id: int, **kwargs) -> Dict:
"""Update a power port."""
return self.client.patch(f'{self.base_endpoint}/power-ports', id, kwargs)
async def delete_power_port(self, id: int) -> None:
"""Delete a power port."""
self.client.delete(f'{self.base_endpoint}/power-ports', id)
# ==================== Power Outlets ====================
async def list_power_outlets(self, device_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all power outlets."""
params = {k: v for k, v in {'device_id': device_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/power-outlets', params=params)
async def get_power_outlet(self, id: int) -> Dict:
"""Get a specific power outlet by ID."""
return self.client.get(f'{self.base_endpoint}/power-outlets', id)
async def create_power_outlet(self, device: int, name: str, **kwargs) -> Dict:
"""Create a new power outlet."""
data = {'device': device, 'name': name, **kwargs}
return self.client.create(f'{self.base_endpoint}/power-outlets', data)
async def update_power_outlet(self, id: int, **kwargs) -> Dict:
"""Update a power outlet."""
return self.client.patch(f'{self.base_endpoint}/power-outlets', id, kwargs)
async def delete_power_outlet(self, id: int) -> None:
"""Delete a power outlet."""
self.client.delete(f'{self.base_endpoint}/power-outlets', id)
# ==================== Power Panels ====================
async def list_power_panels(self, site_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all power panels."""
params = {k: v for k, v in {'site_id': site_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/power-panels', params=params)
async def get_power_panel(self, id: int) -> Dict:
"""Get a specific power panel by ID."""
return self.client.get(f'{self.base_endpoint}/power-panels', id)
async def create_power_panel(self, site: int, name: str, location: Optional[int] = None, **kwargs) -> Dict:
"""Create a new power panel."""
data = {'site': site, 'name': name, **kwargs}
if location:
data['location'] = location
return self.client.create(f'{self.base_endpoint}/power-panels', data)
async def update_power_panel(self, id: int, **kwargs) -> Dict:
"""Update a power panel."""
return self.client.patch(f'{self.base_endpoint}/power-panels', id, kwargs)
async def delete_power_panel(self, id: int) -> None:
"""Delete a power panel."""
self.client.delete(f'{self.base_endpoint}/power-panels', id)
# ==================== Power Feeds ====================
async def list_power_feeds(self, power_panel_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all power feeds."""
params = {k: v for k, v in {'power_panel_id': power_panel_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/power-feeds', params=params)
async def get_power_feed(self, id: int) -> Dict:
"""Get a specific power feed by ID."""
return self.client.get(f'{self.base_endpoint}/power-feeds', id)
async def create_power_feed(
self,
power_panel: int,
name: str,
status: str = 'active',
type: str = 'primary',
supply: str = 'ac',
phase: str = 'single-phase',
voltage: int = 120,
amperage: int = 20,
**kwargs
) -> Dict:
"""Create a new power feed."""
data = {
'power_panel': power_panel, 'name': name, 'status': status,
'type': type, 'supply': supply, 'phase': phase,
'voltage': voltage, 'amperage': amperage, **kwargs
}
return self.client.create(f'{self.base_endpoint}/power-feeds', data)
async def update_power_feed(self, id: int, **kwargs) -> Dict:
"""Update a power feed."""
return self.client.patch(f'{self.base_endpoint}/power-feeds', id, kwargs)
async def delete_power_feed(self, id: int) -> None:
"""Delete a power feed."""
self.client.delete(f'{self.base_endpoint}/power-feeds', id)
# ==================== Cables ====================
async def list_cables(
self,
site_id: Optional[int] = None,
device_id: Optional[int] = None,
rack_id: Optional[int] = None,
type: Optional[str] = None,
status: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all cables."""
params = {k: v for k, v in {
'site_id': site_id, 'device_id': device_id, 'rack_id': rack_id,
'type': type, 'status': status, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/cables', params=params)
async def get_cable(self, id: int) -> Dict:
"""Get a specific cable by ID."""
return self.client.get(f'{self.base_endpoint}/cables', id)
async def create_cable(
self,
a_terminations: List[Dict],
b_terminations: List[Dict],
type: Optional[str] = None,
status: str = 'connected',
label: Optional[str] = None,
color: Optional[str] = None,
length: Optional[float] = None,
length_unit: Optional[str] = None,
**kwargs
) -> Dict:
"""
Create a new cable.
a_terminations and b_terminations are lists of dicts with:
- object_type: e.g., 'dcim.interface'
- object_id: ID of the object
"""
data = {
'a_terminations': a_terminations,
'b_terminations': b_terminations,
'status': status,
**kwargs
}
for key, val in [
('type', type), ('label', label), ('color', color),
('length', length), ('length_unit', length_unit)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/cables', data)
async def update_cable(self, id: int, **kwargs) -> Dict:
"""Update a cable."""
return self.client.patch(f'{self.base_endpoint}/cables', id, kwargs)
async def delete_cable(self, id: int) -> None:
"""Delete a cable."""
self.client.delete(f'{self.base_endpoint}/cables', id)
# ==================== Virtual Chassis ====================
async def list_virtual_chassis(self, **kwargs) -> List[Dict]:
"""List all virtual chassis."""
return self.client.list(f'{self.base_endpoint}/virtual-chassis', params=kwargs)
async def get_virtual_chassis(self, id: int) -> Dict:
"""Get a specific virtual chassis by ID."""
return self.client.get(f'{self.base_endpoint}/virtual-chassis', id)
async def create_virtual_chassis(self, name: str, domain: Optional[str] = None, **kwargs) -> Dict:
"""Create a new virtual chassis."""
data = {'name': name, **kwargs}
if domain:
data['domain'] = domain
return self.client.create(f'{self.base_endpoint}/virtual-chassis', data)
async def update_virtual_chassis(self, id: int, **kwargs) -> Dict:
"""Update a virtual chassis."""
return self.client.patch(f'{self.base_endpoint}/virtual-chassis', id, kwargs)
async def delete_virtual_chassis(self, id: int) -> None:
"""Delete a virtual chassis."""
self.client.delete(f'{self.base_endpoint}/virtual-chassis', id)
# ==================== Inventory Items ====================
async def list_inventory_items(self, device_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all inventory items."""
params = {k: v for k, v in {'device_id': device_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/inventory-items', params=params)
async def get_inventory_item(self, id: int) -> Dict:
"""Get a specific inventory item by ID."""
return self.client.get(f'{self.base_endpoint}/inventory-items', id)
async def create_inventory_item(
self,
device: int,
name: str,
parent: Optional[int] = None,
manufacturer: Optional[int] = None,
part_id: Optional[str] = None,
serial: Optional[str] = None,
asset_tag: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new inventory item."""
data = {'device': device, 'name': name, **kwargs}
for key, val in [
('parent', parent), ('manufacturer', manufacturer),
('part_id', part_id), ('serial', serial), ('asset_tag', asset_tag)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/inventory-items', data)
async def update_inventory_item(self, id: int, **kwargs) -> Dict:
"""Update an inventory item."""
return self.client.patch(f'{self.base_endpoint}/inventory-items', id, kwargs)
async def delete_inventory_item(self, id: int) -> None:
"""Delete an inventory item."""
self.client.delete(f'{self.base_endpoint}/inventory-items', id)

View File

@@ -0,0 +1,560 @@
"""
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}', {})

View File

@@ -0,0 +1,718 @@
"""
IPAM (IP Address Management) tools for NetBox MCP Server.
Covers: IP Addresses, Prefixes, VLANs, VRFs, ASNs, and related models.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class IPAMTools:
"""Tools for IPAM operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'ipam'
# ==================== ASN Ranges ====================
async def list_asn_ranges(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all ASN ranges."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/asn-ranges', params=params)
async def get_asn_range(self, id: int) -> Dict:
"""Get a specific ASN range by ID."""
return self.client.get(f'{self.base_endpoint}/asn-ranges', id)
async def create_asn_range(self, name: str, slug: str, rir: int, start: int, end: int, **kwargs) -> Dict:
"""Create a new ASN range."""
data = {'name': name, 'slug': slug, 'rir': rir, 'start': start, 'end': end, **kwargs}
return self.client.create(f'{self.base_endpoint}/asn-ranges', data)
async def update_asn_range(self, id: int, **kwargs) -> Dict:
"""Update an ASN range."""
return self.client.patch(f'{self.base_endpoint}/asn-ranges', id, kwargs)
async def delete_asn_range(self, id: int) -> None:
"""Delete an ASN range."""
self.client.delete(f'{self.base_endpoint}/asn-ranges', id)
# ==================== ASNs ====================
async def list_asns(
self,
asn: Optional[int] = None,
rir_id: Optional[int] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all ASNs."""
params = {k: v for k, v in {
'asn': asn, 'rir_id': rir_id, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/asns', params=params)
async def get_asn(self, id: int) -> Dict:
"""Get a specific ASN by ID."""
return self.client.get(f'{self.base_endpoint}/asns', id)
async def create_asn(
self,
asn: int,
rir: int,
tenant: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new ASN."""
data = {'asn': asn, 'rir': rir, **kwargs}
if tenant:
data['tenant'] = tenant
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/asns', data)
async def update_asn(self, id: int, **kwargs) -> Dict:
"""Update an ASN."""
return self.client.patch(f'{self.base_endpoint}/asns', id, kwargs)
async def delete_asn(self, id: int) -> None:
"""Delete an ASN."""
self.client.delete(f'{self.base_endpoint}/asns', id)
# ==================== RIRs ====================
async def list_rirs(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all RIRs (Regional Internet Registries)."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/rirs', params=params)
async def get_rir(self, id: int) -> Dict:
"""Get a specific RIR by ID."""
return self.client.get(f'{self.base_endpoint}/rirs', id)
async def create_rir(self, name: str, slug: str, is_private: bool = False, **kwargs) -> Dict:
"""Create a new RIR."""
data = {'name': name, 'slug': slug, 'is_private': is_private, **kwargs}
return self.client.create(f'{self.base_endpoint}/rirs', data)
async def update_rir(self, id: int, **kwargs) -> Dict:
"""Update a RIR."""
return self.client.patch(f'{self.base_endpoint}/rirs', id, kwargs)
async def delete_rir(self, id: int) -> None:
"""Delete a RIR."""
self.client.delete(f'{self.base_endpoint}/rirs', id)
# ==================== Aggregates ====================
async def list_aggregates(
self,
prefix: Optional[str] = None,
rir_id: Optional[int] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all aggregates."""
params = {k: v for k, v in {
'prefix': prefix, 'rir_id': rir_id, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/aggregates', params=params)
async def get_aggregate(self, id: int) -> Dict:
"""Get a specific aggregate by ID."""
return self.client.get(f'{self.base_endpoint}/aggregates', id)
async def create_aggregate(
self,
prefix: str,
rir: int,
tenant: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new aggregate."""
data = {'prefix': prefix, 'rir': rir, **kwargs}
if tenant:
data['tenant'] = tenant
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/aggregates', data)
async def update_aggregate(self, id: int, **kwargs) -> Dict:
"""Update an aggregate."""
return self.client.patch(f'{self.base_endpoint}/aggregates', id, kwargs)
async def delete_aggregate(self, id: int) -> None:
"""Delete an aggregate."""
self.client.delete(f'{self.base_endpoint}/aggregates', id)
# ==================== Roles ====================
async def list_roles(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all IPAM roles."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/roles', params=params)
async def get_role(self, id: int) -> Dict:
"""Get a specific role by ID."""
return self.client.get(f'{self.base_endpoint}/roles', id)
async def create_role(self, name: str, slug: str, weight: int = 1000, **kwargs) -> Dict:
"""Create a new IPAM role."""
data = {'name': name, 'slug': slug, 'weight': weight, **kwargs}
return self.client.create(f'{self.base_endpoint}/roles', data)
async def update_role(self, id: int, **kwargs) -> Dict:
"""Update a role."""
return self.client.patch(f'{self.base_endpoint}/roles', id, kwargs)
async def delete_role(self, id: int) -> None:
"""Delete a role."""
self.client.delete(f'{self.base_endpoint}/roles', id)
# ==================== Prefixes ====================
async def list_prefixes(
self,
prefix: Optional[str] = None,
site_id: Optional[int] = None,
vrf_id: Optional[int] = None,
vlan_id: Optional[int] = None,
role_id: Optional[int] = None,
tenant_id: Optional[int] = None,
status: Optional[str] = None,
family: Optional[int] = None,
is_pool: Optional[bool] = None,
within: Optional[str] = None,
within_include: Optional[str] = None,
contains: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all prefixes with optional filtering."""
params = {k: v for k, v in {
'prefix': prefix, 'site_id': site_id, 'vrf_id': vrf_id,
'vlan_id': vlan_id, 'role_id': role_id, 'tenant_id': tenant_id,
'status': status, 'family': family, 'is_pool': is_pool,
'within': within, 'within_include': within_include, 'contains': contains, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/prefixes', params=params)
async def get_prefix(self, id: int) -> Dict:
"""Get a specific prefix by ID."""
return self.client.get(f'{self.base_endpoint}/prefixes', id)
async def create_prefix(
self,
prefix: str,
status: str = 'active',
site: Optional[int] = None,
vrf: Optional[int] = None,
vlan: Optional[int] = None,
role: Optional[int] = None,
tenant: Optional[int] = None,
is_pool: bool = False,
mark_utilized: bool = False,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new prefix."""
data = {'prefix': prefix, 'status': status, 'is_pool': is_pool, 'mark_utilized': mark_utilized, **kwargs}
for key, val in [
('site', site), ('vrf', vrf), ('vlan', vlan),
('role', role), ('tenant', tenant), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/prefixes', data)
async def update_prefix(self, id: int, **kwargs) -> Dict:
"""Update a prefix."""
return self.client.patch(f'{self.base_endpoint}/prefixes', id, kwargs)
async def delete_prefix(self, id: int) -> None:
"""Delete a prefix."""
self.client.delete(f'{self.base_endpoint}/prefixes', id)
async def list_available_prefixes(self, id: int) -> List[Dict]:
"""List available child prefixes within a prefix."""
return self.client.list(f'{self.base_endpoint}/prefixes/{id}/available-prefixes', paginate=False)
async def create_available_prefix(self, id: int, prefix_length: int, **kwargs) -> Dict:
"""Create a new prefix from available space."""
data = {'prefix_length': prefix_length, **kwargs}
return self.client.create(f'{self.base_endpoint}/prefixes/{id}/available-prefixes', data)
async def list_available_ips(self, id: int) -> List[Dict]:
"""List available IP addresses within a prefix."""
return self.client.list(f'{self.base_endpoint}/prefixes/{id}/available-ips', paginate=False)
async def create_available_ip(self, id: int, **kwargs) -> Dict:
"""Create a new IP address from available space in prefix."""
return self.client.create(f'{self.base_endpoint}/prefixes/{id}/available-ips', kwargs)
# ==================== IP Ranges ====================
async def list_ip_ranges(
self,
start_address: Optional[str] = None,
end_address: Optional[str] = None,
vrf_id: Optional[int] = None,
tenant_id: Optional[int] = None,
status: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all IP ranges."""
params = {k: v for k, v in {
'start_address': start_address, 'end_address': end_address,
'vrf_id': vrf_id, 'tenant_id': tenant_id, 'status': status, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ip-ranges', params=params)
async def get_ip_range(self, id: int) -> Dict:
"""Get a specific IP range by ID."""
return self.client.get(f'{self.base_endpoint}/ip-ranges', id)
async def create_ip_range(
self,
start_address: str,
end_address: str,
status: str = 'active',
vrf: Optional[int] = None,
tenant: Optional[int] = None,
role: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IP range."""
data = {'start_address': start_address, 'end_address': end_address, 'status': status, **kwargs}
for key, val in [('vrf', vrf), ('tenant', tenant), ('role', role), ('description', description)]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/ip-ranges', data)
async def update_ip_range(self, id: int, **kwargs) -> Dict:
"""Update an IP range."""
return self.client.patch(f'{self.base_endpoint}/ip-ranges', id, kwargs)
async def delete_ip_range(self, id: int) -> None:
"""Delete an IP range."""
self.client.delete(f'{self.base_endpoint}/ip-ranges', id)
async def list_available_ips_in_range(self, id: int) -> List[Dict]:
"""List available IP addresses within an IP range."""
return self.client.list(f'{self.base_endpoint}/ip-ranges/{id}/available-ips', paginate=False)
# ==================== IP Addresses ====================
async def list_ip_addresses(
self,
address: Optional[str] = None,
vrf_id: Optional[int] = None,
tenant_id: Optional[int] = None,
status: Optional[str] = None,
role: Optional[str] = None,
interface_id: Optional[int] = None,
device_id: Optional[int] = None,
virtual_machine_id: Optional[int] = None,
family: Optional[int] = None,
parent: Optional[str] = None,
dns_name: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all IP addresses with optional filtering."""
params = {k: v for k, v in {
'address': address, 'vrf_id': vrf_id, 'tenant_id': tenant_id,
'status': status, 'role': role, 'interface_id': interface_id,
'device_id': device_id, 'virtual_machine_id': virtual_machine_id,
'family': family, 'parent': parent, 'dns_name': dns_name, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ip-addresses', params=params)
async def get_ip_address(self, id: int) -> Dict:
"""Get a specific IP address by ID."""
return self.client.get(f'{self.base_endpoint}/ip-addresses', id)
async def create_ip_address(
self,
address: str,
status: str = 'active',
vrf: Optional[int] = None,
tenant: Optional[int] = None,
role: Optional[str] = None,
assigned_object_type: Optional[str] = None,
assigned_object_id: Optional[int] = None,
nat_inside: Optional[int] = None,
dns_name: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IP address."""
data = {'address': address, 'status': status, **kwargs}
for key, val in [
('vrf', vrf), ('tenant', tenant), ('role', role),
('assigned_object_type', assigned_object_type),
('assigned_object_id', assigned_object_id),
('nat_inside', nat_inside), ('dns_name', dns_name),
('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/ip-addresses', data)
async def update_ip_address(self, id: int, **kwargs) -> Dict:
"""Update an IP address."""
return self.client.patch(f'{self.base_endpoint}/ip-addresses', id, kwargs)
async def delete_ip_address(self, id: int) -> None:
"""Delete an IP address."""
self.client.delete(f'{self.base_endpoint}/ip-addresses', id)
# ==================== FHRP Groups ====================
async def list_fhrp_groups(
self,
protocol: Optional[str] = None,
group_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all FHRP groups."""
params = {k: v for k, v in {'protocol': protocol, 'group_id': group_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/fhrp-groups', params=params)
async def get_fhrp_group(self, id: int) -> Dict:
"""Get a specific FHRP group by ID."""
return self.client.get(f'{self.base_endpoint}/fhrp-groups', id)
async def create_fhrp_group(
self,
protocol: str,
group_id: int,
auth_type: Optional[str] = None,
auth_key: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new FHRP group."""
data = {'protocol': protocol, 'group_id': group_id, **kwargs}
if auth_type:
data['auth_type'] = auth_type
if auth_key:
data['auth_key'] = auth_key
return self.client.create(f'{self.base_endpoint}/fhrp-groups', data)
async def update_fhrp_group(self, id: int, **kwargs) -> Dict:
"""Update an FHRP group."""
return self.client.patch(f'{self.base_endpoint}/fhrp-groups', id, kwargs)
async def delete_fhrp_group(self, id: int) -> None:
"""Delete an FHRP group."""
self.client.delete(f'{self.base_endpoint}/fhrp-groups', id)
# ==================== FHRP Group Assignments ====================
async def list_fhrp_group_assignments(self, group_id: Optional[int] = None, **kwargs) -> List[Dict]:
"""List all FHRP group assignments."""
params = {k: v for k, v in {'group_id': group_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/fhrp-group-assignments', params=params)
async def get_fhrp_group_assignment(self, id: int) -> Dict:
"""Get a specific FHRP group assignment by ID."""
return self.client.get(f'{self.base_endpoint}/fhrp-group-assignments', id)
async def create_fhrp_group_assignment(
self,
group: int,
interface_type: str,
interface_id: int,
priority: int = 100,
**kwargs
) -> Dict:
"""Create a new FHRP group assignment."""
data = {
'group': group, 'interface_type': interface_type,
'interface_id': interface_id, 'priority': priority, **kwargs
}
return self.client.create(f'{self.base_endpoint}/fhrp-group-assignments', data)
async def update_fhrp_group_assignment(self, id: int, **kwargs) -> Dict:
"""Update an FHRP group assignment."""
return self.client.patch(f'{self.base_endpoint}/fhrp-group-assignments', id, kwargs)
async def delete_fhrp_group_assignment(self, id: int) -> None:
"""Delete an FHRP group assignment."""
self.client.delete(f'{self.base_endpoint}/fhrp-group-assignments', id)
# ==================== VLAN Groups ====================
async def list_vlan_groups(
self,
name: Optional[str] = None,
site_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all VLAN groups."""
params = {k: v for k, v in {'name': name, 'site_id': site_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/vlan-groups', params=params)
async def get_vlan_group(self, id: int) -> Dict:
"""Get a specific VLAN group by ID."""
return self.client.get(f'{self.base_endpoint}/vlan-groups', id)
async def create_vlan_group(
self,
name: str,
slug: str,
scope_type: Optional[str] = None,
scope_id: Optional[int] = None,
min_vid: int = 1,
max_vid: int = 4094,
**kwargs
) -> Dict:
"""Create a new VLAN group."""
data = {'name': name, 'slug': slug, 'min_vid': min_vid, 'max_vid': max_vid, **kwargs}
if scope_type:
data['scope_type'] = scope_type
if scope_id:
data['scope_id'] = scope_id
return self.client.create(f'{self.base_endpoint}/vlan-groups', data)
async def update_vlan_group(self, id: int, **kwargs) -> Dict:
"""Update a VLAN group."""
return self.client.patch(f'{self.base_endpoint}/vlan-groups', id, kwargs)
async def delete_vlan_group(self, id: int) -> None:
"""Delete a VLAN group."""
self.client.delete(f'{self.base_endpoint}/vlan-groups', id)
async def list_available_vlans(self, id: int) -> List[Dict]:
"""List available VLANs in a VLAN group."""
return self.client.list(f'{self.base_endpoint}/vlan-groups/{id}/available-vlans', paginate=False)
# ==================== VLANs ====================
async def list_vlans(
self,
vid: Optional[int] = None,
name: Optional[str] = None,
site_id: Optional[int] = None,
group_id: Optional[int] = None,
role_id: Optional[int] = None,
tenant_id: Optional[int] = None,
status: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all VLANs with optional filtering."""
params = {k: v for k, v in {
'vid': vid, 'name': name, 'site_id': site_id, 'group_id': group_id,
'role_id': role_id, 'tenant_id': tenant_id, 'status': status, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/vlans', params=params)
async def get_vlan(self, id: int) -> Dict:
"""Get a specific VLAN by ID."""
return self.client.get(f'{self.base_endpoint}/vlans', id)
async def create_vlan(
self,
vid: int,
name: str,
status: str = 'active',
site: Optional[int] = None,
group: Optional[int] = None,
role: Optional[int] = None,
tenant: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new VLAN."""
data = {'vid': vid, 'name': name, 'status': status, **kwargs}
for key, val in [
('site', site), ('group', group), ('role', role),
('tenant', tenant), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/vlans', data)
async def update_vlan(self, id: int, **kwargs) -> Dict:
"""Update a VLAN."""
return self.client.patch(f'{self.base_endpoint}/vlans', id, kwargs)
async def delete_vlan(self, id: int) -> None:
"""Delete a VLAN."""
self.client.delete(f'{self.base_endpoint}/vlans', id)
# ==================== VRFs ====================
async def list_vrfs(
self,
name: Optional[str] = None,
rd: Optional[str] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all VRFs with optional filtering."""
params = {k: v for k, v in {
'name': name, 'rd': rd, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/vrfs', params=params)
async def get_vrf(self, id: int) -> Dict:
"""Get a specific VRF by ID."""
return self.client.get(f'{self.base_endpoint}/vrfs', id)
async def create_vrf(
self,
name: str,
rd: Optional[str] = None,
tenant: Optional[int] = None,
enforce_unique: bool = True,
description: Optional[str] = None,
import_targets: Optional[List[int]] = None,
export_targets: Optional[List[int]] = None,
**kwargs
) -> Dict:
"""Create a new VRF."""
data = {'name': name, 'enforce_unique': enforce_unique, **kwargs}
for key, val in [
('rd', rd), ('tenant', tenant), ('description', description),
('import_targets', import_targets), ('export_targets', export_targets)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/vrfs', data)
async def update_vrf(self, id: int, **kwargs) -> Dict:
"""Update a VRF."""
return self.client.patch(f'{self.base_endpoint}/vrfs', id, kwargs)
async def delete_vrf(self, id: int) -> None:
"""Delete a VRF."""
self.client.delete(f'{self.base_endpoint}/vrfs', id)
# ==================== Route Targets ====================
async def list_route_targets(
self,
name: Optional[str] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all route targets."""
params = {k: v for k, v in {'name': name, 'tenant_id': tenant_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/route-targets', params=params)
async def get_route_target(self, id: int) -> Dict:
"""Get a specific route target by ID."""
return self.client.get(f'{self.base_endpoint}/route-targets', id)
async def create_route_target(
self,
name: str,
tenant: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new route target."""
data = {'name': name, **kwargs}
if tenant:
data['tenant'] = tenant
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/route-targets', data)
async def update_route_target(self, id: int, **kwargs) -> Dict:
"""Update a route target."""
return self.client.patch(f'{self.base_endpoint}/route-targets', id, kwargs)
async def delete_route_target(self, id: int) -> None:
"""Delete a route target."""
self.client.delete(f'{self.base_endpoint}/route-targets', id)
# ==================== Services ====================
async def list_services(
self,
device_id: Optional[int] = None,
virtual_machine_id: Optional[int] = None,
name: Optional[str] = None,
protocol: Optional[str] = None,
port: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all services."""
params = {k: v for k, v in {
'device_id': device_id, 'virtual_machine_id': virtual_machine_id,
'name': name, 'protocol': protocol, 'port': port, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/services', params=params)
async def get_service(self, id: int) -> Dict:
"""Get a specific service by ID."""
return self.client.get(f'{self.base_endpoint}/services', id)
async def create_service(
self,
name: str,
ports: List[int],
protocol: str,
device: Optional[int] = None,
virtual_machine: Optional[int] = None,
ipaddresses: Optional[List[int]] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new service."""
data = {'name': name, 'ports': ports, 'protocol': protocol, **kwargs}
for key, val in [
('device', device), ('virtual_machine', virtual_machine),
('ipaddresses', ipaddresses), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/services', data)
async def update_service(self, id: int, **kwargs) -> Dict:
"""Update a service."""
return self.client.patch(f'{self.base_endpoint}/services', id, kwargs)
async def delete_service(self, id: int) -> None:
"""Delete a service."""
self.client.delete(f'{self.base_endpoint}/services', id)
# ==================== Service Templates ====================
async def list_service_templates(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all service 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}/service-templates', params=params)
async def get_service_template(self, id: int) -> Dict:
"""Get a specific service template by ID."""
return self.client.get(f'{self.base_endpoint}/service-templates', id)
async def create_service_template(
self,
name: str,
ports: List[int],
protocol: str,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new service template."""
data = {'name': name, 'ports': ports, 'protocol': protocol, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/service-templates', data)
async def update_service_template(self, id: int, **kwargs) -> Dict:
"""Update a service template."""
return self.client.patch(f'{self.base_endpoint}/service-templates', id, kwargs)
async def delete_service_template(self, id: int) -> None:
"""Delete a service template."""
self.client.delete(f'{self.base_endpoint}/service-templates', id)

View File

@@ -0,0 +1,281 @@
"""
Tenancy tools for NetBox MCP Server.
Covers: Tenants, Tenant Groups, Contacts, Contact Groups, and Contact Roles.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class TenancyTools:
"""Tools for Tenancy operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'tenancy'
# ==================== Tenant Groups ====================
async def list_tenant_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
parent_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all tenant groups."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'parent_id': parent_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/tenant-groups', params=params)
async def get_tenant_group(self, id: int) -> Dict:
"""Get a specific tenant group by ID."""
return self.client.get(f'{self.base_endpoint}/tenant-groups', id)
async def create_tenant_group(
self,
name: str,
slug: str,
parent: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new tenant group."""
data = {'name': name, 'slug': slug, **kwargs}
if parent:
data['parent'] = parent
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/tenant-groups', data)
async def update_tenant_group(self, id: int, **kwargs) -> Dict:
"""Update a tenant group."""
return self.client.patch(f'{self.base_endpoint}/tenant-groups', id, kwargs)
async def delete_tenant_group(self, id: int) -> None:
"""Delete a tenant group."""
self.client.delete(f'{self.base_endpoint}/tenant-groups', id)
# ==================== Tenants ====================
async def list_tenants(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
group_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all tenants with optional filtering."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'group_id': group_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/tenants', params=params)
async def get_tenant(self, id: int) -> Dict:
"""Get a specific tenant by ID."""
return self.client.get(f'{self.base_endpoint}/tenants', id)
async def create_tenant(
self,
name: str,
slug: str,
group: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new tenant."""
data = {'name': name, 'slug': slug, **kwargs}
if group:
data['group'] = group
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/tenants', data)
async def update_tenant(self, id: int, **kwargs) -> Dict:
"""Update a tenant."""
return self.client.patch(f'{self.base_endpoint}/tenants', id, kwargs)
async def delete_tenant(self, id: int) -> None:
"""Delete a tenant."""
self.client.delete(f'{self.base_endpoint}/tenants', id)
# ==================== Contact Groups ====================
async def list_contact_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
parent_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all contact groups."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'parent_id': parent_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/contact-groups', params=params)
async def get_contact_group(self, id: int) -> Dict:
"""Get a specific contact group by ID."""
return self.client.get(f'{self.base_endpoint}/contact-groups', id)
async def create_contact_group(
self,
name: str,
slug: str,
parent: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new contact group."""
data = {'name': name, 'slug': slug, **kwargs}
if parent:
data['parent'] = parent
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/contact-groups', data)
async def update_contact_group(self, id: int, **kwargs) -> Dict:
"""Update a contact group."""
return self.client.patch(f'{self.base_endpoint}/contact-groups', id, kwargs)
async def delete_contact_group(self, id: int) -> None:
"""Delete a contact group."""
self.client.delete(f'{self.base_endpoint}/contact-groups', id)
# ==================== Contact Roles ====================
async def list_contact_roles(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all contact roles."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/contact-roles', params=params)
async def get_contact_role(self, id: int) -> Dict:
"""Get a specific contact role by ID."""
return self.client.get(f'{self.base_endpoint}/contact-roles', id)
async def create_contact_role(
self,
name: str,
slug: str,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new contact role."""
data = {'name': name, 'slug': slug, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/contact-roles', data)
async def update_contact_role(self, id: int, **kwargs) -> Dict:
"""Update a contact role."""
return self.client.patch(f'{self.base_endpoint}/contact-roles', id, kwargs)
async def delete_contact_role(self, id: int) -> None:
"""Delete a contact role."""
self.client.delete(f'{self.base_endpoint}/contact-roles', id)
# ==================== Contacts ====================
async def list_contacts(
self,
name: Optional[str] = None,
group_id: Optional[int] = None,
email: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all contacts with optional filtering."""
params = {k: v for k, v in {
'name': name, 'group_id': group_id, 'email': email, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/contacts', params=params)
async def get_contact(self, id: int) -> Dict:
"""Get a specific contact by ID."""
return self.client.get(f'{self.base_endpoint}/contacts', id)
async def create_contact(
self,
name: str,
group: Optional[int] = None,
title: Optional[str] = None,
phone: Optional[str] = None,
email: Optional[str] = None,
address: Optional[str] = None,
link: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new contact."""
data = {'name': name, **kwargs}
for key, val in [
('group', group), ('title', title), ('phone', phone),
('email', email), ('address', address), ('link', link),
('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/contacts', data)
async def update_contact(self, id: int, **kwargs) -> Dict:
"""Update a contact."""
return self.client.patch(f'{self.base_endpoint}/contacts', id, kwargs)
async def delete_contact(self, id: int) -> None:
"""Delete a contact."""
self.client.delete(f'{self.base_endpoint}/contacts', id)
# ==================== Contact Assignments ====================
async def list_contact_assignments(
self,
contact_id: Optional[int] = None,
role_id: Optional[int] = None,
object_type: Optional[str] = None,
object_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all contact assignments."""
params = {k: v for k, v in {
'contact_id': contact_id, 'role_id': role_id,
'object_type': object_type, 'object_id': object_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/contact-assignments', params=params)
async def get_contact_assignment(self, id: int) -> Dict:
"""Get a specific contact assignment by ID."""
return self.client.get(f'{self.base_endpoint}/contact-assignments', id)
async def create_contact_assignment(
self,
contact: int,
role: int,
object_type: str,
object_id: int,
priority: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new contact assignment."""
data = {
'contact': contact, 'role': role,
'object_type': object_type, 'object_id': object_id, **kwargs
}
if priority:
data['priority'] = priority
return self.client.create(f'{self.base_endpoint}/contact-assignments', data)
async def update_contact_assignment(self, id: int, **kwargs) -> Dict:
"""Update a contact assignment."""
return self.client.patch(f'{self.base_endpoint}/contact-assignments', id, kwargs)
async def delete_contact_assignment(self, id: int) -> None:
"""Delete a contact assignment."""
self.client.delete(f'{self.base_endpoint}/contact-assignments', id)

View File

@@ -0,0 +1,296 @@
"""
Virtualization tools for NetBox MCP Server.
Covers: Clusters, Virtual Machines, VM Interfaces, and related models.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class VirtualizationTools:
"""Tools for Virtualization operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'virtualization'
# ==================== Cluster Types ====================
async def list_cluster_types(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all cluster types."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/cluster-types', params=params)
async def get_cluster_type(self, id: int) -> Dict:
"""Get a specific cluster type by ID."""
return self.client.get(f'{self.base_endpoint}/cluster-types', id)
async def create_cluster_type(
self,
name: str,
slug: str,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new cluster type."""
data = {'name': name, 'slug': slug, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/cluster-types', data)
async def update_cluster_type(self, id: int, **kwargs) -> Dict:
"""Update a cluster type."""
return self.client.patch(f'{self.base_endpoint}/cluster-types', id, kwargs)
async def delete_cluster_type(self, id: int) -> None:
"""Delete a cluster type."""
self.client.delete(f'{self.base_endpoint}/cluster-types', id)
# ==================== Cluster Groups ====================
async def list_cluster_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all cluster groups."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/cluster-groups', params=params)
async def get_cluster_group(self, id: int) -> Dict:
"""Get a specific cluster group by ID."""
return self.client.get(f'{self.base_endpoint}/cluster-groups', id)
async def create_cluster_group(
self,
name: str,
slug: str,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new cluster group."""
data = {'name': name, 'slug': slug, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/cluster-groups', data)
async def update_cluster_group(self, id: int, **kwargs) -> Dict:
"""Update a cluster group."""
return self.client.patch(f'{self.base_endpoint}/cluster-groups', id, kwargs)
async def delete_cluster_group(self, id: int) -> None:
"""Delete a cluster group."""
self.client.delete(f'{self.base_endpoint}/cluster-groups', id)
# ==================== Clusters ====================
async def list_clusters(
self,
name: Optional[str] = None,
type_id: Optional[int] = None,
group_id: Optional[int] = None,
site_id: Optional[int] = None,
tenant_id: Optional[int] = None,
status: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all clusters with optional filtering."""
params = {k: v for k, v in {
'name': name, 'type_id': type_id, 'group_id': group_id,
'site_id': site_id, 'tenant_id': tenant_id, 'status': status, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/clusters', params=params)
async def get_cluster(self, id: int) -> Dict:
"""Get a specific cluster by ID."""
return self.client.get(f'{self.base_endpoint}/clusters', id)
async def create_cluster(
self,
name: str,
type: int,
status: str = 'active',
group: Optional[int] = None,
site: Optional[int] = None,
tenant: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new cluster."""
data = {'name': name, 'type': type, 'status': status, **kwargs}
for key, val in [
('group', group), ('site', site), ('tenant', tenant), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/clusters', data)
async def update_cluster(self, id: int, **kwargs) -> Dict:
"""Update a cluster."""
return self.client.patch(f'{self.base_endpoint}/clusters', id, kwargs)
async def delete_cluster(self, id: int) -> None:
"""Delete a cluster."""
self.client.delete(f'{self.base_endpoint}/clusters', id)
# ==================== Virtual Machines ====================
async def list_virtual_machines(
self,
name: Optional[str] = None,
cluster_id: Optional[int] = None,
site_id: Optional[int] = None,
role_id: Optional[int] = None,
tenant_id: Optional[int] = None,
platform_id: Optional[int] = None,
status: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all virtual machines with optional filtering."""
params = {k: v for k, v in {
'name': name, 'cluster_id': cluster_id, 'site_id': site_id,
'role_id': role_id, 'tenant_id': tenant_id, 'platform_id': platform_id,
'status': status, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/virtual-machines', params=params)
async def get_virtual_machine(self, id: int) -> Dict:
"""Get a specific virtual machine by ID."""
return self.client.get(f'{self.base_endpoint}/virtual-machines', id)
async def create_virtual_machine(
self,
name: str,
status: str = 'active',
cluster: Optional[int] = None,
site: Optional[int] = None,
role: Optional[int] = None,
tenant: Optional[int] = None,
platform: Optional[int] = None,
primary_ip4: Optional[int] = None,
primary_ip6: Optional[int] = None,
vcpus: Optional[float] = None,
memory: Optional[int] = None,
disk: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new virtual machine."""
data = {'name': name, 'status': status, **kwargs}
for key, val in [
('cluster', cluster), ('site', site), ('role', role),
('tenant', tenant), ('platform', platform),
('primary_ip4', primary_ip4), ('primary_ip6', primary_ip6),
('vcpus', vcpus), ('memory', memory), ('disk', disk),
('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/virtual-machines', data)
async def update_virtual_machine(self, id: int, **kwargs) -> Dict:
"""Update a virtual machine."""
return self.client.patch(f'{self.base_endpoint}/virtual-machines', id, kwargs)
async def delete_virtual_machine(self, id: int) -> None:
"""Delete a virtual machine."""
self.client.delete(f'{self.base_endpoint}/virtual-machines', id)
# ==================== VM Interfaces ====================
async def list_vm_interfaces(
self,
virtual_machine_id: Optional[int] = None,
name: Optional[str] = None,
enabled: Optional[bool] = None,
**kwargs
) -> List[Dict]:
"""List all VM interfaces."""
params = {k: v for k, v in {
'virtual_machine_id': virtual_machine_id, 'name': name, 'enabled': enabled, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/interfaces', params=params)
async def get_vm_interface(self, id: int) -> Dict:
"""Get a specific VM interface by ID."""
return self.client.get(f'{self.base_endpoint}/interfaces', id)
async def create_vm_interface(
self,
virtual_machine: int,
name: str,
enabled: bool = True,
mtu: Optional[int] = None,
mac_address: Optional[str] = None,
description: Optional[str] = None,
mode: Optional[str] = None,
untagged_vlan: Optional[int] = None,
tagged_vlans: Optional[List[int]] = None,
**kwargs
) -> Dict:
"""Create a new VM interface."""
data = {'virtual_machine': virtual_machine, 'name': name, 'enabled': enabled, **kwargs}
for key, val in [
('mtu', mtu), ('mac_address', mac_address), ('description', description),
('mode', mode), ('untagged_vlan', untagged_vlan), ('tagged_vlans', tagged_vlans)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/interfaces', data)
async def update_vm_interface(self, id: int, **kwargs) -> Dict:
"""Update a VM interface."""
return self.client.patch(f'{self.base_endpoint}/interfaces', id, kwargs)
async def delete_vm_interface(self, id: int) -> None:
"""Delete a VM interface."""
self.client.delete(f'{self.base_endpoint}/interfaces', id)
# ==================== Virtual Disks ====================
async def list_virtual_disks(
self,
virtual_machine_id: Optional[int] = None,
name: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all virtual disks."""
params = {k: v for k, v in {
'virtual_machine_id': virtual_machine_id, 'name': name, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/virtual-disks', params=params)
async def get_virtual_disk(self, id: int) -> Dict:
"""Get a specific virtual disk by ID."""
return self.client.get(f'{self.base_endpoint}/virtual-disks', id)
async def create_virtual_disk(
self,
virtual_machine: int,
name: str,
size: int,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new virtual disk."""
data = {'virtual_machine': virtual_machine, 'name': name, 'size': size, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/virtual-disks', data)
async def update_virtual_disk(self, id: int, **kwargs) -> Dict:
"""Update a virtual disk."""
return self.client.patch(f'{self.base_endpoint}/virtual-disks', id, kwargs)
async def delete_virtual_disk(self, id: int) -> None:
"""Delete a virtual disk."""
self.client.delete(f'{self.base_endpoint}/virtual-disks', id)

View File

@@ -0,0 +1,428 @@
"""
VPN tools for NetBox MCP Server.
Covers: Tunnels, Tunnel Groups, Tunnel Terminations, IKE/IPSec Policies, and L2VPN.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class VPNTools:
"""Tools for VPN operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'vpn'
# ==================== Tunnel Groups ====================
async def list_tunnel_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all tunnel groups."""
params = {k: v for k, v in {'name': name, 'slug': slug, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/tunnel-groups', params=params)
async def get_tunnel_group(self, id: int) -> Dict:
"""Get a specific tunnel group by ID."""
return self.client.get(f'{self.base_endpoint}/tunnel-groups', id)
async def create_tunnel_group(
self,
name: str,
slug: str,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new tunnel group."""
data = {'name': name, 'slug': slug, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/tunnel-groups', data)
async def update_tunnel_group(self, id: int, **kwargs) -> Dict:
"""Update a tunnel group."""
return self.client.patch(f'{self.base_endpoint}/tunnel-groups', id, kwargs)
async def delete_tunnel_group(self, id: int) -> None:
"""Delete a tunnel group."""
self.client.delete(f'{self.base_endpoint}/tunnel-groups', id)
# ==================== Tunnels ====================
async def list_tunnels(
self,
name: Optional[str] = None,
status: Optional[str] = None,
group_id: Optional[int] = None,
encapsulation: Optional[str] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all tunnels with optional filtering."""
params = {k: v for k, v in {
'name': name, 'status': status, 'group_id': group_id,
'encapsulation': encapsulation, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/tunnels', params=params)
async def get_tunnel(self, id: int) -> Dict:
"""Get a specific tunnel by ID."""
return self.client.get(f'{self.base_endpoint}/tunnels', id)
async def create_tunnel(
self,
name: str,
status: str = 'active',
encapsulation: str = 'ipsec-tunnel',
group: Optional[int] = None,
ipsec_profile: Optional[int] = None,
tenant: Optional[int] = None,
tunnel_id: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new tunnel."""
data = {'name': name, 'status': status, 'encapsulation': encapsulation, **kwargs}
for key, val in [
('group', group), ('ipsec_profile', ipsec_profile),
('tenant', tenant), ('tunnel_id', tunnel_id), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/tunnels', data)
async def update_tunnel(self, id: int, **kwargs) -> Dict:
"""Update a tunnel."""
return self.client.patch(f'{self.base_endpoint}/tunnels', id, kwargs)
async def delete_tunnel(self, id: int) -> None:
"""Delete a tunnel."""
self.client.delete(f'{self.base_endpoint}/tunnels', id)
# ==================== Tunnel Terminations ====================
async def list_tunnel_terminations(
self,
tunnel_id: Optional[int] = None,
role: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all tunnel terminations."""
params = {k: v for k, v in {
'tunnel_id': tunnel_id, 'role': role, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/tunnel-terminations', params=params)
async def get_tunnel_termination(self, id: int) -> Dict:
"""Get a specific tunnel termination by ID."""
return self.client.get(f'{self.base_endpoint}/tunnel-terminations', id)
async def create_tunnel_termination(
self,
tunnel: int,
role: str,
termination_type: str,
termination_id: int,
outside_ip: Optional[int] = None,
**kwargs
) -> Dict:
"""Create a new tunnel termination."""
data = {
'tunnel': tunnel, 'role': role,
'termination_type': termination_type, 'termination_id': termination_id, **kwargs
}
if outside_ip:
data['outside_ip'] = outside_ip
return self.client.create(f'{self.base_endpoint}/tunnel-terminations', data)
async def update_tunnel_termination(self, id: int, **kwargs) -> Dict:
"""Update a tunnel termination."""
return self.client.patch(f'{self.base_endpoint}/tunnel-terminations', id, kwargs)
async def delete_tunnel_termination(self, id: int) -> None:
"""Delete a tunnel termination."""
self.client.delete(f'{self.base_endpoint}/tunnel-terminations', id)
# ==================== IKE Proposals ====================
async def list_ike_proposals(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all IKE proposals."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ike-proposals', params=params)
async def get_ike_proposal(self, id: int) -> Dict:
"""Get a specific IKE proposal by ID."""
return self.client.get(f'{self.base_endpoint}/ike-proposals', id)
async def create_ike_proposal(
self,
name: str,
authentication_method: str,
encryption_algorithm: str,
authentication_algorithm: str,
group: int,
sa_lifetime: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IKE proposal."""
data = {
'name': name, 'authentication_method': authentication_method,
'encryption_algorithm': encryption_algorithm,
'authentication_algorithm': authentication_algorithm, 'group': group, **kwargs
}
if sa_lifetime:
data['sa_lifetime'] = sa_lifetime
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/ike-proposals', data)
async def update_ike_proposal(self, id: int, **kwargs) -> Dict:
"""Update an IKE proposal."""
return self.client.patch(f'{self.base_endpoint}/ike-proposals', id, kwargs)
async def delete_ike_proposal(self, id: int) -> None:
"""Delete an IKE proposal."""
self.client.delete(f'{self.base_endpoint}/ike-proposals', id)
# ==================== IKE Policies ====================
async def list_ike_policies(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all IKE policies."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ike-policies', params=params)
async def get_ike_policy(self, id: int) -> Dict:
"""Get a specific IKE policy by ID."""
return self.client.get(f'{self.base_endpoint}/ike-policies', id)
async def create_ike_policy(
self,
name: str,
version: int,
mode: str,
proposals: List[int],
preshared_key: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IKE policy."""
data = {'name': name, 'version': version, 'mode': mode, 'proposals': proposals, **kwargs}
if preshared_key:
data['preshared_key'] = preshared_key
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/ike-policies', data)
async def update_ike_policy(self, id: int, **kwargs) -> Dict:
"""Update an IKE policy."""
return self.client.patch(f'{self.base_endpoint}/ike-policies', id, kwargs)
async def delete_ike_policy(self, id: int) -> None:
"""Delete an IKE policy."""
self.client.delete(f'{self.base_endpoint}/ike-policies', id)
# ==================== IPSec Proposals ====================
async def list_ipsec_proposals(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all IPSec proposals."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ipsec-proposals', params=params)
async def get_ipsec_proposal(self, id: int) -> Dict:
"""Get a specific IPSec proposal by ID."""
return self.client.get(f'{self.base_endpoint}/ipsec-proposals', id)
async def create_ipsec_proposal(
self,
name: str,
encryption_algorithm: str,
authentication_algorithm: str,
sa_lifetime_seconds: Optional[int] = None,
sa_lifetime_data: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IPSec proposal."""
data = {
'name': name, 'encryption_algorithm': encryption_algorithm,
'authentication_algorithm': authentication_algorithm, **kwargs
}
for key, val in [
('sa_lifetime_seconds', sa_lifetime_seconds),
('sa_lifetime_data', sa_lifetime_data), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/ipsec-proposals', data)
async def update_ipsec_proposal(self, id: int, **kwargs) -> Dict:
"""Update an IPSec proposal."""
return self.client.patch(f'{self.base_endpoint}/ipsec-proposals', id, kwargs)
async def delete_ipsec_proposal(self, id: int) -> None:
"""Delete an IPSec proposal."""
self.client.delete(f'{self.base_endpoint}/ipsec-proposals', id)
# ==================== IPSec Policies ====================
async def list_ipsec_policies(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all IPSec policies."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ipsec-policies', params=params)
async def get_ipsec_policy(self, id: int) -> Dict:
"""Get a specific IPSec policy by ID."""
return self.client.get(f'{self.base_endpoint}/ipsec-policies', id)
async def create_ipsec_policy(
self,
name: str,
proposals: List[int],
pfs_group: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IPSec policy."""
data = {'name': name, 'proposals': proposals, **kwargs}
if pfs_group:
data['pfs_group'] = pfs_group
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/ipsec-policies', data)
async def update_ipsec_policy(self, id: int, **kwargs) -> Dict:
"""Update an IPSec policy."""
return self.client.patch(f'{self.base_endpoint}/ipsec-policies', id, kwargs)
async def delete_ipsec_policy(self, id: int) -> None:
"""Delete an IPSec policy."""
self.client.delete(f'{self.base_endpoint}/ipsec-policies', id)
# ==================== IPSec Profiles ====================
async def list_ipsec_profiles(self, name: Optional[str] = None, **kwargs) -> List[Dict]:
"""List all IPSec profiles."""
params = {k: v for k, v in {'name': name, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/ipsec-profiles', params=params)
async def get_ipsec_profile(self, id: int) -> Dict:
"""Get a specific IPSec profile by ID."""
return self.client.get(f'{self.base_endpoint}/ipsec-profiles', id)
async def create_ipsec_profile(
self,
name: str,
mode: str,
ike_policy: int,
ipsec_policy: int,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new IPSec profile."""
data = {'name': name, 'mode': mode, 'ike_policy': ike_policy, 'ipsec_policy': ipsec_policy, **kwargs}
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/ipsec-profiles', data)
async def update_ipsec_profile(self, id: int, **kwargs) -> Dict:
"""Update an IPSec profile."""
return self.client.patch(f'{self.base_endpoint}/ipsec-profiles', id, kwargs)
async def delete_ipsec_profile(self, id: int) -> None:
"""Delete an IPSec profile."""
self.client.delete(f'{self.base_endpoint}/ipsec-profiles', id)
# ==================== L2VPN ====================
async def list_l2vpns(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
type: Optional[str] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all L2VPNs with optional filtering."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'type': type, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/l2vpns', params=params)
async def get_l2vpn(self, id: int) -> Dict:
"""Get a specific L2VPN by ID."""
return self.client.get(f'{self.base_endpoint}/l2vpns', id)
async def create_l2vpn(
self,
name: str,
slug: str,
type: str,
identifier: Optional[int] = None,
tenant: Optional[int] = None,
description: Optional[str] = None,
import_targets: Optional[List[int]] = None,
export_targets: Optional[List[int]] = None,
**kwargs
) -> Dict:
"""Create a new L2VPN."""
data = {'name': name, 'slug': slug, 'type': type, **kwargs}
for key, val in [
('identifier', identifier), ('tenant', tenant), ('description', description),
('import_targets', import_targets), ('export_targets', export_targets)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/l2vpns', data)
async def update_l2vpn(self, id: int, **kwargs) -> Dict:
"""Update an L2VPN."""
return self.client.patch(f'{self.base_endpoint}/l2vpns', id, kwargs)
async def delete_l2vpn(self, id: int) -> None:
"""Delete an L2VPN."""
self.client.delete(f'{self.base_endpoint}/l2vpns', id)
# ==================== L2VPN Terminations ====================
async def list_l2vpn_terminations(
self,
l2vpn_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all L2VPN terminations."""
params = {k: v for k, v in {'l2vpn_id': l2vpn_id, **kwargs}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/l2vpn-terminations', params=params)
async def get_l2vpn_termination(self, id: int) -> Dict:
"""Get a specific L2VPN termination by ID."""
return self.client.get(f'{self.base_endpoint}/l2vpn-terminations', id)
async def create_l2vpn_termination(
self,
l2vpn: int,
assigned_object_type: str,
assigned_object_id: int,
**kwargs
) -> Dict:
"""Create a new L2VPN termination."""
data = {
'l2vpn': l2vpn, 'assigned_object_type': assigned_object_type,
'assigned_object_id': assigned_object_id, **kwargs
}
return self.client.create(f'{self.base_endpoint}/l2vpn-terminations', data)
async def update_l2vpn_termination(self, id: int, **kwargs) -> Dict:
"""Update an L2VPN termination."""
return self.client.patch(f'{self.base_endpoint}/l2vpn-terminations', id, kwargs)
async def delete_l2vpn_termination(self, id: int) -> None:
"""Delete an L2VPN termination."""
self.client.delete(f'{self.base_endpoint}/l2vpn-terminations', id)

View File

@@ -0,0 +1,166 @@
"""
Wireless tools for NetBox MCP Server.
Covers: Wireless LANs, Wireless LAN Groups, and Wireless Links.
"""
import logging
from typing import List, Dict, Optional, Any
from ..netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
class WirelessTools:
"""Tools for Wireless operations in NetBox"""
def __init__(self, client: NetBoxClient):
self.client = client
self.base_endpoint = 'wireless'
# ==================== Wireless LAN Groups ====================
async def list_wireless_lan_groups(
self,
name: Optional[str] = None,
slug: Optional[str] = None,
parent_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all wireless LAN groups."""
params = {k: v for k, v in {
'name': name, 'slug': slug, 'parent_id': parent_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/wireless-lan-groups', params=params)
async def get_wireless_lan_group(self, id: int) -> Dict:
"""Get a specific wireless LAN group by ID."""
return self.client.get(f'{self.base_endpoint}/wireless-lan-groups', id)
async def create_wireless_lan_group(
self,
name: str,
slug: str,
parent: Optional[int] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new wireless LAN group."""
data = {'name': name, 'slug': slug, **kwargs}
if parent:
data['parent'] = parent
if description:
data['description'] = description
return self.client.create(f'{self.base_endpoint}/wireless-lan-groups', data)
async def update_wireless_lan_group(self, id: int, **kwargs) -> Dict:
"""Update a wireless LAN group."""
return self.client.patch(f'{self.base_endpoint}/wireless-lan-groups', id, kwargs)
async def delete_wireless_lan_group(self, id: int) -> None:
"""Delete a wireless LAN group."""
self.client.delete(f'{self.base_endpoint}/wireless-lan-groups', id)
# ==================== Wireless LANs ====================
async def list_wireless_lans(
self,
ssid: Optional[str] = None,
group_id: Optional[int] = None,
vlan_id: Optional[int] = None,
tenant_id: Optional[int] = None,
status: Optional[str] = None,
auth_type: Optional[str] = None,
**kwargs
) -> List[Dict]:
"""List all wireless LANs with optional filtering."""
params = {k: v for k, v in {
'ssid': ssid, 'group_id': group_id, 'vlan_id': vlan_id,
'tenant_id': tenant_id, 'status': status, 'auth_type': auth_type, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/wireless-lans', params=params)
async def get_wireless_lan(self, id: int) -> Dict:
"""Get a specific wireless LAN by ID."""
return self.client.get(f'{self.base_endpoint}/wireless-lans', id)
async def create_wireless_lan(
self,
ssid: str,
status: str = 'active',
group: Optional[int] = None,
vlan: Optional[int] = None,
tenant: Optional[int] = None,
auth_type: Optional[str] = None,
auth_cipher: Optional[str] = None,
auth_psk: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new wireless LAN."""
data = {'ssid': ssid, 'status': status, **kwargs}
for key, val in [
('group', group), ('vlan', vlan), ('tenant', tenant),
('auth_type', auth_type), ('auth_cipher', auth_cipher),
('auth_psk', auth_psk), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/wireless-lans', data)
async def update_wireless_lan(self, id: int, **kwargs) -> Dict:
"""Update a wireless LAN."""
return self.client.patch(f'{self.base_endpoint}/wireless-lans', id, kwargs)
async def delete_wireless_lan(self, id: int) -> None:
"""Delete a wireless LAN."""
self.client.delete(f'{self.base_endpoint}/wireless-lans', id)
# ==================== Wireless Links ====================
async def list_wireless_links(
self,
ssid: Optional[str] = None,
status: Optional[str] = None,
tenant_id: Optional[int] = None,
**kwargs
) -> List[Dict]:
"""List all wireless links with optional filtering."""
params = {k: v for k, v in {
'ssid': ssid, 'status': status, 'tenant_id': tenant_id, **kwargs
}.items() if v is not None}
return self.client.list(f'{self.base_endpoint}/wireless-links', params=params)
async def get_wireless_link(self, id: int) -> Dict:
"""Get a specific wireless link by ID."""
return self.client.get(f'{self.base_endpoint}/wireless-links', id)
async def create_wireless_link(
self,
interface_a: int,
interface_b: int,
ssid: Optional[str] = None,
status: str = 'connected',
tenant: Optional[int] = None,
auth_type: Optional[str] = None,
auth_cipher: Optional[str] = None,
auth_psk: Optional[str] = None,
description: Optional[str] = None,
**kwargs
) -> Dict:
"""Create a new wireless link."""
data = {'interface_a': interface_a, 'interface_b': interface_b, 'status': status, **kwargs}
for key, val in [
('ssid', ssid), ('tenant', tenant), ('auth_type', auth_type),
('auth_cipher', auth_cipher), ('auth_psk', auth_psk), ('description', description)
]:
if val is not None:
data[key] = val
return self.client.create(f'{self.base_endpoint}/wireless-links', data)
async def update_wireless_link(self, id: int, **kwargs) -> Dict:
"""Update a wireless link."""
return self.client.patch(f'{self.base_endpoint}/wireless-links', id, kwargs)
async def delete_wireless_link(self, id: int) -> None:
"""Delete a wireless link."""
self.client.delete(f'{self.base_endpoint}/wireless-links', id)