From 73df36b11da0373d6e66af3b97e9b31f33b0ef7d Mon Sep 17 00:00:00 2001 From: Frank Clements Date: Mon, 25 Aug 2025 11:58:17 -0400 Subject: [PATCH 1/2] Add unified transport support and environment configuration - Implement unified FastMCP server supporting both stdio and HTTP transports - Add .env file support with python-dotenv for configuration management - Create env.sample with documented configuration options - Update dependencies to include MCP 1.13.1 and python-dotenv - Add comprehensive .gitignore for Python projects - Remove temporary documentation files (will be cleaned up later) Features: - Single server.py supports both transports via --transport flag - Environment variable configuration with CLI argument override - NetBox client with configurable SSL verification - Proper tool registration using FastMCP decorators --- .gitignore | 73 +++++++++++++++++++++++++++++ env.sample | 35 ++++++++++++++ pyproject.toml | 4 +- requirements.txt | 4 +- server.py | 116 ++++++++++++++++++++++++++++++++++++++++++----- 5 files changed, 217 insertions(+), 15 deletions(-) create mode 100644 .gitignore create mode 100644 env.sample diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2c3b04a --- /dev/null +++ b/.gitignore @@ -0,0 +1,73 @@ +# Environment files +.env +.env.local +.env.*.local + +# Python cache files +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +.hypothesis/ +.pytest_cache/ + +# Virtual environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# IDE files +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS files +.DS_Store +.DS_Store? +._* +.Spotlight-V100 +.Trashes +ehthumbs.db +Thumbs.db diff --git a/env.sample b/env.sample new file mode 100644 index 0000000..423168e --- /dev/null +++ b/env.sample @@ -0,0 +1,35 @@ +# NetBox MCP Server Configuration +# Copy this file to .env and update the values for your environment + +# NetBox Configuration +# The base URL of your NetBox instance (required) +NETBOX_URL=https://your-netbox-instance.com + +# NetBox API token for authentication (required) +NETBOX_TOKEN=your_netbox_api_token_here + +# MCP Server Configuration +# Log level for the MCP server (DEBUG, INFO, WARNING, ERROR, CRITICAL) +LOG_LEVEL=INFO + +# Transport method for MCP communication (stdio, http) +MCP_TRANSPORT=stdio + +# Server host for HTTP transport (only used when MCP_TRANSPORT=http) +MCP_SERVER_HOST=localhost + +# Server port for HTTP transport (only used when MCP_TRANSPORT=http) +MCP_SERVER_PORT=8000 + +# SSL Configuration +# Whether to verify SSL certificates when connecting to NetBox +VERIFY_SSL=true + +# Example values: +# NETBOX_URL=https://netbox.example.com +# NETBOX_TOKEN=0123456789abcdef0123456789abcdef01234567 +# LOG_LEVEL=DEBUG +# MCP_TRANSPORT=http +# MCP_SERVER_HOST=0.0.0.0 +# MCP_SERVER_PORT=3000 +# VERIFY_SSL=false diff --git a/pyproject.toml b/pyproject.toml index ee6cd23..8cf0a85 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ description = "A read-only MCP server for NetBox" readme = "README.md" requires-python = ">=3.13" dependencies = [ - "httpx>=0.28.1", - "mcp[cli]>=1.3.0", + "mcp>=1.13.1", "requests>=2.31.0", + "python-dotenv>=1.0.0", ] diff --git a/requirements.txt b/requirements.txt index 1a020ff..ee69b1b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,3 @@ -requests>=2.25.0 \ No newline at end of file +requests>=2.25.0 +python-dotenv>=1.0.0 +mcp>=1.13.1 \ No newline at end of file diff --git a/server.py b/server.py index 98e58ee..4bc5037 100644 --- a/server.py +++ b/server.py @@ -1,6 +1,10 @@ from mcp.server.fastmcp import FastMCP from netbox_client import NetBoxRestClient import os +import argparse +import sys +import asyncio +from dotenv import load_dotenv # Mapping of simple object names to API endpoints NETBOX_OBJECT_TYPES = { @@ -97,10 +101,11 @@ "webhooks": "extras/webhooks", } -mcp = FastMCP("NetBox", log_level="DEBUG") +# Global variables +app = FastMCP("NetBox") netbox = None -@mcp.tool() +@app.tool() def netbox_get_objects(object_type: str, filters: dict): """ Get objects from NetBox based on their type and filters @@ -204,7 +209,7 @@ def netbox_get_objects(object_type: str, filters: dict): # Make API call return netbox.get(endpoint, params=filters) -@mcp.tool() +@app.tool() def netbox_get_object_by_id(object_type: str, object_id: int): """ Get detailed information about a specific NetBox object by its ID. @@ -226,7 +231,7 @@ def netbox_get_object_by_id(object_type: str, object_id: int): return netbox.get(endpoint) -@mcp.tool() +@app.tool() def netbox_get_changelogs(filters: dict): """ Get object change records (changelogs) from NetBox based on filters. @@ -275,15 +280,102 @@ def netbox_get_changelogs(filters: dict): # Make API call return netbox.get(endpoint, params=filters) -if __name__ == "__main__": - # Load NetBox configuration from environment variables - netbox_url = os.getenv("NETBOX_URL") - netbox_token = os.getenv("NETBOX_TOKEN") + +def load_config(): + """Load configuration from .env file and CLI arguments.""" + # Load environment variables from .env file + load_dotenv() + + # Set up argument parser + parser = argparse.ArgumentParser(description="NetBox MCP Server") + parser.add_argument( + "--netbox-url", + default=os.getenv("NETBOX_URL"), + help="NetBox instance URL (default: from NETBOX_URL env var)" + ) + parser.add_argument( + "--netbox-token", + default=os.getenv("NETBOX_TOKEN"), + help="NetBox API token (default: from NETBOX_TOKEN env var)" + ) + parser.add_argument( + "--log-level", + choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], + default=os.getenv("LOG_LEVEL", "INFO"), + help="Log level (default: INFO)" + ) + parser.add_argument( + "--transport", + choices=["stdio", "http"], + default=os.getenv("MCP_TRANSPORT", "stdio"), + help="MCP transport method (default: stdio)" + ) + parser.add_argument( + "--host", + default=os.getenv("MCP_SERVER_HOST", "localhost"), + help="Server host for HTTP transport (default: localhost)" + ) + parser.add_argument( + "--port", + type=int, + default=int(os.getenv("MCP_SERVER_PORT", "8000")), + help="Server port for HTTP transport (default: 8000)" + ) + parser.add_argument( + "--verify-ssl", + action="store_true", + default=os.getenv("VERIFY_SSL", "true").lower() == "true", + help="Verify SSL certificates (default: true)" + ) + parser.add_argument( + "--no-verify-ssl", + action="store_false", + dest="verify_ssl", + help="Disable SSL certificate verification" + ) - if not netbox_url or not netbox_token: - raise ValueError("NETBOX_URL and NETBOX_TOKEN environment variables must be set") + return parser.parse_args() + + +def initialize_netbox_client(config): + """Initialize the NetBox client with the given configuration.""" + # Validate required configuration + if not config.netbox_url: + raise ValueError("NetBox URL must be provided via --netbox-url or NETBOX_URL environment variable") + if not config.netbox_token: + raise ValueError("NetBox token must be provided via --netbox-token or NETBOX_TOKEN environment variable") # Initialize NetBox client - netbox = NetBoxRestClient(url=netbox_url, token=netbox_token) + global netbox + netbox = NetBoxRestClient( + url=config.netbox_url, + token=config.netbox_token, + verify_ssl=config.verify_ssl + ) + + +def main(): + """Main entry point for the server.""" + # Load configuration from .env and CLI arguments + config = load_config() + + # Initialize NetBox client + initialize_netbox_client(config) - mcp.run(transport="stdio") + # Run the server with the specified transport + if config.transport == "stdio": + # Run with stdio transport (default FastMCP behavior) + app.run() + elif config.transport == "http": + # Run with HTTP transport + # FastMCP supports this via the run method with transport parameter + app.run(transport="streamable-http") + else: + raise ValueError(f"Unsupported transport: {config.transport}") + +if __name__ == "__main__": + try: + main() + except Exception as e: + print(f"Error starting NetBox MCP Server: {e}", file=sys.stderr) + sys.exit(1) From 3896c2c220205ef6734665b27503d3e732f07387 Mon Sep 17 00:00:00 2001 From: Frank Clements Date: Mon, 25 Aug 2025 15:21:55 -0400 Subject: [PATCH 2/2] feat: Add streamable-http transport, .env config, CLI args, and enhanced logging - Add support for streamable-http transport alongside stdio - Implement flexible configuration system (.env file + CLI arguments) - Add comprehensive logging with debug-level visibility - Enhance NetBox client with detailed request/response logging - Add Host header to prevent 'No Host Supplied' errors - Support JSON string filter parsing in MCP tools - Update README with complete configuration and usage guide - Add troubleshooting section for common issues Breaking changes: - Server now requires explicit configuration via .env or CLI args - Enhanced error handling may change error message format Closes: Configuration and transport enhancement requests --- README.md | 195 ++++++++++++++++++++++++++++++++++++++++++----- netbox_client.py | 66 ++++++++++++++-- server.py | 169 +++++++++++++++++++++++++++++++++++++--- 3 files changed, 393 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index fd263b9..a277441 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,8 @@ # NetBox MCP Server -This is a simple read-only [Model Context Protocol](https://modelcontextprotocol.io/) server for NetBox. It enables you to interact with your data in NetBox directly via LLMs that support MCP. +This is a simple read-only [Model Context Protocol](https://modelcontextprotocol.io/) server for NetBox. It enables you to interact with your data in NetBox directly via LLMs that support MCP. + +The server supports both **stdio** (for local/CLI usage) and **streamable-http** (for web services) transport protocols, with flexible configuration via environment variables or CLI arguments. ## Tools @@ -12,38 +14,145 @@ This is a simple read-only [Model Context Protocol](https://modelcontextprotocol > Note: the set of supported object types is explicitly defined and limited to the core NetBox objects for now, and won't work with object types from plugins. +## Configuration + +The server can be configured using either a `.env` file or CLI arguments. CLI arguments take precedence over `.env` file values. + +### Environment Variables (.env file) + +Create a `.env` file in the project directory (see `env.sample` for reference): + +```bash +# NetBox connection settings +NETBOX_URL=https://netbox.example.com/ +NETBOX_TOKEN=your-api-token-here + +# MCP server settings +MCP_TRANSPORT=stdio # stdio or streamable-http +MCP_SERVER_HOST=localhost # Only used for streamable-http +MCP_SERVER_PORT=8000 # Only used for streamable-http + +# Optional settings +LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR +VERIFY_SSL=true # SSL certificate verification +``` + +### CLI Arguments + +All configuration options can be overridden via command line: + +```bash +python3 server.py --help + +# Examples: +python3 server.py --netbox-url https://netbox.example.com/ --netbox-token your-token +python3 server.py --transport streamable-http --host 0.0.0.0 --port 8080 +python3 server.py --log-level DEBUG --no-verify-ssl +``` + ## Usage +### Prerequisites + 1. Create a read-only API token in NetBox with sufficient permissions for the tool to access the data you want to make available via MCP. -2. Install dependencies: `uv add -r requirements.txt` +2. Install dependencies: `pip install -r requirements.txt` or `uv add -r requirements.txt` + +### Running the Server -3. Verify the server can run: `NETBOX_URL=https://netbox.example.com/ NETBOX_TOKEN= uv run server.py` +#### Option 1: Using .env file (Recommended) -3. Add the MCP server configuration to your LLM client. For example, in Claude Desktop (Mac): +1. Copy `env.sample` to `.env` and configure your settings +2. Run the server: `python3 server.py` + +#### Option 2: Using CLI arguments + +```bash +python3 server.py --netbox-url https://netbox.example.com/ --netbox-token your-token +``` + +#### Option 3: Using environment variables + +```bash +NETBOX_URL=https://netbox.example.com/ NETBOX_TOKEN=your-token python3 server.py +``` + +### Transport Protocols + +#### stdio Transport (Default) +For local usage with LLM clients like Claude Desktop: + +```bash +python3 server.py --transport stdio +``` + +#### streamable-http Transport +For web services and remote access: + +```bash +python3 server.py --transport streamable-http --host 0.0.0.0 --port 8000 +``` + +### LLM Client Configuration + +#### For stdio transport (Claude Desktop example): ```json { "mcpServers": { - "netbox": { - "command": "uv", - "args": [ - "--directory", - "/path/to/netbox-mcp-server", - "run", - "server.py" - ], - "env": { - "NETBOX_URL": "https://netbox.example.com/", - "NETBOX_TOKEN": "" - } - } + "netbox": { + "command": "python3", + "args": ["/path/to/netbox-mcp-server/server.py"], + "env": { + "NETBOX_URL": "https://netbox.example.com/", + "NETBOX_TOKEN": "your-api-token-here" + } + } + } } ``` -> On Windows, use full, escaped path to your instance, such as `C:\\Users\\myuser\\.local\\bin\\uv` and `C:\\Users\\myuser\\netbox-mcp-server`. + +Or using uv: +```json +{ + "mcpServers": { + "netbox": { + "command": "uv", + "args": [ + "--directory", + "/path/to/netbox-mcp-server", + "run", + "server.py" + ], + "env": { + "NETBOX_URL": "https://netbox.example.com/", + "NETBOX_TOKEN": "your-api-token-here" + } + } + } +} +``` + +#### For streamable-http transport + +When using streamable-http transport, the server runs as a web service. Configure your MCP client to connect to the HTTP endpoint: + +```json +{ + "mcpServers": { + "netbox": { + "url": "http://localhost:8000" + } + } +} +``` + +> **Note**: On Windows, use full, escaped paths such as `C:\\Users\\myuser\\.local\\bin\\uv` and `C:\\Users\\myuser\\netbox-mcp-server`. > For detailed troubleshooting, consult the [MCP quickstart](https://modelcontextprotocol.io/quickstart/user). -4. Use the tools in your LLM client. For example: +### Example Usage + +Once configured, you can use the tools in your LLM client: ```text > Get all devices in the 'Equinix DC14' site @@ -57,9 +166,55 @@ This is a simple read-only [Model Context Protocol](https://modelcontextprotocol > Show me all configuration changes to the core router in the last month ``` +## Debugging and Logging + +The server includes comprehensive logging to help troubleshoot issues: + +### Enable Debug Logging + +```bash +# Via .env file +echo "LOG_LEVEL=DEBUG" >> .env + +# Via CLI argument +python3 server.py --log-level DEBUG + +# Via environment variable +LOG_LEVEL=DEBUG python3 server.py +``` + +### Debug Output + +With debug logging enabled, you'll see detailed information about: +- Configuration loading and validation +- NetBox API requests and responses +- MCP tool calls and parameters +- HTTP request/response details +- Error stack traces + +### Common Issues + +**"No Host Supplied" Error**: This typically indicates a networking or SSL issue. Try: +```bash +python3 server.py --no-verify-ssl --log-level DEBUG +``` + +**Connection Refused**: Verify your NetBox URL and network connectivity: +```bash +curl -H "Authorization: Token your-token" https://your-netbox-url/api/ +``` + ## Development -Contributions are welcome! Please open an issue or submit a PR. +Contributions are welcome! Please open an issue or submit a PR. + +### Development Setup + +1. Clone the repository +2. Install dependencies: `pip install -r requirements.txt` +3. Copy `env.sample` to `.env` and configure +4. Run tests: `python3 -m pytest` (when available) +5. Run the server: `python3 server.py --log-level DEBUG` ## License diff --git a/netbox_client.py b/netbox_client.py index 054d870..461406b 100644 --- a/netbox_client.py +++ b/netbox_client.py @@ -8,6 +8,7 @@ import abc from typing import Any, Dict, List, Optional, Union import requests +import logging class NetBoxClientBase(abc.ABC): @@ -156,16 +157,33 @@ def __init__(self, url: str, token: str, verify_ssl: bool = True): token: API token for authentication verify_ssl: Whether to verify SSL certificates """ + self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") + self.base_url = url.rstrip('/') self.api_url = f"{self.base_url}/api" self.token = token self.verify_ssl = verify_ssl + + self.logger.debug(f"Initializing NetBox REST client for {self.base_url}") + self.logger.debug(f"SSL verification: {verify_ssl}") + self.session = requests.Session() + + # Extract hostname from URL for Host header + from urllib.parse import urlparse + parsed_url = urlparse(self.base_url) + host_header = parsed_url.netloc + self.session.headers.update({ 'Authorization': f'Token {token}', 'Content-Type': 'application/json', 'Accept': 'application/json', + 'Host': host_header, + 'User-Agent': 'NetBox-MCP-Server/1.0', }) + + self.logger.debug(f"Session headers configured: {dict(self.session.headers)}") + self.logger.debug("NetBox REST client initialized successfully") def _build_url(self, endpoint: str, id: Optional[int] = None) -> str: """Build the full URL for an API request.""" @@ -190,14 +208,48 @@ def get(self, endpoint: str, id: Optional[int] = None, params: Optional[Dict[str requests.HTTPError: If the request fails """ url = self._build_url(endpoint, id) - response = self.session.get(url, params=params, verify=self.verify_ssl) - response.raise_for_status() + self.logger.debug(f"Built URL: {url}") + self.logger.debug(f"Request params: {params}") + self.logger.debug(f"Request headers: {dict(self.session.headers)}") - data = response.json() - if id is None and 'results' in data: - # Handle paginated results - return data['results'] - return data + try: + response = self.session.get(url, params=params, verify=self.verify_ssl) + + # Log detailed request information + self.logger.debug(f"Final request URL: {response.url}") + self.logger.debug(f"Request method: {response.request.method}") + self.logger.debug(f"Request headers sent: {dict(response.request.headers)}") + self.logger.debug(f"Response status: {response.status_code}") + self.logger.debug(f"Response headers: {dict(response.headers)}") + + # Log response content for debugging + if response.status_code >= 400: + self.logger.error(f"Error response body: {response.text}") + else: + # Only log first 500 chars of successful response to avoid spam + response_preview = response.text[:500] + "..." if len(response.text) > 500 else response.text + self.logger.debug(f"Response body preview: {response_preview}") + + response.raise_for_status() + + data = response.json() + if id is None and 'results' in data: + # Handle paginated results + result_count = len(data['results']) + total_count = data.get('count', result_count) + self.logger.debug(f"Retrieved {result_count} objects from paginated response (total: {total_count})") + return data['results'] + else: + self.logger.debug("Retrieved single object") + return data + + except requests.exceptions.RequestException as e: + self.logger.error(f"HTTP request failed for {url}: {str(e)}") + self.logger.error(f"Request details - Method: GET, URL: {url}, Params: {params}") + if hasattr(e, 'response') and e.response is not None: + self.logger.error(f"Response status: {e.response.status_code}") + self.logger.error(f"Response body: {e.response.text}") + raise def create(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]: """ diff --git a/server.py b/server.py index 4bc5037..e74de75 100644 --- a/server.py +++ b/server.py @@ -4,6 +4,7 @@ import argparse import sys import asyncio +import logging from dotenv import load_dotenv # Mapping of simple object names to API endpoints @@ -104,6 +105,36 @@ # Global variables app = FastMCP("NetBox") netbox = None +logger = logging.getLogger(__name__) + +def setup_logging(log_level: str): + """Setup logging configuration.""" + # Convert string level to logging constant + numeric_level = getattr(logging, log_level.upper(), None) + if not isinstance(numeric_level, int): + raise ValueError(f'Invalid log level: {log_level}') + + # Configure root logger + logging.basicConfig( + level=numeric_level, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' + ) + + # Set specific logger levels + logger.setLevel(numeric_level) + + # Configure third-party loggers + if numeric_level == logging.DEBUG: + # Enable debug logging for requests when in debug mode + logging.getLogger("requests").setLevel(logging.DEBUG) + logging.getLogger("urllib3").setLevel(logging.DEBUG) + else: + # Suppress noisy third-party loggers + logging.getLogger("requests").setLevel(logging.WARNING) + logging.getLogger("urllib3").setLevel(logging.WARNING) + + logger.info(f"Logging configured at {log_level.upper()} level") @app.tool() def netbox_get_objects(object_type: str, filters: dict): @@ -198,16 +229,49 @@ def netbox_get_objects(object_type: str, filters: dict): See NetBox API documentation for filtering options for each object type. """ + logger.info(f"netbox_get_objects called with object_type='{object_type}', filters={filters}") + logger.debug(f"Filters type: {type(filters)}") + + # Handle filters parameter - it might be a JSON string that needs parsing + processed_filters = filters + if isinstance(filters, str): + try: + import json + processed_filters = json.loads(filters) + logger.debug(f"Parsed JSON filters string into dict: {processed_filters}") + except json.JSONDecodeError as e: + logger.warning(f"Failed to parse filters as JSON: {e}. Using as-is: {filters}") + processed_filters = filters + + logger.debug(f"Final processed filters: {processed_filters} (type: {type(processed_filters)})") + # Validate object_type exists in mapping if object_type not in NETBOX_OBJECT_TYPES: + logger.error(f"Invalid object_type '{object_type}' requested") valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys())) raise ValueError(f"Invalid object_type. Must be one of:\n{valid_types}") # Get API endpoint from mapping endpoint = NETBOX_OBJECT_TYPES[object_type] + logger.debug(f"Mapped object_type '{object_type}' to endpoint '{endpoint}'") # Make API call - return netbox.get(endpoint, params=filters) + try: + logger.debug(f"Making NetBox API call to endpoint '{endpoint}' with params: {processed_filters}") + result = netbox.get(endpoint, params=processed_filters) + + if isinstance(result, list): + logger.info(f"Successfully retrieved {len(result)} {object_type} objects from NetBox") + else: + logger.info(f"Successfully retrieved {object_type} object from NetBox") + + logger.debug(f"NetBox API response: {result}") + return result + + except Exception as e: + logger.error(f"Failed to retrieve {object_type} from NetBox: {str(e)}") + logger.debug("Full exception details:", exc_info=True) + raise @app.tool() def netbox_get_object_by_id(object_type: str, object_id: int): @@ -221,15 +285,28 @@ def netbox_get_object_by_id(object_type: str, object_id: int): Returns: Complete object details """ + logger.info(f"netbox_get_object_by_id called with object_type='{object_type}', object_id={object_id}") + # Validate object_type exists in mapping if object_type not in NETBOX_OBJECT_TYPES: + logger.error(f"Invalid object_type '{object_type}' requested for ID lookup") valid_types = "\n".join(f"- {t}" for t in sorted(NETBOX_OBJECT_TYPES.keys())) 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}" + logger.debug(f"Constructed endpoint '{endpoint}' for {object_type} ID {object_id}") - return netbox.get(endpoint) + try: + logger.debug(f"Making NetBox API call to retrieve {object_type} with ID {object_id}") + result = netbox.get(endpoint) + logger.info(f"Successfully retrieved {object_type} object with ID {object_id}") + logger.debug(f"NetBox API response: {result}") + return result + + except Exception as e: + logger.error(f"Failed to retrieve {object_type} with ID {object_id} from NetBox: {str(e)}") + raise @app.tool() def netbox_get_changelogs(filters: dict): @@ -275,10 +352,26 @@ def netbox_get_changelogs(filters: dict): - postchange_data: The object's data after the change (null for deletions) - time: The timestamp when the change was made """ + logger.info(f"netbox_get_changelogs called with filters={filters}") + endpoint = "core/object-changes" + logger.debug(f"Using changelog endpoint '{endpoint}'") - # Make API call - return netbox.get(endpoint, params=filters) + try: + logger.debug(f"Making NetBox API call to retrieve changelogs with filters: {filters}") + result = netbox.get(endpoint, params=filters) + + if isinstance(result, list): + logger.info(f"Successfully retrieved {len(result)} changelog entries from NetBox") + else: + logger.info("Successfully retrieved changelog data from NetBox") + + logger.debug(f"NetBox API response: {result}") + return result + + except Exception as e: + logger.error(f"Failed to retrieve changelogs from NetBox: {str(e)}") + raise def load_config(): @@ -339,43 +432,99 @@ def load_config(): def initialize_netbox_client(config): """Initialize the NetBox client with the given configuration.""" + logger.info("Initializing NetBox client...") + # Validate required configuration if not config.netbox_url: + logger.error("NetBox URL not provided") raise ValueError("NetBox URL must be provided via --netbox-url or NETBOX_URL environment variable") if not config.netbox_token: + logger.error("NetBox token not provided") raise ValueError("NetBox token must be provided via --netbox-token or NETBOX_TOKEN environment variable") + logger.debug(f"NetBox URL: {config.netbox_url}") + logger.debug(f"SSL verification: {config.verify_ssl}") + logger.debug(f"Token provided: {'Yes' if config.netbox_token else 'No'}") + # Initialize NetBox client global netbox - netbox = NetBoxRestClient( - url=config.netbox_url, - token=config.netbox_token, - verify_ssl=config.verify_ssl - ) + try: + netbox = NetBoxRestClient( + url=config.netbox_url, + token=config.netbox_token, + verify_ssl=config.verify_ssl + ) + logger.info(f"NetBox client initialized successfully for {config.netbox_url}") + except Exception as e: + logger.error(f"Failed to initialize NetBox client: {str(e)}") + raise def main(): """Main entry point for the server.""" + print("Starting NetBox MCP Server...") + # Load configuration from .env and CLI arguments config = load_config() + # Setup logging first + setup_logging(config.log_level) + + logger.info("="*60) + logger.info("NetBox MCP Server Starting") + logger.info("="*60) + + # Log configuration details + logger.info(f"Configuration loaded:") + logger.info(f" Transport: {config.transport}") + logger.info(f" Log Level: {config.log_level}") + logger.info(f" NetBox URL: {config.netbox_url}") + logger.info(f" SSL Verification: {config.verify_ssl}") + + if config.transport == "http": + logger.info(f" HTTP Host: {config.host}") + logger.info(f" HTTP Port: {config.port}") + # Initialize NetBox client initialize_netbox_client(config) + # Log available tools + logger.info("Available MCP tools:") + logger.info(" - netbox_get_objects: Get objects from NetBox by type and filters") + logger.info(" - netbox_get_object_by_id: Get specific NetBox object by ID") + logger.info(" - netbox_get_changelogs: Get NetBox change records") + # Run the server with the specified transport + logger.info(f"Starting MCP server with {config.transport} transport...") + if config.transport == "stdio": + logger.info("Server ready for stdio communication") # Run with stdio transport (default FastMCP behavior) app.run() elif config.transport == "http": + logger.info(f"Server will be available at http://{config.host}:{config.port}") # Run with HTTP transport # FastMCP supports this via the run method with transport parameter app.run(transport="streamable-http") else: + logger.error(f"Unsupported transport: {config.transport}") raise ValueError(f"Unsupported transport: {config.transport}") if __name__ == "__main__": try: main() + except KeyboardInterrupt: + print("\nServer interrupted by user", file=sys.stderr) + if 'logger' in globals(): + logger.info("Server shutdown requested by user") + sys.exit(0) except Exception as e: - print(f"Error starting NetBox MCP Server: {e}", file=sys.stderr) + error_msg = f"Error starting NetBox MCP Server: {e}" + print(error_msg, file=sys.stderr) + + # Log the error if logger is available + if 'logger' in globals(): + logger.error(error_msg) + logger.debug("Full error details:", exc_info=True) + sys.exit(1)