NetBox MCP Server

#!.venv/bin/python """ NetBox MCP + Internal LLM ChatBot Connects to your NetBox MCP server (via mcpo) and your internal OpenAI-compatible LLM API. """ import json import sys import signal import asyncio import requests from typing import Dict, Any, List from openai import OpenAI from dotenv import load_dotenv import os class NetBoxLLMChatBot: def __init__(self, mcpo_url: str, mcpo_api_key: str, llm_url: str, llm_api_key: str, llm_model: str = "gpt-4"): """ Initialize the chatbot. Args: mcpo_url: Your mcpo server URL (e.g., "http://localhost:8000") mcpo_api_key: Your API key for mcpo server llm_url: Your internal LLM API URL (e.g., "http://localhost:11434/v1") llm_api_key: Your LLM API key llm_model: Model name to use """ self.mcpo_url = mcpo_url.rstrip('/') self.mcpo_api_key = mcpo_api_key self.llm_url = llm_url.rstrip('/') self.llm_api_key = llm_api_key self.llm_model = llm_model # Initialize OpenAI client self.openai_client = OpenAI( api_key=llm_api_key, base_url=llm_url ) # Headers for mcpo server self.mcpo_headers = { "Authorization": f"Bearer {mcpo_api_key}", "Content-Type": "application/json" } # NetBox tool schemas for the LLM self.tools = [ { "type": "function", "function": { "name": "netbox_get_objects", "description": "Get objects from NetBox based on their type and filters. Use this to retrieve devices, sites, IP addresses, VLANs, etc.", "parameters": { "type": "object", "properties": { "object_type": { "type": "string", "description": "Type of NetBox object (e.g., 'devices', 'ip-addresses', 'sites', 'vlans', 'interfaces', 'racks', 'device-types', 'manufacturers', 'tenants', 'circuits', 'virtual-machines', 'prefixes', 'services')" }, "filters": { "type": "object", "description": "Filters to apply to the API call. Common filters include 'site', 'status', 'limit', 'name', etc." } }, "required": ["object_type"] } } }, { "type": "function", "function": { "name": "netbox_get_object_by_id", "description": "Get detailed information about a specific NetBox object by its ID.", "parameters": { "type": "object", "properties": { "object_type": { "type": "string", "description": "Type of NetBox object" }, "object_id": { "type": "integer", "description": "The numeric ID of the object" } }, "required": ["object_type", "object_id"] } } }, { "type": "function", "function": { "name": "netbox_get_changelogs", "description": "Get object change records (changelogs) from NetBox based on filters. Use this to see audit trails and recent changes.", "parameters": { "type": "object", "properties": { "filters": { "type": "object", "description": "Filters to apply to the API call. Common filters include 'limit', 'time_after', 'time_before', 'action', 'user', 'changed_object_id'" } }, "required": [] } } } ] def call_netbox_tool(self, tool_name: str, arguments: Dict) -> Dict: """Call a NetBox MCP tool via the mcpo HTTP interface.""" try: response = requests.post( f"{self.mcpo_url}/netbox/{tool_name}", headers=self.mcpo_headers, json=arguments ) if response.status_code == 200: return response.json() else: print(f"❌ NetBox Tool Call Error: {response.status_code} - {response.text}") return {"error": f"Tool call failed: {response.status_code} - {response.text}"} except Exception as e: print(f"❌ NetBox Tool Call Exception: {e}") return {"error": f"Tool call error: {e}"} def call_llm(self, messages: List[Dict], tools: List[Dict] = None, force_tool_call: str = None) -> Dict: """Call the internal LLM API using OpenAI client.""" try: # Prepare the request parameters request_params = { "model": self.llm_model, "messages": messages, "temperature": 0.1, "stream": False } if tools: request_params["tools"] = tools if force_tool_call: request_params["tool_choice"] = {"type": "function", "function": {"name": force_tool_call}} else: request_params["tool_choice"] = "auto" # Make the API call using OpenAI client response = self.openai_client.chat.completions.create(**request_params) # Convert response to dict format return { "choices": [{ "message": { "role": response.choices[0].message.role, "content": response.choices[0].message.content, "tool_calls": response.choices[0].message.tool_calls } }] } except Exception as e: print(f"❌ LLM Request Exception: {e}") return {"error": f"LLM request error: {e}"} def format_netbox_result(self, tool_name: str, result: Dict) -> str: """Format NetBox tool result for LLM context.""" if "error" in result: return f"Error: {result['error']}" # Convert result to a readable string if isinstance(result, list): if len(result) == 0: return "No results found." formatted = f"Found {len(result)} items:\n" for i, item in enumerate(result[:10]): # Limit to 10 items name = item.get('name', item.get('display', item.get('id', 'Unknown'))) formatted += f"{i+1}. {name}\n" if len(result) > 10: formatted += f"... and {len(result) - 10} more items\n" return formatted elif isinstance(result, dict): # Single object result formatted = "Object details:\n" if 'name' in result: formatted += f"Name: {result['name']}\n" if 'display' in result: formatted += f"Display: {result['display']}\n" if 'id' in result: formatted += f"ID: {result['id']}\n" if 'status' in result: formatted += f"Status: {result['status'].get('value', 'Unknown')}\n" return formatted return str(result) def run_conversation(self, user_input: str, conversation_history: List[Dict]) -> tuple: """Run a conversation turn with the LLM.""" # Add user message to history conversation_history.append({"role": "user", "content": user_input}) # Check if this looks like a NetBox query and force function calling netbox_keywords = ['device', 'devices', 'site', 'sites', 'ip', 'address', 'vlan', 'interface', 'rack', 'change', 'log', 'netbox'] should_force_function = any(keyword in user_input.lower() for keyword in netbox_keywords) # Get LLM response with tool calling llm_response = self.call_llm(conversation_history, self.tools) if "error" in llm_response: print(f"❌ LLM Error Details: {llm_response['error']}") return f"❌ LLM Error: {llm_response['error']}", conversation_history message = llm_response.get('choices', [{}])[0].get('message', {}) # Check if LLM wants to call a tool if message.get('tool_calls'): tool_call = message['tool_calls'][0] # Get first tool call function_name = tool_call.function.name try: arguments = json.loads(tool_call.function.arguments) except json.JSONDecodeError as e: print(f"❌ JSON Decode Error: {e}") return "❌ Invalid tool arguments", conversation_history # Call the NetBox tool print(f"🔧 Calling NetBox tool: {function_name}") tool_result = self.call_netbox_tool(function_name, arguments) # Format the result formatted_result = self.format_netbox_result(function_name, tool_result) print(f"🔍 Tool result: {formatted_result[:200]}..." if len(str(formatted_result)) > 200 else f"🔍 Tool result: {formatted_result}") # Add tool result to conversation conversation_history.append({ "role": "tool", "tool_call_id": tool_call.id, "content": formatted_result }) # Get follow-up response from LLM follow_up = self.call_llm(conversation_history, self.tools) if "error" in follow_up: print(f"❌ Follow-up Error Details: {follow_up['error']}") return f"❌ Follow-up Error: {follow_up['error']}", conversation_history final_message = follow_up.get('choices', [{}])[0].get('message', {}) response_text = final_message.get('content', 'No response') print(f"🧠 LLM response: {response_text[:200]}..." if len(str(response_text)) > 200 else f"🧠 LLM response: {response_text}") # Check if LLM returned raw JSON instead of a proper response if response_text.strip().startswith('{') and response_text.strip().endswith('}'): try: json.loads(response_text.strip()) # If it's valid JSON, replace with a proper response response_text = f"I searched for sites but didn't find any results. The search returned: {formatted_result}" except json.JSONDecodeError: pass # Not JSON, keep original response # Add assistant response to history conversation_history.append({"role": "assistant", "content": response_text}) return response_text, conversation_history elif should_force_function: # LLM didn't call a tool but this looks like a NetBox query print("🤖 LLM didn't call NetBox tool, but this looks like a NetBox query. Calling NetBox directly...") # Try to determine which tool to call based on the query if any(word in user_input.lower() for word in ['device', 'devices']): tool_name = "netbox_get_objects" arguments = {"object_type": "devices", "filters": {"limit": 10}} elif any(word in user_input.lower() for word in ['site', 'sites']): tool_name = "netbox_get_objects" # Extract potential company/site name from user input words = user_input.lower().split() site_name = None # Look for patterns like "how many sites does [company] have" if "does" in words and "have" in words: try: does_index = words.index("does") have_index = words.index("have") if does_index + 1 < have_index: site_name = words[does_index + 1] except ValueError: pass filters = {"limit": 10} if site_name: filters["name__icontains"] = site_name arguments = {"object_type": "sites", "filters": filters} elif any(word in user_input.lower() for word in ['ip', 'address']): tool_name = "netbox_get_objects" arguments = {"object_type": "ip-addresses", "filters": {"limit": 10}} elif any(word in user_input.lower() for word in ['vlan']): tool_name = "netbox_get_objects" arguments = {"object_type": "vlans", "filters": {"limit": 10}} elif any(word in user_input.lower() for word in ['change', 'log']): tool_name = "netbox_get_changelogs" arguments = {"filters": {"limit": 10}} else: tool_name = "netbox_get_objects" arguments = {"object_type": "devices", "filters": {"limit": 5}} print(f"🔧 Calling NetBox tool: {tool_name}") tool_result = self.call_netbox_tool(tool_name, arguments) # Format the result for display formatted_result = self.format_netbox_result(tool_name, tool_result) # Create a comprehensive response with the NetBox data if "error" not in tool_result: if isinstance(tool_result, list) and len(tool_result) > 0: response_text = f"Here's the NetBox data I found:\n\n{formatted_result}\n\n" # Add some analysis based on the data if tool_name == "netbox_get_objects": if arguments["object_type"] == "devices": response_text += f"I found {len(tool_result)} devices in your NetBox instance. " if len(tool_result) > 0: sites = set() for device in tool_result: if 'site' in device and device['site']: sites.add(device['site'].get('name', 'Unknown')) if sites: response_text += f"They are located at {len(sites)} different sites: {', '.join(list(sites)[:5])}" if len(sites) > 5: response_text += f" and {len(sites) - 5} more sites." elif arguments["object_type"] == "sites": response_text += f"I found {len(tool_result)} sites in your NetBox instance." elif arguments["object_type"] == "ip-addresses": response_text += f"I found {len(tool_result)} IP addresses in your NetBox instance." elif arguments["object_type"] == "vlans": response_text += f"I found {len(tool_result)} VLANs in your NetBox instance." elif tool_name == "netbox_get_changelogs": response_text += f"I found {len(tool_result)} recent changes in your NetBox instance." else: response_text = f"NetBox returned: {formatted_result}" else: response_text = f"❌ Error retrieving NetBox data: {tool_result['error']}" # Add assistant response to history conversation_history.append({"role": "assistant", "content": response_text}) return response_text, conversation_history else: # Direct response from LLM response_text = message.get('content', 'No response') conversation_history.append({"role": "assistant", "content": response_text}) return response_text, conversation_history def show_help(self): """Show help information.""" print("\n🤖 NetBox + LLM ChatBot Help") print("=" * 40) print("Ask questions about your NetBox data in natural language!") print("The LLM will automatically use NetBox tools to get the information.") print("\nExamples:") print("• 'Show me all devices in my network'") print("• 'What IP addresses are assigned to VLAN 100?'") print("• 'List all sites in NetBox'") print("• 'Tell me about device ID 123'") print("• 'Show me recent changes to devices'") print("• 'What Cisco devices do we have?'") print("• 'How many VLANs are configured?'") print("\nCommands:") print("• 'help' - Show this help") print("• 'tools' - Show available NetBox tools") print("• 'quit' or 'exit' - Exit the chatbot") print("\nPress Ctrl+C to exit anytime.") def show_tools(self): """Show available NetBox tools.""" print("\n🔧 Available NetBox Tools:") for tool in self.tools: func = tool['function'] print(f"\n• {func['name']}") print(f" Description: {func['description']}") print(f" Required: {', '.join(func['parameters']['required'])}") print(f" Properties: {', '.join(func['parameters']['properties'].keys())}") def run(self): """Run the interactive chatbot.""" print("🤖 NetBox + LLM ChatBot") print("=" * 30) print(f"📡 NetBox MCP: {self.mcpo_url}") print(f"🧠 LLM API: {self.llm_url}") print(f"🤖 Model: {self.llm_model}") print("\nConnected! Ask questions about your NetBox data.") print("Type 'help' for commands or ask questions.") print("Press Ctrl+C to exit.\n") conversation_history = [ { "role": "system", "content": "You are a helpful assistant that can query NetBox data using the available tools. You MUST use the NetBox tools (netbox_get_objects, netbox_get_object_by_id, netbox_get_changelogs) to answer questions about network infrastructure, devices, IP addresses, VLANs, and other network data. When users ask about devices, sites, IPs, VLANs, or any network data, you MUST call the appropriate NetBox tool first before responding. Always provide clear, helpful responses based on the actual data you retrieve from NetBox. NEVER return raw JSON or tool arguments - always provide a conversational response explaining what you found or didn't find." } ] while True: try: user_input = input("👤 You: ").strip() if not user_input: continue # Handle special commands if user_input.lower() in ['quit', 'exit', 'q']: print("👋 Goodbye!") break elif user_input.lower() == 'help': self.show_help() continue elif user_input.lower() == 'tools': self.show_tools() continue # Process with LLM print("🤖 Thinking...") response, conversation_history = self.run_conversation(user_input, conversation_history) print(f"🤖 {response}\n") except KeyboardInterrupt: print("\n👋 Goodbye!") break except Exception as e: print(f"❌ Unexpected Error: {e}") import traceback print(f"❌ Full Traceback: {traceback.format_exc()}") print() def signal_handler(sig, frame): """Handle Ctrl+C gracefully.""" print("\n👋 Goodbye!") sys.exit(0) def main(): """Main function.""" # Configuration - EDIT THESE VALUES MCPO_URL = "http://localhost:8000" # Your mcpo server URL MCPO_API_KEY = "netbox-secret-key" # Your mcpo API key load_dotenv() LLM_URL = os.getenv("LLM_URL") # Your internal LLM API URL LLM_API_KEY = os.getenv("LLM_API_KEY") # Your LLM API key LLM_MODEL = os.getenv("LLM_MODEL") # Your LLM model name print("🚀 Starting NetBox + LLM ChatBot...") print(f"📡 NetBox MCP: {MCPO_URL}") print(f"🧠 LLM API: {LLM_URL}") print(f"🤖 Model: {LLM_MODEL}") # Set up signal handler for Ctrl+C signal.signal(signal.SIGINT, signal_handler) # Create and run chatbot chatbot = NetBoxLLMChatBot(MCPO_URL, MCPO_API_KEY, LLM_URL, LLM_API_KEY, LLM_MODEL) chatbot.run() if __name__ == "__main__": main()