development #458

Merged
lmiranda merged 2 commits from development into main 2026-02-09 02:28:49 +00:00
13 changed files with 317 additions and 4775 deletions

View File

@@ -6,6 +6,37 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
## [Unreleased] ## [Unreleased]
### Changed — BREAKING
#### NetBox MCP Server: Gutted to 37 Tools (from 182)
Removed all tools not needed for tracking applications, services, databases, and VPS/servers.
**Deleted modules (entire files removed):**
- `circuits.py` — providers, circuits, terminations
- `tenancy.py` — tenants, contacts
- `vpn.py` — tunnels, IKE/IPSec, L2VPN
- `wireless.py` — WLANs, links, groups
**Deleted tool categories within remaining modules:**
- DCIM: regions, locations, racks, rack roles, manufacturers, device types, platforms, cables, console ports, front/rear ports, module bays, power panels, power feeds, virtual chassis, inventory items
- IPAM: VLANs, VLAN groups, VRFs, ASNs, RIRs, aggregates, available IPs, available prefixes
- Virtualization: cluster types, cluster groups, all delete operations
- Extras: custom fields, webhooks, config contexts, update/delete for tags
**Deleted infrastructure:**
- `NETBOX_ENABLED_MODULES` env var and all module filtering code
- `ALL_MODULES` constant, `PREFIX_TO_MODULE` dict, `_get_tool_module()` function
- Conditional module instantiation in server.py
**Token impact:** ~19,810 → ~3,700 tokens (~81% reduction)
**Remaining tools (37):**
- DCIM: sites (4), devices (4), interfaces (3) = 11
- IPAM: IPs (4), prefixes (3), services (3) = 10
- Virtualization: clusters (3), VMs (4), VM interfaces (3) = 10
- Extras: tags (3), journal entries (3) = 6
### Added ### Added
- **All plugins:** Dispatch files now active command handlers — bare `/noun` shows available sub-commands and prompts for selection instead of doing nothing - **All plugins:** Dispatch files now active command handlers — bare `/noun` shows available sub-commands and prompts for selection instead of doing nothing

View File

