From ae6794b27ca914279cecee7b153cdb90984ab2b6 Mon Sep 17 00:00:00 2001 From: Luke Tucker Date: Tue, 28 Oct 2025 09:03:07 -0400 Subject: [PATCH 1/3] use app.model convention for object types --- netbox_types.py | 506 +++++++++++++++++++++++++++++++++++++++++ server.py | 265 +++++---------------- tests/test_brief.py | 12 +- tests/test_ordering.py | 12 +- tests/test_search.py | 41 ++-- 5 files changed, 590 insertions(+), 246 deletions(-) create mode 100644 netbox_types.py diff --git a/netbox_types.py b/netbox_types.py new file mode 100644 index 0000000..746b035 --- /dev/null +++ b/netbox_types.py @@ -0,0 +1,506 @@ +NETBOX_OBJECT_TYPES = { + "circuits.circuit": { + "name": "Circuit", + "endpoint": "circuits/circuits", + }, + "circuits.circuitgroup": { + "name": "CircuitGroup", + "endpoint": "circuits/circuit-groups", + }, + "circuits.circuitgroupassignment": { + "name": "CircuitGroupAssignment", + "endpoint": "circuits/circuit-group-assignments", + }, + "circuits.circuittermination": { + "name": "CircuitTermination", + "endpoint": "circuits/circuit-terminations", + }, + "circuits.circuittype": { + "name": "CircuitType", + "endpoint": "circuits/circuit-types", + }, + "circuits.provider": { + "name": "Provider", + "endpoint": "circuits/providers", + }, + "circuits.provideraccount": { + "name": "ProviderAccount", + "endpoint": "circuits/provider-accounts", + }, + "circuits.providernetwork": { + "name": "ProviderNetwork", + "endpoint": "circuits/provider-networks", + }, + "circuits.virtualcircuit": { + "name": "VirtualCircuit", + "endpoint": "circuits/virtual-circuits", + }, + "circuits.virtualcircuittermination": { + "name": "VirtualCircuitTermination", + "endpoint": "circuits/virtual-circuit-terminations", + }, + "circuits.virtualcircuittype": { + "name": "VirtualCircuitType", + "endpoint": "circuits/virtual-circuit-types", + }, + "core.datafile": { + "name": "DataFile", + "endpoint": "core/data-files", + }, + "core.datasource": { + "name": "DataSource", + "endpoint": "core/data-sources", + }, + "core.job": { + "name": "Job", + "endpoint": "core/jobs", + }, + "core.objectchange": { + "name": "ObjectChange", + "endpoint": "core/object-changes", + }, + "core.objecttype": { + "name": "ObjectType", + "endpoint": "extras/object-types", + }, + "dcim.cable": { + "name": "Cable", + "endpoint": "dcim/cables", + }, + "dcim.cabletermination": { + "name": "CableTermination", + "endpoint": "dcim/cable-terminations", + }, + "dcim.consoleport": { + "name": "ConsolePort", + "endpoint": "dcim/console-ports", + }, + "dcim.consoleporttemplate": { + "name": "ConsolePortTemplate", + "endpoint": "dcim/console-port-templates", + }, + "dcim.consoleserverport": { + "name": "ConsoleServerPort", + "endpoint": "dcim/console-server-ports", + }, + "dcim.consoleserverporttemplate": { + "name": "ConsoleServerPortTemplate", + "endpoint": "dcim/console-server-port-templates", + }, + "dcim.device": { + "name": "Device", + "endpoint": "dcim/devices", + }, + "dcim.devicebay": { + "name": "DeviceBay", + "endpoint": "dcim/device-bays", + }, + "dcim.devicebaytemplate": { + "name": "DeviceBayTemplate", + "endpoint": "dcim/device-bay-templates", + }, + "dcim.devicerole": { + "name": "DeviceRole", + "endpoint": "dcim/device-roles", + }, + "dcim.devicetype": { + "name": "DeviceType", + "endpoint": "dcim/device-types", + }, + "dcim.frontport": { + "name": "FrontPort", + "endpoint": "dcim/front-ports", + }, + "dcim.frontporttemplate": { + "name": "FrontPortTemplate", + "endpoint": "dcim/front-port-templates", + }, + "dcim.interface": { + "name": "Interface", + "endpoint": "dcim/interfaces", + }, + "dcim.interfacetemplate": { + "name": "InterfaceTemplate", + "endpoint": "dcim/interface-templates", + }, + "dcim.inventoryitem": { + "name": "InventoryItem", + "endpoint": "dcim/inventory-items", + }, + "dcim.inventoryitemrole": { + "name": "InventoryItemRole", + "endpoint": "dcim/inventory-item-roles", + }, + "dcim.inventoryitemtemplate": { + "name": "InventoryItemTemplate", + "endpoint": "dcim/inventory-item-templates", + }, + "dcim.location": { + "name": "Location", + "endpoint": "dcim/locations", + }, + "dcim.macaddress": { + "name": "MACAddress", + "endpoint": "dcim/mac-addresses", + }, + "dcim.manufacturer": { + "name": "Manufacturer", + "endpoint": "dcim/manufacturers", + }, + "dcim.module": { + "name": "Module", + "endpoint": "dcim/modules", + }, + "dcim.modulebay": { + "name": "ModuleBay", + "endpoint": "dcim/module-bays", + }, + "dcim.modulebaytemplate": { + "name": "ModuleBayTemplate", + "endpoint": "dcim/module-bay-templates", + }, + "dcim.moduletype": { + "name": "ModuleType", + "endpoint": "dcim/module-types", + }, + "dcim.moduletypeprofile": { + "name": "ModuleTypeProfile", + "endpoint": "dcim/module-type-profiles", + }, + "dcim.platform": { + "name": "Platform", + "endpoint": "dcim/platforms", + }, + "dcim.powerfeed": { + "name": "PowerFeed", + "endpoint": "dcim/power-feeds", + }, + "dcim.poweroutlet": { + "name": "PowerOutlet", + "endpoint": "dcim/power-outlets", + }, + "dcim.poweroutlettemplate": { + "name": "PowerOutletTemplate", + "endpoint": "dcim/power-outlet-templates", + }, + "dcim.powerpanel": { + "name": "PowerPanel", + "endpoint": "dcim/power-panels", + }, + "dcim.powerport": { + "name": "PowerPort", + "endpoint": "dcim/power-ports", + }, + "dcim.powerporttemplate": { + "name": "PowerPortTemplate", + "endpoint": "dcim/power-port-templates", + }, + "dcim.rack": { + "name": "Rack", + "endpoint": "dcim/racks", + }, + "dcim.rackreservation": { + "name": "RackReservation", + "endpoint": "dcim/rack-reservations", + }, + "dcim.rackrole": { + "name": "RackRole", + "endpoint": "dcim/rack-roles", + }, + "dcim.racktype": { + "name": "RackType", + "endpoint": "dcim/rack-types", + }, + "dcim.rearport": { + "name": "RearPort", + "endpoint": "dcim/rear-ports", + }, + "dcim.rearporttemplate": { + "name": "RearPortTemplate", + "endpoint": "dcim/rear-port-templates", + }, + "dcim.region": { + "name": "Region", + "endpoint": "dcim/regions", + }, + "dcim.site": { + "name": "Site", + "endpoint": "dcim/sites", + }, + "dcim.sitegroup": { + "name": "SiteGroup", + "endpoint": "dcim/site-groups", + }, + "dcim.virtualchassis": { + "name": "VirtualChassis", + "endpoint": "dcim/virtual-chassis", + }, + "dcim.virtualdevicecontext": { + "name": "VirtualDeviceContext", + "endpoint": "dcim/virtual-device-contexts", + }, + "extras.bookmark": { + "name": "Bookmark", + "endpoint": "extras/bookmarks", + }, + "extras.configcontext": { + "name": "ConfigContext", + "endpoint": "extras/config-contexts", + }, + "extras.configtemplate": { + "name": "ConfigTemplate", + "endpoint": "extras/config-templates", + }, + "extras.customfield": { + "name": "CustomField", + "endpoint": "extras/custom-fields", + }, + "extras.customfieldchoiceset": { + "name": "CustomFieldChoiceSet", + "endpoint": "extras/custom-field-choice-sets", + }, + "extras.customlink": { + "name": "CustomLink", + "endpoint": "extras/custom-links", + }, + "extras.eventrule": { + "name": "EventRule", + "endpoint": "extras/event-rules", + }, + "extras.exporttemplate": { + "name": "ExportTemplate", + "endpoint": "extras/export-templates", + }, + "extras.imageattachment": { + "name": "ImageAttachment", + "endpoint": "extras/image-attachments", + }, + "extras.journalentry": { + "name": "JournalEntry", + "endpoint": "extras/journal-entries", + }, + "extras.notification": { + "name": "Notification", + "endpoint": "extras/notifications", + }, + "extras.notificationgroup": { + "name": "NotificationGroup", + "endpoint": "extras/notification-groups", + }, + "extras.savedfilter": { + "name": "SavedFilter", + "endpoint": "extras/saved-filters", + }, + "extras.script": { + "name": "Script", + "endpoint": "extras/scripts", + }, + "extras.subscription": { + "name": "Subscription", + "endpoint": "extras/subscriptions", + }, + "extras.tableconfig": { + "name": "TableConfig", + "endpoint": "extras/table-configs", + }, + "extras.tag": { + "name": "Tag", + "endpoint": "extras/tags", + }, + "extras.taggeditem": { + "name": "TaggedItem", + "endpoint": "extras/tagged-objects", + }, + "extras.webhook": { + "name": "Webhook", + "endpoint": "extras/webhooks", + }, + "ipam.aggregate": { + "name": "Aggregate", + "endpoint": "ipam/aggregates", + }, + "ipam.asn": { + "name": "ASN", + "endpoint": "ipam/asns", + }, + "ipam.asnrange": { + "name": "ASNRange", + "endpoint": "ipam/asn-ranges", + }, + "ipam.fhrpgroup": { + "name": "FHRPGroup", + "endpoint": "ipam/fhrp-groups", + }, + "ipam.fhrpgroupassignment": { + "name": "FHRPGroupAssignment", + "endpoint": "ipam/fhrp-group-assignments", + }, + "ipam.ipaddress": { + "name": "IPAddress", + "endpoint": "ipam/ip-addresses", + }, + "ipam.iprange": { + "name": "IPRange", + "endpoint": "ipam/ip-ranges", + }, + "ipam.prefix": { + "name": "Prefix", + "endpoint": "ipam/prefixes", + }, + "ipam.rir": { + "name": "RIR", + "endpoint": "ipam/rirs", + }, + "ipam.role": { + "name": "Role", + "endpoint": "ipam/roles", + }, + "ipam.routetarget": { + "name": "RouteTarget", + "endpoint": "ipam/route-targets", + }, + "ipam.service": { + "name": "Service", + "endpoint": "ipam/services", + }, + "ipam.servicetemplate": { + "name": "ServiceTemplate", + "endpoint": "ipam/service-templates", + }, + "ipam.vlan": { + "name": "VLAN", + "endpoint": "ipam/vlans", + }, + "ipam.vlangroup": { + "name": "VLANGroup", + "endpoint": "ipam/vlan-groups", + }, + "ipam.vlantranslationpolicy": { + "name": "VLANTranslationPolicy", + "endpoint": "ipam/vlan-translation-policies", + }, + "ipam.vlantranslationrule": { + "name": "VLANTranslationRule", + "endpoint": "ipam/vlan-translation-rules", + }, + "ipam.vrf": { + "name": "VRF", + "endpoint": "ipam/vrfs", + }, + "tenancy.contact": { + "name": "Contact", + "endpoint": "tenancy/contacts", + }, + "tenancy.contactassignment": { + "name": "ContactAssignment", + "endpoint": "tenancy/contact-assignments", + }, + "tenancy.contactgroup": { + "name": "ContactGroup", + "endpoint": "tenancy/contact-groups", + }, + "tenancy.contactrole": { + "name": "ContactRole", + "endpoint": "tenancy/contact-roles", + }, + "tenancy.tenant": { + "name": "Tenant", + "endpoint": "tenancy/tenants", + }, + "tenancy.tenantgroup": { + "name": "TenantGroup", + "endpoint": "tenancy/tenant-groups", + }, + "users.group": { + "name": "Group", + "endpoint": "users/groups", + }, + "users.objectpermission": { + "name": "ObjectPermission", + "endpoint": "users/permissions", + }, + "users.token": { + "name": "Token", + "endpoint": "users/tokens", + }, + "users.user": { + "name": "User", + "endpoint": "users/users", + }, + "virtualization.cluster": { + "name": "Cluster", + "endpoint": "virtualization/clusters", + }, + "virtualization.clustergroup": { + "name": "ClusterGroup", + "endpoint": "virtualization/cluster-groups", + }, + "virtualization.clustertype": { + "name": "ClusterType", + "endpoint": "virtualization/cluster-types", + }, + "virtualization.virtualdisk": { + "name": "VirtualDisk", + "endpoint": "virtualization/virtual-disks", + }, + "virtualization.virtualmachine": { + "name": "VirtualMachine", + "endpoint": "virtualization/virtual-machines", + }, + "virtualization.vminterface": { + "name": "VMInterface", + "endpoint": "virtualization/interfaces", + }, + "vpn.ikepolicy": { + "name": "IKEPolicy", + "endpoint": "vpn/ike-policies", + }, + "vpn.ikeproposal": { + "name": "IKEProposal", + "endpoint": "vpn/ike-proposals", + }, + "vpn.ipsecpolicy": { + "name": "IPSecPolicy", + "endpoint": "vpn/ipsec-policies", + }, + "vpn.ipsecprofile": { + "name": "IPSecProfile", + "endpoint": "vpn/ipsec-profiles", + }, + "vpn.ipsecproposal": { + "name": "IPSecProposal", + "endpoint": "vpn/ipsec-proposals", + }, + "vpn.l2vpn": { + "name": "L2VPN", + "endpoint": "vpn/l2vpns", + }, + "vpn.l2vpntermination": { + "name": "L2VPNTermination", + "endpoint": "vpn/l2vpn-terminations", + }, + "vpn.tunnel": { + "name": "Tunnel", + "endpoint": "vpn/tunnels", + }, + "vpn.tunnelgroup": { + "name": "TunnelGroup", + "endpoint": "vpn/tunnel-groups", + }, + "vpn.tunneltermination": { + "name": "TunnelTermination", + "endpoint": "vpn/tunnel-terminations", + }, + "wireless.wirelesslan": { + "name": "WirelessLAN", + "endpoint": "wireless/wireless-lans", + }, + "wireless.wirelesslangroup": { + "name": "WirelessLANGroup", + "endpoint": "wireless/wireless-lan-groups", + }, + "wireless.wirelesslink": { + "name": "WirelessLink", + "endpoint": "wireless/wireless-links", + }, +} diff --git a/server.py b/server.py index 2dbd7f1..1ab0370 100644 --- a/server.py +++ b/server.py @@ -8,6 +8,7 @@ from config import Settings, configure_logging from netbox_client import NetBoxRestClient +from netbox_types import NETBOX_OBJECT_TYPES def parse_cli_args() -> dict[str, Any]: @@ -97,107 +98,16 @@ def parse_cli_args() -> dict[str, Any]: return overlay -# Mapping of simple object names to API endpoints -NETBOX_OBJECT_TYPES = { - # DCIM (Device and Infrastructure) - "cables": "dcim/cables", - "console-ports": "dcim/console-ports", - "console-server-ports": "dcim/console-server-ports", - "devices": "dcim/devices", - "device-bays": "dcim/device-bays", - "device-roles": "dcim/device-roles", - "device-types": "dcim/device-types", - "front-ports": "dcim/front-ports", - "interfaces": "dcim/interfaces", - "inventory-items": "dcim/inventory-items", - "locations": "dcim/locations", - "manufacturers": "dcim/manufacturers", - "modules": "dcim/modules", - "module-bays": "dcim/module-bays", - "module-types": "dcim/module-types", - "platforms": "dcim/platforms", - "power-feeds": "dcim/power-feeds", - "power-outlets": "dcim/power-outlets", - "power-panels": "dcim/power-panels", - "power-ports": "dcim/power-ports", - "racks": "dcim/racks", - "rack-reservations": "dcim/rack-reservations", - "rack-roles": "dcim/rack-roles", - "regions": "dcim/regions", - "sites": "dcim/sites", - "site-groups": "dcim/site-groups", - "virtual-chassis": "dcim/virtual-chassis", - # IPAM (IP Address Management) - "asns": "ipam/asns", - "asn-ranges": "ipam/asn-ranges", - "aggregates": "ipam/aggregates", - "fhrp-groups": "ipam/fhrp-groups", - "ip-addresses": "ipam/ip-addresses", - "ip-ranges": "ipam/ip-ranges", - "prefixes": "ipam/prefixes", - "rirs": "ipam/rirs", - "roles": "ipam/roles", - "route-targets": "ipam/route-targets", - "services": "ipam/services", - "vlans": "ipam/vlans", - "vlan-groups": "ipam/vlan-groups", - "vrfs": "ipam/vrfs", - # Circuits - "circuits": "circuits/circuits", - "circuit-types": "circuits/circuit-types", - "circuit-terminations": "circuits/circuit-terminations", - "providers": "circuits/providers", - "provider-networks": "circuits/provider-networks", - # Virtualization - "clusters": "virtualization/clusters", - "cluster-groups": "virtualization/cluster-groups", - "cluster-types": "virtualization/cluster-types", - "virtual-machines": "virtualization/virtual-machines", - "vm-interfaces": "virtualization/interfaces", - # Tenancy - "tenants": "tenancy/tenants", - "tenant-groups": "tenancy/tenant-groups", - "contacts": "tenancy/contacts", - "contact-groups": "tenancy/contact-groups", - "contact-roles": "tenancy/contact-roles", - # VPN - "ike-policies": "vpn/ike-policies", - "ike-proposals": "vpn/ike-proposals", - "ipsec-policies": "vpn/ipsec-policies", - "ipsec-profiles": "vpn/ipsec-profiles", - "ipsec-proposals": "vpn/ipsec-proposals", - "l2vpns": "vpn/l2vpns", - "tunnels": "vpn/tunnels", - "tunnel-groups": "vpn/tunnel-groups", - # Wireless - "wireless-lans": "wireless/wireless-lans", - "wireless-lan-groups": "wireless/wireless-lan-groups", - "wireless-links": "wireless/wireless-links", - # Core (introduced in NetBox v3.5) - "data-files": "core/data-files", - "data-sources": "core/data-sources", - "jobs": "core/jobs", - # Extras - "config-contexts": "extras/config-contexts", - "custom-fields": "extras/custom-fields", - "export-templates": "extras/export-templates", - "image-attachments": "extras/image-attachments", - "saved-filters": "extras/saved-filters", - "scripts": "extras/scripts", - "tags": "extras/tags", - "webhooks": "extras/webhooks", -} - # Default object types for global search DEFAULT_SEARCH_TYPES = [ - "devices", # Most common search target - "sites", # Site names frequently searched - "ip-addresses", # IP searches very common - "interfaces", # Interface names/descriptions - "racks", # Rack identifiers - "vlans", # VLAN names/IDs - "circuits", # Circuit identifiers - "virtual-machines", # VM names + "dcim.device", # Most common search target + "dcim.site", # Site names frequently searched + "ipam.ipaddress", # IP searches very common + "dcim.interface", # Interface names/descriptions + "dcim.rack", # Rack identifiers + "ipam.vlan", # VLAN names/IDs + "circuits.circuit", # Circuit identifiers + "virtualization.virtualmachine", # VM names ] mcp = FastMCP("NetBox") @@ -264,17 +174,8 @@ def validate_filters(filters: dict) -> None: ) -@mcp.tool -def netbox_get_objects( - object_type: str, - filters: dict, - fields: list[str] | None = None, - brief: bool = False, - limit: Annotated[int, Field(default=5, ge=1, le=100)] = 5, - offset: Annotated[int, Field(default=0, ge=0)] = 0, - ordering: str | list[str] | None = None, -): - """ +@mcp.tool( + description=""" Get objects from NetBox based on their type and filters Args: @@ -343,94 +244,25 @@ def netbox_get_objects( Valid object_type values: - DCIM (Device and Infrastructure): - - cables - - console-ports - - console-server-ports - - devices - - device-bays - - device-roles - - device-types - - front-ports - - interfaces - - inventory-items - - locations - - manufacturers - - modules - - module-bays - - module-types - - platforms - - power-feeds - - power-outlets - - power-panels - - power-ports - - racks - - rack-reservations - - rack-roles - - regions - - sites - - site-groups - - virtual-chassis - - IPAM (IP Address Management): - - asns - - asn-ranges - - aggregates - - fhrp-groups - - ip-addresses - - ip-ranges - - prefixes - - rirs - - roles - - route-targets - - services - - vlans - - vlan-groups - - vrfs - - Circuits: - - circuits - - circuit-types - - circuit-terminations - - providers - - provider-networks - - Virtualization: - - clusters - - cluster-groups - - cluster-types - - virtual-machines - - vm-interfaces - - Tenancy: - - tenants - - tenant-groups - - contacts - - contact-groups - - contact-roles - - VPN: - - ike-policies - - ike-proposals - - ipsec-policies - - ipsec-profiles - - ipsec-proposals - - l2vpns - - tunnels - - tunnel-groups - - Wireless: - - wireless-lans - - wireless-lan-groups - - wireless-links - - Core (NetBox v3.5+): - - data-files - - data-sources - - jobs + """ + + "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys())) + + """ See NetBox API documentation for filtering options for each object type. """ +) +def netbox_get_objects( + object_type: str, + filters: dict, + fields: list[str] | None = None, + brief: bool = False, + limit: Annotated[int, Field(default=5, ge=1, le=100)] = 5, + offset: Annotated[int, Field(default=0, ge=0)] = 0, + ordering: str | list[str] | None = None, +): + """ + Get objects from NetBox based on their type and filters + """ # Validate object_type exists in mapping if object_type not in NETBOX_OBJECT_TYPES: valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys())) @@ -440,7 +272,7 @@ def netbox_get_objects( validate_filters(filters) # Get API endpoint from mapping - endpoint = NETBOX_OBJECT_TYPES[object_type] + endpoint = _endpoint_for_type(object_type) # Build params with pagination (parameters override filters dict) params = filters.copy() @@ -474,7 +306,7 @@ def netbox_get_object_by_id( Get detailed information about a specific NetBox object by its ID. Args: - object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses") + object_type: String representing the NetBox object type (e.g. "dcim.device", "ipam.ipaddress") object_id: The numeric ID of the object fields: Optional list of specific fields to return **IMPORTANT: ALWAYS USE THIS PARAMETER TO MINIMIZE TOKEN USAGE** @@ -502,7 +334,7 @@ def netbox_get_object_by_id( raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}") # Get API endpoint from mapping - endpoint = f"{NETBOX_OBJECT_TYPES[object_type]}/{object_id}" + endpoint = f"{_endpoint_for_type(object_type)}/{object_id}" params = {} if fields: @@ -572,14 +404,8 @@ def netbox_get_changelogs(filters: dict): return netbox.get(endpoint, params=filters) -@mcp.tool -def netbox_search_objects( - query: str, - object_types: list[str] | None = None, - fields: list[str] | None = None, - limit: Annotated[int, Field(default=5, ge=1, le=100)] = 5, -) -> dict[str, list[dict]]: - """ +@mcp.tool( + description=""" Perform global search across NetBox infrastructure. Searches names, descriptions, IP addresses, serial numbers, asset tags, @@ -589,9 +415,8 @@ def netbox_search_objects( query: Search term (device names, IPs, serial numbers, hostnames, site names) Examples: 'switch01', '192.168.1.1', 'NYC-DC1', 'SN123456' object_types: Limit search to specific types (optional) - Default: ['devices', 'sites', 'ip-addresses', 'interfaces', - 'racks', 'vlans', 'circuits', 'virtual-machines'] - Examples: ['devices', 'ip-addresses', 'sites'] + Default: [""" + "', '".join(DEFAULT_SEARCH_TYPES) + """] + Examples: ['dcim.device', 'ipam.ipaddress', 'dcim.site'] fields: Optional list of specific fields to return (reduces response size) IT IS STRONGLY RECOMMENDED TO USE THIS PARAMETER TO MINIMIZE TOKEN USAGE. - None or [] = returns all fields (no filtering) - ['id', 'name'] = returns only specified fields @@ -607,25 +432,35 @@ def netbox_search_objects( # Search for anything matching "switch" results = netbox_search_objects('switch') # Returns: { - # 'devices': [{'id': 1, 'name': 'switch-01', ...}], - # 'sites': [], + # 'dcim.device': [{'id': 1, 'name': 'switch-01', ...}], + # 'dcim.site': [], # ... # } # Search for IP address results = netbox_search_objects('192.168.1.100') # Returns: { - # 'ip-addresses': [{'id': 42, 'address': '192.168.1.100/24', ...}], + # 'ipam.ipaddress': [{'id': 42, 'address': '192.168.1.100/24', ...}], # ... # } # Limit search to specific types with field projection results = netbox_search_objects( 'NYC', - object_types=['sites', 'locations'], + object_types=['dcim.site', 'dcim.location'], fields=['id', 'name', 'status'] ) """ +) +def netbox_search_objects( + query: str, + object_types: list[str] | None = None, + fields: list[str] | None = None, + limit: Annotated[int, Field(default=5, ge=1, le=100)] = 5, +) -> dict[str, list[dict]]: + """ + Perform global search across NetBox infrastructure. + """ if object_types is None: search_types = DEFAULT_SEARCH_TYPES else: @@ -647,7 +482,7 @@ def netbox_search_objects( for obj_type in search_types: try: response = netbox.get( - NETBOX_OBJECT_TYPES[obj_type], + _endpoint_for_type(obj_type), params={ "q": query, "limit": limit, @@ -663,6 +498,8 @@ def netbox_search_objects( return results +def _endpoint_for_type(object_type): + return NETBOX_OBJECT_TYPES[object_type]['endpoint'] if __name__ == "__main__": cli_overlay: dict[str, Any] = parse_cli_args() diff --git a/tests/test_brief.py b/tests/test_brief.py index 0d83b6b..92eefe0 100644 --- a/tests/test_brief.py +++ b/tests/test_brief.py @@ -10,7 +10,7 @@ def test_brief_false_omits_parameter_get_objects(mock_netbox): """When brief=False (default), should not include brief in API params for netbox_get_objects.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, brief=False) + netbox_get_objects.fn(object_type="dcim.site", filters={}, brief=False) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -24,7 +24,7 @@ def test_brief_default_omits_parameter_get_objects(mock_netbox): """When brief not specified (uses default False), should not include brief in API params.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}) + netbox_get_objects.fn(object_type="dcim.site", filters={}) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -38,7 +38,7 @@ def test_brief_true_includes_parameter_get_objects(mock_netbox): """When brief=True, should pass 'brief': '1' to API params for netbox_get_objects.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, brief=True) + netbox_get_objects.fn(object_type="dcim.site", filters={}, brief=True) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -51,7 +51,7 @@ def test_brief_false_omits_parameter_get_by_id(mock_netbox): """When brief=False (default), should not include brief in API params for netbox_get_object_by_id.""" mock_netbox.get.return_value = {"id": 1, "name": "Test Site"} - netbox_get_object_by_id.fn(object_type="sites", object_id=1, brief=False) + netbox_get_object_by_id.fn(object_type="dcim.site", object_id=1, brief=False) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -65,7 +65,7 @@ def test_brief_default_omits_parameter_get_by_id(mock_netbox): """When brief not specified (uses default False), should not include brief in API params.""" mock_netbox.get.return_value = {"id": 1, "name": "Test Site"} - netbox_get_object_by_id.fn(object_type="sites", object_id=1) + netbox_get_object_by_id.fn(object_type="dcim.site", object_id=1) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -79,7 +79,7 @@ def test_brief_true_includes_parameter_get_by_id(mock_netbox): """When brief=True, should pass 'brief': '1' to API params for netbox_get_object_by_id.""" mock_netbox.get.return_value = {"id": 1, "url": "http://example.com/api/dcim/sites/1/"} - netbox_get_object_by_id.fn(object_type="sites", object_id=1, brief=True) + netbox_get_object_by_id.fn(object_type="dcim.site", object_id=1, brief=True) call_args = mock_netbox.get.call_args params = call_args[1]["params"] diff --git a/tests/test_ordering.py b/tests/test_ordering.py index f349796..02a53e5 100644 --- a/tests/test_ordering.py +++ b/tests/test_ordering.py @@ -27,7 +27,7 @@ def test_ordering_none_omits_parameter(mock_netbox): """When ordering=None, should not include ordering in API params.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, ordering=None) + netbox_get_objects.fn(object_type="dcim.site", filters={}, ordering=None) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -41,7 +41,7 @@ def test_ordering_empty_string_omits_parameter(mock_netbox): """When ordering='', should not include ordering in API params.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, ordering="") + netbox_get_objects.fn(object_type="dcim.site", filters={}, ordering="") call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -55,7 +55,7 @@ def test_ordering_single_field_ascending(mock_netbox): """When ordering='name', should pass 'name' to API params.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, ordering="name") + netbox_get_objects.fn(object_type="dcim.site", filters={}, ordering="name") call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -68,7 +68,7 @@ def test_ordering_single_field_descending(mock_netbox): """When ordering='-id', should pass '-id' to API params.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, ordering="-id") + netbox_get_objects.fn(object_type="dcim.site", filters={}, ordering="-id") call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -81,7 +81,7 @@ def test_ordering_multiple_fields_as_list(mock_netbox): """When ordering=['facility', '-name'], should pass comma-separated string.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, ordering=["facility", "-name"]) + netbox_get_objects.fn(object_type="dcim.site", filters={}, ordering=["facility", "-name"]) call_args = mock_netbox.get.call_args params = call_args[1]["params"] @@ -95,7 +95,7 @@ def test_ordering_empty_list_omits_parameter(mock_netbox): """When ordering=[], should not include ordering in API params.""" mock_netbox.get.return_value = {"count": 0, "results": [], "next": None, "previous": None} - netbox_get_objects.fn(object_type="sites", filters={}, ordering=[]) + netbox_get_objects.fn(object_type="dcim.site", filters={}, ordering=[]) call_args = mock_netbox.get.call_args params = call_args[1]["params"] diff --git a/tests/test_search.py b/tests/test_search.py index 8fe4602..96ea7b1 100644 --- a/tests/test_search.py +++ b/tests/test_search.py @@ -5,7 +5,8 @@ import pytest from pydantic import TypeAdapter, ValidationError -from server import NETBOX_OBJECT_TYPES, netbox_search_objects +from netbox_types import NETBOX_OBJECT_TYPES +from server import netbox_search_objects # ============================================================================ # Parameter Validation Tests @@ -68,11 +69,11 @@ def test_custom_object_types_limits_search_scope(mock_netbox): "results": [], } - result = netbox_search_objects.fn(query="test", object_types=["devices", "sites"]) + result = netbox_search_objects.fn(query="test", object_types=["dcim.device", "dcim.site"]) # Should only search specified types assert mock_netbox.get.call_count == 2 - assert set(result.keys()) == {"devices", "sites"} + assert set(result.keys()) == {"dcim.device", "dcim.site"} # ============================================================================ @@ -91,7 +92,7 @@ def test_field_projection_applied_to_queries(mock_netbox): } netbox_search_objects.fn( - query="test", object_types=["devices", "sites"], fields=["id", "name"] + query="test", object_types=["dcim.device", "dcim.site"], fields=["id", "name"] ) # All calls should include fields parameter @@ -122,16 +123,16 @@ def mock_get_side_effect(endpoint, params): mock_netbox.get.side_effect = mock_get_side_effect result = netbox_search_objects.fn( - query="test", object_types=["devices", "sites", "racks"] + query="test", object_types=["dcim.device", "dcim.site", "dcim.rack"] ) # All types present - assert set(result.keys()) == {"devices", "sites", "racks"} + assert set(result.keys()) == {"dcim.device", "dcim.site", "dcim.rack"} # Populated results contain data - assert result["devices"] == [{"id": 1, "name": "device01"}] + assert result["dcim.device"] == [{"id": 1, "name": "device01"}] # Empty results are empty lists, not missing keys - assert result["sites"] == [] - assert result["racks"] == [] + assert result["dcim.site"] == [] + assert result["dcim.rack"] == [] # ============================================================================ @@ -157,12 +158,12 @@ def mock_get_side_effect(endpoint, params): mock_netbox.get.side_effect = mock_get_side_effect - result = netbox_search_objects.fn(query="test", object_types=["devices", "sites"]) + result = netbox_search_objects.fn(query="test", object_types=["dcim.device", "dcim.site"]) # Should continue despite error - assert result["sites"] == [{"id": 1, "name": "site01"}] + assert result["dcim.site"] == [{"id": 1, "name": "site01"}] # Failed type has empty list - assert result["devices"] == [] + assert result["dcim.device"] == [] # ============================================================================ @@ -181,7 +182,7 @@ def test_api_parameters_passed_correctly(mock_netbox): } netbox_search_objects.fn( - query="switch01", object_types=["devices"], fields=["id"], limit=25 + query="switch01", object_types=["dcim.device"], fields=["id"], limit=25 ) call_args = mock_netbox.get.call_args @@ -202,11 +203,11 @@ def test_uses_correct_api_endpoints(mock_netbox): "results": [], } - netbox_search_objects.fn(query="test", object_types=["devices", "ip-addresses"]) + netbox_search_objects.fn(query="test", object_types=["dcim.device", "ipam.ipaddress"]) called_endpoints = [call[0][0] for call in mock_netbox.get.call_args_list] - assert NETBOX_OBJECT_TYPES["devices"] in called_endpoints - assert NETBOX_OBJECT_TYPES["ip-addresses"] in called_endpoints + assert NETBOX_OBJECT_TYPES["dcim.device"]['endpoint'] in called_endpoints + assert NETBOX_OBJECT_TYPES["ipam.ipaddress"]['endpoint'] in called_endpoints # ============================================================================ @@ -239,14 +240,14 @@ def test_extracts_results_from_paginated_response(mock_netbox): ], } - result = netbox_search_objects.fn(query="test", object_types=["devices"]) + result = netbox_search_objects.fn(query="test", object_types=["dcim.device"]) # Should return dict with object type as key - assert "devices" in result + assert "dcim.device" in result # Value should be a list (array), not a dict - assert isinstance(result["devices"], list) + assert isinstance(result["dcim.device"], list) # Should contain just the results, not the paginated response wrapper - assert result["devices"] == [ + assert result["dcim.device"] == [ {"id": 1, "name": "device01"}, {"id": 2, "name": "device02"}, ] From 0e74479adf428a95fb85e3a7557311a5923585b8 Mon Sep 17 00:00:00 2001 From: Luke Tucker Date: Tue, 28 Oct 2025 09:33:04 -0400 Subject: [PATCH 2/3] add comment --- server.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/server.py b/server.py index 1ab0370..e86afcb 100644 --- a/server.py +++ b/server.py @@ -498,7 +498,11 @@ def netbox_search_objects( return results -def _endpoint_for_type(object_type): +def _endpoint_for_type(object_type: str) -> str: + """ + Returns partial API endpoint prefix for the given object type. + e.g., "dcim.device" -> "dcim/devices" + """ return NETBOX_OBJECT_TYPES[object_type]['endpoint'] if __name__ == "__main__": From 66adbd83660d317717570ff5a59020fab179bdb9 Mon Sep 17 00:00:00 2001 From: Luke Tucker Date: Tue, 28 Oct 2025 10:14:46 -0400 Subject: [PATCH 3/3] fix old object type examples --- server.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/server.py b/server.py index e86afcb..3a5c653 100644 --- a/server.py +++ b/server.py @@ -179,7 +179,7 @@ def validate_filters(filters: dict) -> None: Get objects from NetBox based on their type and filters Args: - object_type: String representing the NetBox object type (e.g. "devices", "ip-addresses") + object_type: String representing the NetBox object type (e.g. "dcim.device", "ipam.ipaddress") filters: dict of filters to apply to the API call based on the NetBox API filtering options FILTER RULES: @@ -191,8 +191,8 @@ def validate_filters(filters: dict) -> None: empty, regex, iregex, lt, lte, gt, gte, in Two-step pattern for cross-relationship queries: - sites = netbox_get_objects('sites', {'name': 'NYC'}) - netbox_get_objects('devices', {'site_id': sites[0]['id']}) + sites = netbox_get_objects('dcim.site', {'name': 'NYC'}) + netbox_get_objects('dcim.device', {'site_id': sites[0]['id']}) fields: Optional list of specific fields to return **IMPORTANT: ALWAYS USE THIS PARAMETER TO MINIMIZE TOKEN USAGE**