splunk-mcp

Instructions

Implementation Reference

• splunk_mcp.py:396-424 (handler)

  The handler function that implements the get_index_info tool logic. It retrieves metadata for a specified Splunk index including event count, sizes, and time ranges. The @mcp.tool() decorator registers it as an MCP tool.
  @mcp.tool() async def get_index_info(index_name: str) -> Dict[str, Any]: """ Get metadata for a specific Splunk index. Args: index_name: Name of the index to get metadata for Returns: Dictionary containing index metadata """ try: service = get_splunk_connection() index = service.indexes[index_name] return { "name": index_name, "total_event_count": str(index["totalEventCount"]), "current_size": str(index["currentDBSizeMB"]), "max_size": str(index["maxTotalDataSizeMB"]), "min_time": str(index["minTime"]), "max_time": str(index["maxTime"]) } except KeyError: logger.error(f"❌ Index not found: {index_name}") raise ValueError(f"Index not found: {index_name}") except Exception as e: logger.error(f"❌ Failed to get index info: {str(e)}") raise
• splunk_mcp.py:396-396 (registration)

  The @mcp.tool() decorator registers the get_index_info function as an MCP tool.
  @mcp.tool()
• splunk_mcp.py:298-332 (helper)

  Helper function used by get_index_info to establish Splunk connection.
  def get_splunk_connection() -> splunklib.client.Service: """ Get a connection to the Splunk service. Supports both username/password and token-based authentication. If SPLUNK_TOKEN is set, it will be used for authentication and username/password will be ignored. Returns: splunklib.client.Service: Connected Splunk service """ try: if SPLUNK_TOKEN: logger.debug(f"🔌 Connecting to Splunk at {SPLUNK_SCHEME}://{SPLUNK_HOST}:{SPLUNK_PORT} using token authentication") service = splunklib.client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, scheme=SPLUNK_SCHEME, verify=VERIFY_SSL, token=f"Bearer {SPLUNK_TOKEN}" ) else: username = os.environ.get("SPLUNK_USERNAME", "admin") logger.debug(f"🔌 Connecting to Splunk at {SPLUNK_SCHEME}://{SPLUNK_HOST}:{SPLUNK_PORT} as {username}") service = splunklib.client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, username=username, password=SPLUNK_PASSWORD, scheme=SPLUNK_SCHEME, verify=VERIFY_SSL ) logger.debug(f"✅ Connected to Splunk successfully") return service except Exception as e: logger.error(f"❌ Failed to connect to Splunk: {str(e)}") raise