@@ -1,19 +1,17 @@
# NetBox MCP Server # NetBox MCP Server
MCP (Model Context Protocol) server for comprehensive NetBox API integration with Claude Code. MCP (Model Context Protocol) server for essential NetBox API integration with Claude Code.
## Overview ## Overview
This MCP server provides Claude Code with full access to the NetBox REST API, enabling infrastructure management, documentation, and automation workflows. It covers all major NetBox application areas: This MCP server provides Claude Code with focused access to the NetBox REST API for tracking **servers, services, IP addresses, and databases**. It has been optimized to include only essential tools:
- **DCIM** - Sites, Locations, Racks, Devices, Interfaces, Cables, Power - **DCIM** - Sites, Devices (servers/VPS), Interfaces
- **IPAM** - IP Addresses, Prefixes, VLANs, VRFs, ASNs, Services - **IPAM** - IP Addresses, Prefixes, Services (applications/databases)
- **Circuits** - Providers, Circuits, Terminations
- **Virtualization** - Clusters, Virtual Machines, VM Interfaces - **Virtualization** - Clusters, Virtual Machines, VM Interfaces
- **Tenancy** - Tenants, Contacts, Contact Assignments - **Extras** - Tags, Journal Entries (audit/notes)
- **VPN** - Tunnels, IKE/IPSec Policies, L2VPN
- **Wireless** - Wireless LANs, Links, Groups **Total:** 37 tools (~3,700 tokens) — down from 182 tools (~19,810 tokens).
- **Extras** - Tags, Custom Fields, Webhooks, Config Contexts, Audit Log
## Installation ## Installation
@@ -49,312 +47,227 @@ EOF
### 3. Register with Claude Code ### 3. Register with Claude Code
Add to your Claude Code MCP configuration (`~/.config/claude/mcp.json` or project `.mcp.json`): Add to your Claude Code MCP configuration (`.claude/mcp.json` or project-level `.mcp.json`):
```json ```json
{ {
"mcpServers": { "mcpServers": {
"netbox": { "netbox": {
"command": "/path/to/mcp-servers/netbox/.venv/bin/python", "command": "/path/to/mcp-servers/netbox/.venv/bin/python",
"args": ["-m", "mcp_server.server"], "args": ["-m", "mcp_server"],
"cwd": "/path/to/mcp-servers/netbox" "cwd": "/path/to/mcp-servers/netbox"
} }
} }
} }
``` ```
**Windows:**
```json
{
"mcpServers": {
"netbox": {
"command": "C:\\path\\to\\mcp-servers\\netbox\\.venv\\Scripts\\python.exe",
"args": ["-m", "mcp_server"],
"cwd": "C:\\path\\to\\mcp-servers\\netbox"
}
}
}
```
## Available Tools (37 Total)
### DCIM: Sites, Devices, Interfaces (11 tools)
**Sites (4):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `dcim_list_sites` | List sites | `name`, `status` |
| `dcim_get_site` | Get site by ID | `id` (required) |
| `dcim_create_site` | Create site | `name`, `slug` (required), `status` |
| `dcim_update_site` | Update site | `id` (required), fields to update |
**Devices (4):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `dcim_list_devices` | List devices (servers/VPS) | `name`, `site_id`, `status`, `role_id` |
| `dcim_get_device` | Get device by ID | `id` (required) |
| `dcim_create_device` | Create device | `name`, `device_type`, `role`, `site` (required) |
| `dcim_update_device` | Update device | `id` (required), fields to update |
**Interfaces (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `dcim_list_interfaces` | List device interfaces | `device_id`, `name`, `type` |
| `dcim_get_interface` | Get interface by ID | `id` (required) |
| `dcim_create_interface` | Create interface | `device`, `name`, `type` (required) |
### IPAM: IPs, Prefixes, Services (10 tools)
**IP Addresses (4):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `ipam_list_ip_addresses` | List IP addresses | `address`, `device_id`, `status` |
| `ipam_get_ip_address` | Get IP by ID | `id` (required) |
| `ipam_create_ip_address` | Create IP address | `address` (required), `status`, `assigned_object_type` |
| `ipam_update_ip_address` | Update IP address | `id` (required), fields to update |
**Prefixes (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `ipam_list_prefixes` | List prefixes | `prefix`, `site_id`, `status` |
| `ipam_get_prefix` | Get prefix by ID | `id` (required) |
| `ipam_create_prefix` | Create prefix | `prefix` (required), `status`, `site` |
**Services (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `ipam_list_services` | List services (apps/databases) | `device_id`, `virtual_machine_id`, `name` |
| `ipam_get_service` | Get service by ID | `id` (required) |
| `ipam_create_service` | Create service | `name`, `ports`, `protocol` (required), `device`, `virtual_machine` |
### Virtualization: Clusters, VMs, VM Interfaces (10 tools)
**Clusters (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `virt_list_clusters` | List virtualization clusters | `name`, `site_id` |
| `virt_get_cluster` | Get cluster by ID | `id` (required) |
| `virt_create_cluster` | Create cluster | `name`, `type` (required), `site` |
**Virtual Machines (4):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `virt_list_vms` | List VMs | `name`, `cluster_id`, `site_id`, `status` |
| `virt_get_vm` | Get VM by ID | `id` (required) |
| `virt_create_vm` | Create VM | `name`, `cluster` (required), `vcpus`, `memory`, `disk` |
| `virt_update_vm` | Update VM | `id` (required), fields to update |
**VM Interfaces (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `virt_list_vm_ifaces` | List VM interfaces | `virtual_machine_id` |
| `virt_get_vm_iface` | Get VM interface by ID | `id` (required) |
| `virt_create_vm_iface` | Create VM interface | `virtual_machine`, `name` (required) |
### Extras: Tags, Journal Entries (6 tools)
**Tags (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `extras_list_tags` | List tags | `name` |
| `extras_get_tag` | Get tag by ID | `id` (required) |
| `extras_create_tag` | Create tag | `name`, `slug` (required), `color` |
**Journal Entries (3):**
| Tool | Description | Parameters |
|------|-------------|-----------|
| `extras_list_journal_entries` | List journal entries | `assigned_object_type`, `assigned_object_id` |
| `extras_get_journal_entry` | Get journal entry by ID | `id` (required) |
| `extras_create_journal_entry` | Create journal entry | `assigned_object_type`, `assigned_object_id`, `comments` (required), `kind` |
## Configuration ## Configuration
### Environment Variables All configuration is done via environment variables in `~/.config/claude/netbox.env`:
| Variable | Required | Default | Description | | Variable | Required | Default | Description |
|----------|----------|---------|-------------| |----------|----------|---------|-------------|
| `NETBOX_API_URL` | Yes | - | Full URL to NetBox API (e.g., `https://netbox.example.com/api`) | | `NETBOX_API_URL` | Yes | | NetBox API URL (e.g., https://netbox.example.com/api) |
| `NETBOX_API_TOKEN` | Yes | - | API authentication token | | `NETBOX_API_TOKEN` | Yes | | NetBox API token |
| `NETBOX_VERIFY_SSL` | No | `true` | Verify SSL certificates | | `NETBOX_VERIFY_SSL` | No | true | Verify SSL certificates |
| `NETBOX_TIMEOUT` | No | `30` | Request timeout in seconds | | `NETBOX_TIMEOUT` | No | 30 | Request timeout in seconds |
### Configuration Hierarchy
1. **System-level** (`~/.config/claude/netbox.env`): Credentials and defaults
2. **Project-level** (`.env` in current directory): Optional overrides
## Module Filtering (Token Optimization)
By default, the NetBox MCP server registers all 182 tools across 8 modules, consuming ~19,810 tokens of context. For most workflows, you only need a subset of modules.
### Configuration
Add `NETBOX_ENABLED_MODULES` to your `~/.config/claude/netbox.env`:
```bash
# Enable only specific modules (comma-separated)
NETBOX_ENABLED_MODULES=dcim,ipam,virtualization,extras
```
If unset, all modules are enabled (backward compatible).
### Available Modules
| Module | Tool Count | Description | cmdb-assistant Commands |
|--------|------------|-------------|------------------------|
| `dcim` | ~60 | Sites, devices, racks, interfaces, cables | `/cmdb device`, `/cmdb site`, `/cmdb search`, `/cmdb topology` |
| `ipam` | ~40 | IP addresses, prefixes, VLANs, VRFs | `/cmdb ip`, `/cmdb ip-conflicts`, `/cmdb search` |
| `virtualization` | ~20 | Clusters, VMs, VM interfaces | `/cmdb search`, `/cmdb audit`, `/cmdb register` |
| `extras` | ~12 | Tags, journal entries, audit log | `/cmdb change-audit`, `/cmdb register` |
| `circuits` | ~15 | Providers, circuits, terminations | — |
| `tenancy` | ~12 | Tenants, contacts | — |
| `vpn` | ~15 | Tunnels, IKE/IPSec policies, L2VPN | — |
| `wireless` | ~8 | Wireless LANs, links, groups | — |
### Recommended Configurations
**For cmdb-assistant users** (~43 tools, ~4,500 tokens):
```bash
NETBOX_ENABLED_MODULES=dcim,ipam,virtualization,extras
```
**Basic infrastructure** (~100 tools):
```bash
NETBOX_ENABLED_MODULES=dcim,ipam
```
**Full CMDB** (all modules, ~182 tools):
```bash
# Omit NETBOX_ENABLED_MODULES or set to all modules
NETBOX_ENABLED_MODULES=dcim,ipam,circuits,virtualization,tenancy,vpn,wireless,extras
```
### Startup Logging
On startup, the server logs enabled modules and tool count:
```
NetBox MCP Server initialized: 43 tools registered (modules: dcim, extras, ipam, virtualization)
```
### Disabled Tool Behavior
Calling a tool from a disabled module returns a clear error:
```
Tool 'circuits_list_circuits' is not available (module 'circuits' not enabled).
Enabled modules: dcim, extras, ipam, virtualization
```
## Available Tools
### DCIM (Data Center Infrastructure Management)
| Tool | Description |
|------|-------------|
| `dcim_list_sites` | List all sites |
| `dcim_get_site` | Get site details |
| `dcim_create_site` | Create a new site |
| `dcim_update_site` | Update a site |
| `dcim_delete_site` | Delete a site |
| `dcim_list_devices` | List all devices |
| `dcim_get_device` | Get device details |
| `dcim_create_device` | Create a new device |
| `dcim_update_device` | Update a device |
| `dcim_delete_device` | Delete a device |
| `dcim_list_interfaces` | List device interfaces |
| `dcim_create_interface` | Create an interface |
| `dcim_list_racks` | List all racks |
| `dcim_create_rack` | Create a new rack |
| `dcim_list_cables` | List all cables |
| `dcim_create_cable` | Create a cable connection |
| ... and many more |
### IPAM (IP Address Management)
| Tool | Description |
|------|-------------|
| `ipam_list_prefixes` | List IP prefixes |
| `ipam_create_prefix` | Create a prefix |
| `ipam_list_available_prefixes` | List available child prefixes |
| `ipam_create_available_prefix` | Auto-allocate a prefix |
| `ipam_list_ip_addresses` | List IP addresses |
| `ipam_create_ip_address` | Create an IP address |
| `ipam_list_available_ips` | List available IPs in prefix |
| `ipam_create_available_ip` | Auto-allocate an IP |
| `ipam_list_vlans` | List VLANs |
| `ipam_create_vlan` | Create a VLAN |
| `ipam_list_vrfs` | List VRFs |
| ... and many more |
### Circuits
| Tool | Description |
|------|-------------|
| `circuits_list_providers` | List circuit providers |
| `circuits_create_provider` | Create a provider |
| `circuits_list_circuits` | List circuits |
| `circuits_create_circuit` | Create a circuit |
| `circ_list_terminations` | List terminations |
| ... and more |
### Virtualization
| Tool | Description |
|------|-------------|
| `virt_list_clusters` | List clusters |
| `virt_create_cluster` | Create a cluster |
| `virt_list_vms` | List VMs |
| `virt_create_vm` | Create a VM |
| `virt_list_vm_ifaces` | List VM interfaces |
| ... and more |
### Tenancy
| Tool | Description |
|------|-------------|
| `tenancy_list_tenants` | List tenants |
| `tenancy_create_tenant` | Create a tenant |
| `tenancy_list_contacts` | List contacts |
| `tenancy_create_contact` | Create a contact |
| ... and more |
### VPN
| Tool | Description |
|------|-------------|
| `vpn_list_tunnels` | List VPN tunnels |
| `vpn_create_tunnel` | Create a tunnel |
| `vpn_list_l2vpns` | List L2VPNs |
| `vpn_list_ike_policies` | List IKE policies |
| `vpn_list_ipsec_policies` | List IPSec policies |
| ... and more |
### Wireless
| Tool | Description |
|------|-------------|
| `wlan_list_lans` | List wireless LANs |
| `wlan_create_lan` | Create a WLAN |
| `wlan_list_links` | List wireless links |
| ... and more |
### Extras
| Tool | Description |
|------|-------------|
| `extras_list_tags` | List all tags |
| `extras_create_tag` | Create a tag |
| `extras_list_custom_fields` | List custom fields |
| `extras_list_webhooks` | List webhooks |
| `extras_list_journal_entries` | List journal entries |
| `extras_create_journal_entry` | Create journal entry |
| `extras_list_object_changes` | View audit log |
| `extras_list_config_contexts` | List config contexts |
| ... and more |
## Usage Examples
### List all devices at a site
```
Use the dcim_list_devices tool with site_id filter to see all devices at site 5
```
### Create a new prefix and allocate IPs
```
1. Use ipam_create_prefix to create 10.0.1.0/24
2. Use ipam_list_available_ips with the prefix ID to see available addresses
3. Use ipam_create_available_ip to auto-allocate the next IP
```
### Document a new server
```
1. Use dcim_create_device to create the device
2. Use dcim_create_interface to add network interfaces
3. Use ipam_create_ip_address to assign IPs to interfaces
4. Use extras_create_journal_entry to add notes
```
### Audit recent changes
```
Use extras_list_object_changes to see recent modifications in NetBox
```
## Architecture ## Architecture
### Hybrid Configuration
- **System-level:** `~/.config/claude/netbox.env` (credentials)
- **Project-level:** `.env` (optional overrides)
### Tool Routing
Tool names follow the pattern `{module}_{action}_{resource}`:
- `dcim_list_sites` → DCIMTools.list_sites()
- `ipam_create_service` → IPAMTools.create_service()
- `virt_list_vms` → VirtualizationTools.list_virtual_machines()
Shortened names (virt_*) are mapped via TOOL_NAME_MAP to meet the 28-character MCP limit.
### Error Handling
All tools return JSON responses. Errors are caught and returned as:
```json
{
"error": "Error message",
"status_code": 404
}
``` ```
mcp-servers/netbox/
## Development
### Testing
```bash
# Test import
python -c "from mcp_server.server import NetBoxMCPServer; print('OK')"
# Test tool count
python -c "from mcp_server.server import TOOL_DEFINITIONS; print(f'{len(TOOL_DEFINITIONS)} tools')"
```
### File Structure
```
netbox/
├── mcp_server/ ├── mcp_server/
│ ├── __init__.py │ ├── __init__.py
│ ├── server.py # Main MCP server (37 TOOL_DEFINITIONS)
│ ├── config.py # Configuration loader │ ├── config.py # Configuration loader
│ ├── netbox_client.py # Generic HTTP client │ ├── netbox_client.py # HTTP client wrapper
│ ├── server.py # MCP server entry point
│ └── tools/ │ └── tools/
│ ├── __init__.py │ ├── __init__.py
│ ├── dcim.py # DCIM operations │ ├── dcim.py # Sites, Devices, Interfaces
│ ├── ipam.py # IPAM operations │ ├── ipam.py # IPs, Prefixes, Services
│ ├── circuits.py # Circuits operations │ ├── virtualization.py # Clusters, VMs, VM Interfaces
── virtualization.py ── extras.py # Tags, Journal Entries
│ ├── tenancy.py ├── .venv/ # Python virtual environment
│ ├── vpn.py
│ ├── wireless.py
│ └── extras.py
├── tests/
│ └── __init__.py
├── requirements.txt ├── requirements.txt
└── README.md └── README.md
``` ```
## API Coverage
This MCP server provides comprehensive coverage of the NetBox REST API v4.x:
- Full CRUD operations for all major models
- Filtering and search capabilities
- Special endpoints (available prefixes, available IPs)
- Pagination handling (automatic)
- Error handling with detailed messages
## Error Handling
The server returns detailed error messages from the NetBox API, including:
- Validation errors
- Authentication failures
- Not found errors
- Permission errors
## Security Notes
- API tokens should be kept secure and not committed to version control
- Use environment variables or the system config file for credentials
- SSL verification is enabled by default
- Consider using read-only tokens for query-only workflows
## Troubleshooting ## Troubleshooting
### Common Issues ### MCP Server Won't Start
1. **Connection refused**: Check `NETBOX_API_URL` is correct and accessible **Check configuration:**
2. **401 Unauthorized**: Verify your API token is valid ```bash
3. **SSL errors**: Set `NETBOX_VERIFY_SSL=false` for self-signed certs (not recommended for production) cat ~/.config/claude/netbox.env
4. **Timeout errors**: Increase `NETBOX_TIMEOUT` for slow connections
### Debug Mode
Enable debug logging:
```python
import logging
logging.basicConfig(level=logging.DEBUG)
``` ```
## Contributing **Test credentials:**
```bash
curl -H "Authorization: Token YOUR_TOKEN" https://netbox.example.com/api/
```
1. Follow the existing code patterns ### Tools Not Appearing in Claude
2. Add tests for new functionality
3. Update documentation for new tools **Verify MCP registration:**
4. Ensure compatibility with NetBox 4.x API ```bash
cat ~/.claude/mcp.json # or project-level .mcp.json
```
**Check MCP server logs:**
Claude Code will show MCP server stderr in the UI.
### Connection Errors
- Verify `NETBOX_API_URL` ends with `/api`
- Check firewall/network connectivity to NetBox instance
- Ensure API token has required permissions
## License ## License
MIT License - Part of the Leo Claude Marketplace. MIT License - See LICENSE file for details.
## Contributing
This MCP server is part of the leo-claude-mktplace project. For issues or contributions, refer to the main repository.

View File

@@ -9,17 +9,11 @@ from pathlib import Path
from dotenv import load_dotenv from dotenv import load_dotenv
import os import os
import logging import logging
from typing import Dict, List, Optional, Set from typing import Dict, Optional
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# All available NetBox modules
ALL_MODULES = frozenset([
'dcim', 'ipam', 'circuits', 'virtualization',
'tenancy', 'vpn', 'wireless', 'extras'
])
class NetBoxConfig: class NetBoxConfig:
"""Configuration loader for NetBox MCP Server""" """Configuration loader for NetBox MCP Server"""
@@ -29,7 +23,6 @@ class NetBoxConfig:
self.api_token: Optional[str] = None self.api_token: Optional[str] = None
self.verify_ssl: bool = True self.verify_ssl: bool = True
self.timeout: int = 30 self.timeout: int = 30
self.enabled_modules: Set[str] = set(ALL_MODULES)
def load(self) -> Dict[str, any]: def load(self) -> Dict[str, any]:
""" """
@@ -80,9 +73,6 @@ class NetBoxConfig:
self.timeout = 30 self.timeout = 30
logger.warning(f"Invalid NETBOX_TIMEOUT value '{timeout_str}', using default 30") logger.warning(f"Invalid NETBOX_TIMEOUT value '{timeout_str}', using default 30")
# Module filtering
self.enabled_modules = self._load_enabled_modules()
# Validate required variables # Validate required variables
self._validate() self._validate()
@@ -94,8 +84,7 @@ class NetBoxConfig:
'api_url': self.api_url, 'api_url': self.api_url,
'api_token': self.api_token, 'api_token': self.api_token,
'verify_ssl': self.verify_ssl, 'verify_ssl': self.verify_ssl,
'timeout': self.timeout, 'timeout': self.timeout
'enabled_modules': self.enabled_modules
} }
def _validate(self) -> None: def _validate(self) -> None:
@@ -117,40 +106,3 @@ class NetBoxConfig:
f"Missing required configuration: {', '.join(missing)}\n" f"Missing required configuration: {', '.join(missing)}\n"
"Check your ~/.config/claude/netbox.env file" "Check your ~/.config/claude/netbox.env file"
) )
def _load_enabled_modules(self) -> Set[str]:
"""
Load enabled modules from NETBOX_ENABLED_MODULES environment variable.
Format: Comma-separated list of module names.
Example: NETBOX_ENABLED_MODULES=dcim,ipam,virtualization,extras
Returns:
Set of enabled module names. If env var is unset/empty, returns all modules.
"""
modules_str = os.getenv('NETBOX_ENABLED_MODULES', '').strip()
if not modules_str:
logger.info("NETBOX_ENABLED_MODULES not set, all modules enabled (default)")
return set(ALL_MODULES)
# Parse comma-separated list, strip whitespace
requested = {m.strip().lower() for m in modules_str.split(',') if m.strip()}
# Validate module names
invalid = requested - ALL_MODULES
if invalid:
logger.warning(
f"Unknown modules in NETBOX_ENABLED_MODULES: {', '.join(sorted(invalid))}. "
f"Valid modules: {', '.join(sorted(ALL_MODULES))}"
)
# Return only valid modules
enabled = requested & ALL_MODULES
if not enabled:
logger.warning("No valid modules enabled, falling back to all modules")
return set(ALL_MODULES)
logger.info(f"Enabled modules: {', '.join(sorted(enabled))}")
return enabled

File diff suppressed because it is too large Load Diff

View File

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

View File

@@ -1,373 +0,0 @@
"""
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

@@ -1,7 +1,7 @@
""" """
DCIM (Data Center Infrastructure Management) tools for NetBox MCP Server. DCIM (Data Center Infrastructure Management) tools for NetBox MCP Server.
Covers: Sites, Locations, Racks, Devices, Cables, Interfaces, and related models. Covers: Sites, Devices, and Interfaces only.
""" """
import logging import logging
from typing import List, Dict, Optional, Any from typing import List, Dict, Optional, Any
@@ -17,74 +17,6 @@ class DCIMTools:
self.client = client self.client = client
self.base_endpoint = 'dcim' 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 ==================== # ==================== Sites ====================
async def list_sites( async def list_sites(
@@ -142,359 +74,6 @@ class DCIMTools:
"""Update a site.""" """Update a site."""
return self.client.patch(f'{self.base_endpoint}/sites', id, kwargs) 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 ==================== # ==================== Devices ====================
async def list_devices( async def list_devices(
@@ -565,34 +144,6 @@ class DCIMTools:
"""Update a device.""" """Update a device."""
return self.client.patch(f'{self.base_endpoint}/devices', id, kwargs) 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 ==================== # ==================== Interfaces ====================
async def list_interfaces( async def list_interfaces(
@@ -636,300 +187,3 @@ class DCIMTools:
if val is not None: if val is not None:
data[key] = val data[key] = val
return self.client.create(f'{self.base_endpoint}/interfaces', data) 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

@@ -1,7 +1,7 @@
""" """
Extras tools for NetBox MCP Server. Extras tools for NetBox MCP Server.
Covers: Tags, Custom Fields, Custom Links, Webhooks, Journal Entries, and more. Covers: Tags and Journal Entries only.
""" """
import logging import logging
from typing import List, Dict, Optional, Any from typing import List, Dict, Optional, Any
@@ -50,209 +50,6 @@ class ExtrasTools:
data['description'] = description data['description'] = description
return self.client.create(f'{self.base_endpoint}/tags', data) 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 ==================== # ==================== Journal Entries ====================
async def list_journal_entries( async def list_journal_entries(
@@ -288,273 +85,3 @@ class ExtrasTools:
'comments': comments, 'kind': kind, **kwargs 'comments': comments, 'kind': kind, **kwargs
} }
return self.client.create(f'{self.base_endpoint}/journal-entries', data) 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

@@ -1,7 +1,7 @@
""" """
IPAM (IP Address Management) tools for NetBox MCP Server. IPAM (IP Address Management) tools for NetBox MCP Server.
Covers: IP Addresses, Prefixes, VLANs, VRFs, ASNs, and related models. Covers: IP Addresses, Prefixes, and Services only.
""" """
import logging import logging
from typing import List, Dict, Optional, Any from typing import List, Dict, Optional, Any
@@ -17,164 +17,6 @@ class IPAMTools:
self.client = client self.client = client
self.base_endpoint = 'ipam' 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 ==================== # ==================== Prefixes ====================
async def list_prefixes( async def list_prefixes(
@@ -230,83 +72,6 @@ class IPAMTools:
data[key] = val data[key] = val
return self.client.create(f'{self.base_endpoint}/prefixes', data) 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 ==================== # ==================== IP Addresses ====================
async def list_ip_addresses( async def list_ip_addresses(
@@ -368,271 +133,6 @@ class IPAMTools:
"""Update an IP address.""" """Update an IP address."""
return self.client.patch(f'{self.base_endpoint}/ip-addresses', id, kwargs) 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 ==================== # ==================== Services ====================
async def list_services( async def list_services(
@@ -675,44 +175,3 @@ class IPAMTools:
if val is not None: if val is not None:
data[key] = val data[key] = val
return self.client.create(f'{self.base_endpoint}/services', data) 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

@@ -1,281 +0,0 @@
"""
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

@@ -1,7 +1,7 @@
""" """
Virtualization tools for NetBox MCP Server. Virtualization tools for NetBox MCP Server.
Covers: Clusters, Virtual Machines, VM Interfaces, and related models. Covers: Clusters, Virtual Machines, and VM Interfaces only.
""" """
import logging import logging
from typing import List, Dict, Optional, Any from typing import List, Dict, Optional, Any
@@ -17,80 +17,6 @@ class VirtualizationTools:
self.client = client self.client = client
self.base_endpoint = 'virtualization' 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 ==================== # ==================== Clusters ====================
async def list_clusters( async def list_clusters(
@@ -134,14 +60,6 @@ class VirtualizationTools:
data[key] = val data[key] = val
return self.client.create(f'{self.base_endpoint}/clusters', data) 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 ==================== # ==================== Virtual Machines ====================
async def list_virtual_machines( async def list_virtual_machines(
@@ -201,10 +119,6 @@ class VirtualizationTools:
"""Update a virtual machine.""" """Update a virtual machine."""
return self.client.patch(f'{self.base_endpoint}/virtual-machines', id, kwargs) 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 ==================== # ==================== VM Interfaces ====================
async def list_vm_interfaces( async def list_vm_interfaces(
@@ -246,51 +160,3 @@ class VirtualizationTools:
if val is not None: if val is not None:
data[key] = val data[key] = val
return self.client.create(f'{self.base_endpoint}/interfaces', data) 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

@@ -1,428 +0,0 @@
"""
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

@@ -1,166 +0,0 @@
"""
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)