hosts.py•7.52 kB
"""MCP tools for host/device queries."""
import logging
from typing import Any, Dict, List, Optional
from mcp.types import Tool, TextContent
from src.netbox_client import NetBoxClient
from src.state_confidence import StateConfidenceClient
logger = logging.getLogger(__name__)
def get_host_tools(
netbox_client: NetBoxClient, state_client: Optional[StateConfidenceClient]
) -> List[Tool]:
"""Get all host-related MCP tools.
Args:
netbox_client: NetBox API client
state_client: Optional state confidence client
Returns:
List of Tool definitions
"""
return [
Tool(
name="list_hosts",
description="List all hosts (devices) from NetBox. Supports filtering by name, IP address, or role.",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Filter by device name (partial match supported)",
},
"primary_ip": {
"type": "string",
"description": "Filter by primary IP address",
},
"role": {
"type": "string",
"description": "Filter by device role slug",
},
"limit": {
"type": "integer",
"description": "Maximum number of results (default: 100)",
"default": 100,
},
"include_certainty": {
"type": "boolean",
"description": "Include state confidence scores in response",
"default": True,
},
},
},
),
Tool(
name="get_host",
description="Get detailed information about a specific host (device) by name.",
inputSchema={
"type": "object",
"properties": {
"hostname": {
"type": "string",
"description": "Device hostname",
},
"include_certainty": {
"type": "boolean",
"description": "Include state confidence score in response",
"default": True,
},
},
"required": ["hostname"],
},
),
Tool(
name="search_hosts",
description="Search for hosts (devices) by name or IP address. Returns matching devices.",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query (hostname or IP address)",
},
"limit": {
"type": "integer",
"description": "Maximum number of results (default: 50)",
"default": 50,
},
"include_certainty": {
"type": "boolean",
"description": "Include state confidence scores in response",
"default": True,
},
},
"required": ["query"],
},
),
]
async def handle_list_hosts(
arguments: Dict[str, Any],
netbox_client: NetBoxClient,
state_client: Optional[StateConfidenceClient],
) -> List[TextContent]:
"""Handle list_hosts tool call.
Args:
arguments: Tool arguments
netbox_client: NetBox API client
state_client: Optional state confidence client
Returns:
List of TextContent with results
"""
name = arguments.get("name")
primary_ip = arguments.get("primary_ip")
role = arguments.get("role")
limit = arguments.get("limit", 100)
include_certainty = arguments.get("include_certainty", True)
devices = netbox_client.list_devices(
name=name, primary_ip=primary_ip, role=role, limit=limit
)
results = []
for device in devices:
result = device.copy()
if include_certainty and state_client:
certainty = state_client.get_certainty_score(
state_id=device.get("name", ""), state_type="device"
)
if certainty:
result["certainty"] = certainty
results.append(result)
import json
return [
TextContent(
type="text",
text=f"Found {len(results)} host(s):\n\n{json.dumps(results, indent=2)}",
)
]
async def handle_get_host(
arguments: Dict[str, Any],
netbox_client: NetBoxClient,
state_client: Optional[StateConfidenceClient],
) -> List[TextContent]:
"""Handle get_host tool call.
Args:
arguments: Tool arguments
netbox_client: NetBox API client
state_client: Optional state confidence client
Returns:
List of TextContent with result
"""
hostname = arguments.get("hostname")
if not hostname:
return [
TextContent(
type="text",
text="Error: hostname parameter is required",
)
]
include_certainty = arguments.get("include_certainty", True)
device = netbox_client.get_device(hostname)
if not device:
return [
TextContent(
type="text",
text=f"Host '{hostname}' not found in NetBox",
)
]
result = device.copy()
if include_certainty and state_client:
certainty = state_client.get_certainty_score(
state_id=hostname, state_type="device"
)
if certainty:
result["certainty"] = certainty
import json
return [
TextContent(
type="text",
text=f"Host details:\n\n{json.dumps(result, indent=2)}",
)
]
async def handle_search_hosts(
arguments: Dict[str, Any],
netbox_client: NetBoxClient,
state_client: Optional[StateConfidenceClient],
) -> List[TextContent]:
"""Handle search_hosts tool call.
Args:
arguments: Tool arguments
netbox_client: NetBox API client
state_client: Optional state confidence client
Returns:
List of TextContent with results
"""
query = arguments.get("query")
if not query:
return [
TextContent(
type="text",
text="Error: query parameter is required",
)
]
limit = arguments.get("limit", 50)
include_certainty = arguments.get("include_certainty", True)
devices = netbox_client.search_devices(query, limit=limit)
results = []
for device in devices:
result = device.copy()
if include_certainty and state_client:
certainty = state_client.get_certainty_score(
state_id=device.get("name", ""), state_type="device"
)
if certainty:
result["certainty"] = certainty
results.append(result)
import json
return [
TextContent(
type="text",
text=f"Found {len(results)} matching host(s):\n\n{json.dumps(results, indent=2)}",
)
]