lmiranda
d84425cbb0
Claude Code only caches the plugin directory when installed from a
marketplace, not parent directories. This broke the shared mcp-servers/
architecture because relative paths like ../../mcp-servers/ resolved
to non-existent locations in the cache.
Changes:
- Move gitea and wikijs MCP servers into plugins/projman/mcp-servers/
- Move netbox MCP server into plugins/cmdb-assistant/mcp-servers/
- Update .mcp.json files to use ${CLAUDE_PLUGIN_ROOT}/mcp-servers/
- Update setup.sh to handle new bundled structure
- Add netbox.env config template to setup.sh
- Update CLAUDE.md and CANONICAL-PATHS.md documentation
This ensures plugins work correctly when installed and cached.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""
|
|
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)
|